Skip to main content
Configure audit logging, data masking, header injection, the message integrity Interceptor and encryption Interceptors to secure and track data in your Kafka topics.
From our blog: FedRAMP High for Kafka without replatforming Meet FedRAMP High encryption requirements on your existing Kafka, no replatforming.Kafka and HIPAA 2026 What the 2026 HIPAA Security Rule changes for Kafka, and how to close the gaps.

Common configuration

The following configurations are shared across multiple Interceptors:

Environment variables as secrets

To ensure your secrets don’t appear in your Interceptors, you can refer to the environment variables set in your Gateway container. Use the format ${MY_ENV_VAR}. We recommend using this for schema registry or Vault secrets and any other values you’d like to hide in the configuration.

Audit Interceptor

This Interceptor logs information from API key requests. To use it, inject it and implement ApiKeyAuditLog interface for audit. The currently supported Kafka API requests are:
  • ProduceRequest (PRODUCE)
  • FetchRequest (FETCH)
  • CreateTopicRequest (CREATE_TOPICS)
  • DeleteTopicRequest (DELETE_TOPICS)
  • AlterConfigRequest (ALTER_CONFIGS)
When incremental fetch sessions are enabled, Gateway records a fetch only when the request lists the topics being read, which a consumer stops doing once it settles into steady-state reads. A long-lived consumer can appear idle in audit logs while it keeps consuming. This affects audit visibility only — it doesn’t change the data a consumer receives.

Configure audit Interceptor

Audit Interceptor example

Data masking Interceptor

Field level data masking Interceptor masks sensitive fields within messages as they are consumed.

Configure data masking Interceptor

The policies will be applied when consuming messages.

Data masking policy

Data masking rule

Masking type

  • MASK_ALL: all data will be masked
  • MASK_FIRST_N: the first n characters will be masked
  • MASK_LAST_N: the last n characters will be masked

Error policy

You can control the plugin behavior when it can’t parse a fetched message through its errorPolicy which can be set to fail_fetch or skip_masking.
The error policy only applies to messages that do not have an associated schema. When a message has a schema (Avro, JSON Schema or Protobuf), the plugin uses the schema to parse the message and the error policy is not triggered.
The default is fail_fetch. In this mode, the plugin will return a failure to read the batch which the fetch record is part of, effectively blocking any consumer. In skip_masking mode, if there’s a failure to parse a message being fetched (e.g. an encrypted record or a schemaless message that can’t be parsed), then that record is skipped and returned un-masked.

Full payload encryption compatibility

Data masking is compatible with full payload encryption. When both Interceptors are applied to the same topic, data masking automatically detects records with full payload encryption headers and skips them, preventing deserialization errors that would otherwise occur when attempting to mask encrypted content. Check out the encryption configuration for details. Field level encryption is not affected by this behavior.

Schema registry

Gateway supports Confluent-like and AWS Glue schema registries. If you don’t supply a basicCredentials section for the AWS Glue schema registry, the client will attempt to find the connection information it needs from the environment (see AWS docs for details ) and the credentials required can be passed this way to Gateway as part of its core configuration. Read our blog about schema registry.

Data masking Interceptor example

Secured schema registry

Dynamic header injection Interceptor

This Interceptor injects headers (such as user IP) to the messages as they are produced through Gateway. We support templating in this format: X-CLIENT_IP: "{{userIp}} testing".

Context variables

These values are available as template variables:
  • uuid
  • userIp
  • vcluster
  • user
  • clientId
  • gatewayIp
  • gatewayHost
  • gatewayVersion
  • apiKey
  • apiKeyVersion
  • timestampMillis

Record extraction templates

You can also extract fields from the record key or value:
  • {{record.key}} - extract the entire key payload as a string
  • {{record.value}} - extract the entire value payload as a string
  • {{record.key.fieldName}} - extract a specific field from the record key
  • {{record.value.fieldName}} - extract a specific field from the record value
For example, if your record has a key with a field named “id”, you can use {{record.key.id}} to extract that value and inject it as a header.
To use field extraction (record.key.fieldName or record.value.fieldName) with Avro, JSON Schema or Protobuf data, you have to configure schemaRegistryConfig so Gateway can deserialize the records. For plain JSON data, no schema registry is needed.

Configure header injection Interceptor

Error handling

The failOnError setting controls how the Interceptor handles errors during header injection:
  • When false (default): errors are logged as warnings and the request continues processing. Headers that fail to be injected are skipped.
  • When true: any error during header injection (such as deserialization failures or missing fields) causes the request to fail with a clear error message.

Header injection Interceptor example

Let’s produce a simple record to the injectHeaderTopic topic.
Let’s consume from our injectHeaderTopic.
You should see the message with headers as below

Message integrity Interceptor

The message integrity Interceptor signs Kafka records on produce and verifies them on fetch, letting consumers detect whether a record changed after it was produced. Two plugins work together:
  • ProduceIntegrityPolicyPlugin signs records using HMAC-SHA256 through Google Tink.
  • FetchIntegrityPolicyPlugin verifies signatures and drops or allows records based on your policy.
You store signing keys in your HashiCorp Vault Key-Value (KV) v2 instance. Gateway reads and caches them locally.

Ordering with other Interceptors

The message integrity Interceptor is always the outermost layer — signing runs last on produce (after all other Interceptors have transformed the record) and verification runs first on fetch (before any normal Interceptor runs). This ensures the signature covers the final produced payload and is verified before any transformation on consume. Gateway enforces this ordering automatically. You don’t have to set specific priority values for integrity Interceptors — Gateway places them in fixed pipeline positions regardless of their configured priority:
  • Produce: the sign plugin always runs after all other Interceptors.
  • Fetch: the verify plugin always runs before all other Interceptors.
Gateway also validates that two integrity Interceptors of the same type don’t have overlapping scopes (Virtual Cluster, group or username). If they do, Gateway rejects the configuration. The Interceptor reads Vault credentials only from its config. You have to specify credential fields (such as token, roleId or secretId) in the config. You can set their values with placeholders like token: "${VAULT_TOKEN}", which Gateway resolves when it loads the config. Gateway does not fall back to environment variable names when you omit a credential field (for example, there is no built-in fallback to VAULT_TOKEN if you leave out token).
When Vault is unreachable (for example, on a cache miss), Gateway propagates the error to the Kafka client: producers receive the failure on produce and consumers receive it on fetch. Gateway does not silently drop records in these cases.

Configure the secretKeyUri

secretKeyUri points to a specific field in a KV v2 secret. Use the format <mount>/data/<path>#<fieldName>. Examples: secret/data/signing-key#key, secret/data/app/keys/signing#hmacKey.

Manage key versions in Vault KV v2

Vault KV v2 versions secrets: each write to the same path creates a new version. Gateway handles versions as follows:
  • Produce (sign): Gateway uses the latest version of the secret at the path you configured. When you write a new value to that path in Vault, Gateway picks it up once the cache entry expires (see cache.ttlMs). Each signed record stores the key version in its signature header.
  • Fetch (verify): The signature on each record identifies the key version used to sign it. Gateway fetches that exact version from Vault to verify, so records signed with an older version still verify correctly after you rotate to a newer version.
  • Older versions: Keep older secret versions readable in Vault until you no longer need to verify records signed with them (for example, until they are consumed or past your retention period).

Configure produce (sign) plugin

If a record already has the signature header, the Interceptor throws PolicyViolationException and does not re-sign.

Configure fetch (verify) plugin

Gateway drops records that fail verification (missing signature, malformed header or invalid MAC (Message Authentication Code)) and emits an audit event. After successful verification, Gateway removes the signature header from the record before returning it to the consumer. For other errors (such as Vault being unreachable on a cache miss), Gateway propagates the error to the Kafka client and does not silently drop the record. When a record is dropped or allowed with missing signature, the fetch (verify) plugin emits a fetch response audit event (error level). To receive these events, enable the audit feature with GATEWAY_FEATURE_FLAGS_AUDIT (see Audit logs and Environment variables). Audit event details:
  • Event type: fetch response audit event (level: error)
  • Information included: topic, partition, offset, Interceptor name, plugin name (FetchIntegrityPolicyPlugin) and a message describing the reason
  • Reason values: missing_signature (no signature header), malformed_signature (header could not be decoded), verification_failed:INVALID_SIGNATURE (MAC does not match) or verification_failed:UNKNOWN_KEY (key version not found in Vault)

Authenticate with Vault for message integrity

All auth types use the common fields: uri (required) and optionally namespace, openTimeoutSeconds (default five), readTimeoutSeconds (default 30), keyStore, trustStore and connectionBackoff. Set type to one of the following and add the corresponding fields. You can also configure TLS for the Vault connection:
  • keyStore: set keyStorePath and keyStorePassword for client certificate authentication
  • trustStore: set trustStorePath and trustStorePassword to verify the Vault server certificate
All auth types except TOKEN support automatic token renewal. Gateway renews Vault tokens in the background so your Interceptor continues to work without interruption. Optional connectionBackoff (for transient Vault failures): backoffDelay (default five), backoffMaxDelay (default 30), backoffChronoUnit (default SECONDS), backoffDelayFactor (default 1.1).

Set up Vault for message integrity

  1. Enable KV v2: vault secrets enable -version=2 kv
  2. Create a signing key for HMAC-SHA256:
    • The key material has to be at least 32 bytes (256 bits) after decoding. Gateway enforces this per NIST SP 800-107 Rev 1 and rejects shorter keys with an error.
    • Store the key in a KV v2 secret as a base64-encoded string. Gateway decodes the Base64 value and uses the resulting bytes, so the decoded length has to be at least 32 bytes.
    • The field name in the secret has to match the <fieldName> in your secretKeyUri (for example, secret/data/signing-key#key uses field name key).
    • Example: generate 32 random bytes, base64-encode them for storage, then write to Vault: KEY=$(openssl rand -base64 32) then vault kv put -mount=secret signing-key key="$KEY"
  3. Create a policy for Gateway with read access on the secret path (for example, path "secret/data/signing-key" { capabilities = ["read"] })

Understand the signature format

Gateway stores each signature in a Kafka header as JSON with two fields: k = the secretKeyUri with its version (for example, secret/data/signing-key#key@1) and s = the base64-encoded HMAC-SHA256 MAC.

Message integrity Interceptor examples

Apply with: conduktor apply -f integrity-sign-interceptor.yaml and conduktor apply -f integrity-verify-interceptor.yaml.

Encryption Interceptors

Gateway encrypts your Kafka data as it passes through the proxy, before it reaches the broker. Unlike TLS (Transport-Level Encryption), Gateway encryption ensures data remains encrypted when stored on Kafka brokers. The section covers all of the configuration options available for every encryption Interceptor. You can also check out other resources:

Encryption configuration

The properties detailed in this section work for the following plugins: Both schema-based and list-based encryption plugins have their configuration, but some properties are common to both of them.

List-based

Decide what you want to encrypt:
  • Record value and record key:
    • Encrypt a set of fields
    • Encrypt the full payload
  • or header keys:
    • Encrypt a set of fields
    • Encrypt the full payload
    • Encrypt a set of headers that match a regex
Record values and record keys Set the following properties for recordValue (value encryption) and/or recordKey (key encryption): Check out the encryption examples. Header keys Set the following properties for recordHeader: Check out the encryption example.

Schema-based

In order to encrypt your data, you can set a few constraints in your schema. These constraints are detailed below, assuming you’re using the default namespace value which is conduktor.​. If you have changed the namespace value in the Interceptor configuration, please change the key name in your schema accordingly. If your field meets one of these three conditions, then it will be encrypted:
  1. This field has a keySecretId set in the schema
  2. This field has a algorithm set in the schema
  3. This field has a set of tags set in the schema, and one of them is part of the tags list specified in the Interceptors.
Check out the encryption example.

Secret keys

Mustache template In all the encryption plugins, you can use mustache templates for the keySecretId. That way, your secret keys will be dynamic.
The value of a field will be replaced with the encrypted value; so using the keyId as the encryption value isn’t allowed.
Here’s a record example:
You can set "keySecretId": "{{record.topic}}-{{record.header.someHeader}}-{{record.key}}" - this will create an encryption key called myTopic-myHeader-myKey in memory. If you want this key to be stored in your Vault KMS, you can set: "keySecretId": "vault-kms://https://vault:8200/transit/keys/{{record.topic}}-{{record.header.someHeader}}-{{record.key}}".

KMS integration

Any keySecretId that doesn’t match one of the schemas detailed below will be rejected and the encryption operation will fail.
Keys are strings that start with a letter followed by a combination of letters, underscores (_), hyphens (-) and numbers. Special characters are not allowed.They also work with the mustache pattern.
If you want to make sure the key is well created in your KMS, you have to (1) make sure you have configured the connection to the KMS and (2) use the following format as keySecretId:
In-memory and test-tokenization modes are for testing and development purposes only. Test tokenization also requires GATEWAY_FEATURE_FLAGS_TEST_TOKENIZATION to be set to TRUE
If not specified, vault KMS scheme defaults to https. This means that vault-kms://https://vault:8200/transit/keys/password-key-id and vault-kms://vault:8200/transit/keys/password-key-id are identical.
Keys are strings that start with a letter followed by a combination of letters, underscores (_), hyphens (-) and numbers. Special characters are not allowed. Keys also work with the Mustache pattern described above.

Tokenization

Tokenization is an alternative to encryption for protecting sensitive data in Kafka messages. Instead of encrypting values, sensitive data is replaced with tokens, while also storing the original values securely in HashiCorp Vault’s transform secrets engine. Key benefits of tokenization:
  • Deterministic tokens: the same input always generates the same token, enabling queries on tokenized data
  • Format-preserving transformation: tokens can maintain the format of the original data
Tokenization is configured using the existing encryption Interceptors (EncryptPlugin, DecryptPlugin, etc.) but only works with the Transform Secrets Engine in Vault Enterprise (find out how to configure Vault). To tokenize data, use the encryption Interceptors with a vault-transform:// prefix in your keySecretId. To de-tokenize and retrieve the original values, use the decryption Interceptors with the same configuration. Here’s a sample field configured for tokenization:

Supported algorithms

  • AES128_GCM (default)
  • AES128_EAX
  • AES256_EAX
  • AES128_CTR_HMAC_SHA256
  • AES256_CTR_HMAC_SHA256
  • CHACHA20_POLY1305
  • XCHACHA20_POLY1305
  • AES256_GCM

Choosing an encryption algorithm

Gateway supports multiple encryption algorithms, with AES128_GCM as the default. When selecting an algorithm, consider your security requirements, performance needs, and message volume. Default algorithm: AES128_GCM AES128_GCM is the default algorithm and is suitable for most use cases. However, it has an important security limitation:
Security consideration: When the same DEK (Data Encryption Key) is used to encrypt approximately 4 billion (2³²) messages, AES-GCM’s security guarantees degrade due to nonce collision risks. According to NIST Special Publication 800-38D (Section 8.3), nonce collisions may expose encryption keys, compromising the confidentiality and integrity of data encrypted with that key.For high-traffic scenarios, this threshold can be reached quickly. For example:
  • At 10-50MB/s with 1KB message sizes, the 4-billion-message threshold can be reached in just over 24 hours
  • Each unique keySecretId uses its own DEK, so the limit applies per key, not globally
If you expect a single DEK to encrypt more than ~2³² messages, consider:
  • Using a different algorithm (see recommendations below)
  • Implementing DEK rotation before reaching the threshold
  • Using multiple keySecretId values to distribute the message count across multiple DEKs
When to keep the default (AES128_GCM)
  • Low to moderate message volume per DEK (well below 2³² messages per key)
  • Need for compatibility with existing AES-GCM implementations
  • Hardware acceleration (AES-NI) is available, providing good performance
If you expect a single DEK to encrypt more than ~2³² messages, consider using alternative algorithms or implementing DEK rotation. Consult your cryptographic library documentation and security requirements to choose the appropriate algorithm for your use case.

Key rotation

Gateway uses envelope encryption with two types of keys: DEK (Data Encryption Key) and KEK (Key Encryption Key). Understanding how and when to rotate these keys is important for maintaining security. KEK rotation
Customer responsibility: KEK (Key Encryption Key) rotation must be performed by the customer and is not handled automatically by Gateway. This is the customer’s responsibility and must be done through your KMS provider (AWS KMS, Azure Key Vault, HashiCorp Vault, GCP KMS, etc.).
When to rotate KEK You should rotate your KEK based on:
  • Security policies: Follow your organization’s key rotation policies and compliance requirements
  • DEK encryption frequency: If DEKs are being encrypted frequently (high message volume), consider more frequent KEK rotation
  • Security incidents: Rotate immediately if a KEK is suspected to be compromised
  • Best practices: Many organizations rotate KEKs annually or quarterly, but the frequency should match your security requirements
After rotating a KEK, Gateway will automatically use the new KEK version for encrypting new DEKs. However, existing EDEKs encrypted with the old KEK version will still be de-cryptable as long as the old KEK version remains available in your KMS.Most KMS providers retain old key versions for backward compatibility, allowing you to decrypt historical data while new data uses the rotated key.

Supported compression types

  • none
  • gzip
  • snappy
  • lz4
  • zstd

Encryption error policy

This policy determines the actions when an encryption Interceptor encounters a record that’s already encrypted. Example configuration with error policy:

Decryption configuration

Now that your fields or payload are encrypted, you can decrypt them using the Interceptor DecryptPlugin.

Decryption error policy

This policy determines the action if there is an error during decryption. Gateway supports Confluent-like and AWS Glue schema registries. If you don’t supply a basicCredentials section for the AWS Glue schema registry, the client will attempt to find the connection information it needs from the environment (see AWS docs for details ) and the credentials required can be passed this way to Gateway as part of its core configuration.

KMS configuration

Find out how to configure the different KMS within your encrypt and decrypt Interceptors.

Configuration properties

Choose your KMS provider

In-memory KMS

This should not be used on production data.
Keys in in-memory KMS are not persisted, this means that if you do one of the following, you won’t be able to decrypt old records, losing the data.
  • Use a Gateway cluster with more than a single node or
  • restart Gateway or
  • change the Interceptor configuration

Gateway KMS

This KMS type is effectively a delegated storage model and is designed to support encryption use cases which generate unique secret Ids per record or even field (typically via the Mustache template support for a secret Id). This technique is used in crypto-shredding type scenarios e.g. encrypting records per user with their own key.It provides the option to leverage your KMS for security via a single master key, but efficiently and securely store many per-record level encryption keys (DEKs) in the Gateway managed store. For some architectures this can provide performance and cost savings for encryption use cases which generate a high volume of secret key Ids.The masterKeyId is used to secure every key for this configuration, stored by Gateway. Find out more about the secret key formats. You have to also supply a valid configuration for the KMS type referenced by the master key so this can be used.If this key is dropped from the backing KMS, then all keys stored by Gateway for that master key will become unreadable.Gateway KMS encryption exampleHere’s a sample configuration for the Gateway KMS using a Vault-based master key:
This can then be used to encrypt a field using gateway-kms:// as the secret key type:
When processing a record for the first time using this configuration, Gateway will:
  1. generate a DEK to encrypt the field data,
  2. turn it into an EDEK by encrypting with the masterKeyId secret from vault and
  3. store the EDEK in Gateway storage.
If a record key was 123456, the associated EDEK would be stored on a kafka record with the following key:
Multiple records produced against this config would cause multiple EDEKs to be saved in the Gateway storage (due to the {{record.key}} template giving a unique key for each Kafka record key).If there are multiple Gateway nodes running, it’s also possible for multiple DEKs/EDEKs to be generated for the same record key. Two nodes processing different records with the same record key at the same time could both assume they were generating a DEK/EDEK for the first time. In this scenario, there would be two EDEKs in the Gateway storage with the same keyId but they would each have a different UUID.For example:
Nonetheless, there will only ever be one master key stored in the vault KMS, which is used to encrypt every DEK.This feature provides flexibility for your KMS storage and key management setups - and is particularly useful for high volume crypto shredding.Decryption using Gateway KMSWhen using the gateway-kms secret key Id type, the decryption configuration used to decrypt the data has to also specify the masterKeyId, so that it can securely decrypt the keys stored in the local Gateway storage.Here’s a sample setup:
Crypto shreddingWhen using the gateway-kms secret key Id type, you can efficiently crypto shred EDEKs in the Gateway storage, so that anyone using the decryption plugin will immediately lose access to the associated encrypted data.To do this, scan the Gateway storage Kafka topic (by default, _conduktor_gateway_encryption_keys) for every message matching the associated qualified secret Id.For example, a qualified secretId of gateway-kms://fieldKeySecret-name-123456 might have the following keys:
Publishing a message for each of these keys back to the same topic with a value of null (i.e. a tombstone) will effectively perform Crypto Shredding.This process won’t prevent the creation of new keys if new messages are sent using the same record key; it only ensures that messages using the crypto shredded keys remain unrecoverable.

AWS KMS

To set your AWS KMS, include this section in your Interceptor config, below aws.You can use one of these two authentication methods:
  • basic authentication or
  • session.
Make sure to follow the right method and provided the correct properties.For enhanced security, you can hide the sensitive values using environment variables as secrets.

Azure KMS

To set your Azure KMS, include this section in your Interceptor config, below azure.You can use one of these two authentication methods:
  • token or
  • username and password.
Make sure you’ve followed the right method, and that you’ve provided the correct properties.For enhanced security, you can hide the sensitive values using environment variables as secrets.

Fortanix KMS

To set your Fortanix KMS, include this section in your Interceptor config under fortanix:Alternatively, you can use the environment variables for sensitive credentials and configuration values. Set them in your Gateway deployment and they will be resolved at runtime. In that case, you don’t have to supply a Fortanix {} block in the KMS config.If the specified encryption key doesn’t exist in Fortanix DSM, Gateway will automatically create it with the following configuration:
  • Key type: AES symmetric key
  • Mode: CBC
  • Key Size: 256
  • Permissions: ENCRYPT, DECRYPT

Google Cloud Platform KMS

To set your Google Cloud Platform (GCP) KMS, include this section in your Interceptor config, under gcp:You must first configure the service account key file.For enhanced security, you can hide the sensitive values using environment variables as secrets.

Vault KMS

Gateway supports two data security backends for Vault:To configure Vault, add a new Vault section in your Interceptor configuration. For enhanced security, you can hide the sensitive values using the environment variables as secrets.TrustStore
When using HTTPS Vault URIs without an explicitly configured trust store, Gateway will attempt to use the VAULT_SSL_TRUST_STORE_PATH and VAULT_SSL_TRUST_STORE_PASSWORD environment variables if available.
KeyStore
The Vault client library requires that JKS key stores contain both a private key and client certificate with identical passwords. When using HTTPS Vault URIs without an explicitly configured key store, Gateway will attempt to use the VAULT_SSL_KEY_STORE_PATH and VAULT_SSL_KEY_STORE_PASSWORD environment variables if available.
Vault authentication typesExample:
Connection backoffIf there’s a connection failure (e.g., a network partition error), Gateway will automatically keep trying to reconnect to Vault.You can adjust this by modifying the kmsConfig.vault.connectionBackoff object:Example:
Configuring Vault Transit Secrets EngineHere’s the minimum Vault policy required for encryption and decryption to work with vault:// prefixed keys:
Configuring Vault Transform Secrets EngineHere’s the minimum Vault policy required for encryption and decryption to work with vault-transform:// prefixed keys,
When using the Vault transform secrets engine, you can configure additional caching using the kmsConfig.vault.transformEngineCache object:Example:

Client throttling configuration

When encryption or decryption operations fail, you can configure client throttling to help protect your system from being overwhelmed during error conditions. To enable client throttling, set the throttleTimeMs parameter in your encryption/decryption Interceptor config:
  • throttleTimeMs = 0 (default): no throttling - clients receive immediate error responses
  • throttleTimeMs > 0: clients will be throttled for the specified time in milliseconds when operations fail
When throttleTimeMs is configured with a value greater than 0:
  • cluster stability is protected: Gateway automatically throttles clients to prevent system overload
  • client compliance is built-in: Kafka clients automatically pause for the specified throttle time between requests
  • failure cascades are prevented: throttling reduces retry pressure, allowing brokers to recover
Configuration example: