Large Kafka message handling — offload to cloud storage

The for handling large messages will save the actual messages produced to into a cloud storage service. This helps to protect data or optimize storage in actual Kafka. We currently support:

Amazon S3 (Amazon Simple Storage Service) is a service offered by AWS (Amazon Web Services) that provides object storage through a web service interface.
Azure Blob Storage is a service offered by Microsoft Azure that provides blob storage.

Configuration

Key	Type	Default	Description
topic	string	`.*`	Topics matching this regex have the Interceptor applied.
s3Config	S3		Amazon S3 configuration. Mutually exclusive with `azureConfig`.
azureConfig	Azure		Azure Blob Storage configuration. Mutually exclusive with `s3Config`.
minimumSizeInBytes	int	`0`	Records with `sizeInBytes ≥ minimumSizeInBytes` are offloaded to cloud storage. Smaller records pass through unchanged.
localDiskDirectory	string	`${java.io.tmpdir}/myStorage/`	Path to the local on-disk cache used for both produce-time mirroring and fetch-time caching. See Local disk cache.
localCacheItemSize	int	`10000`	Maximum number of cached files (entries, not bytes). Oldest entries are evicted past this count.
localCacheExpireAfterWriteInSeconds	long	`3600`	Time-to-live since the cache entry was written. Expired entries are removed (and their files deleted) lazily on subsequent activity.

Local disk cache

The interceptor maintains a local on-disk cache at localDiskDirectory that mirrors objects stored in cloud storage (S3 or Azure Blob). The same path and behavior apply to both backends.

When files are written to disk

On produce. Every record ≥ minimumSizeInBytes is uploaded to cloud storage and a copy is written to localDiskDirectory so subsequent fetches do not need to re-download.
On fetch (cache miss). When a consumer fetches a record whose payload lives in cloud storage, Gateway downloads it and writes a copy to localDiskDirectory as well.

Each cached object is one file under localDiskDirectory/<topic>/<uuid>.

When files are removed

Files are removed only when their cache entry is evicted, which is triggered by cache activity (addition or removal). The eviction is determined based on two cases:

Count-based: more than localCacheItemSize entries are present (default 10000). The least-recently-used entry is evicted.
TTL-based: an entry is older than localCacheExpireAfterWriteInSeconds since it was written (default 1 hour). Expired entries are removed by subsequent cache activity, not by a wall-clock timer.

The cache has no byte-size limit. With the defaults, up to 10 000 × max-record-size can accumulate on disk before count-based eviction kicks in. If the volume is smaller than that, lower localCacheItemSize (and/or localCacheExpireAfterWriteInSeconds) to keep peak usage within the volume. There is no automatic cleanup on Gateway restart. Files written by a previous Gateway process remain on disk and continue to count against the volume until the cache is rebuilt and entries are evicted, or until they are removed manually.

Sizing the volume

As a starting point, size the volume so that

localCacheItemSize  ×  expected_max_record_size  +  headroom  ≤  volume_capacity

If the volume is constrained, lower localCacheItemSize first. This is the only setting that strictly bounds the worst-case file count.

Permissions and startup behavior

If localDiskDirectory does not exist, Gateway attempts to create it at interceptor configuration time. If creation fails, the interceptor configuration is rejected.
If the directory exists but is not writable by the Gateway process, the interceptor configuration is rejected.
The Gateway process needs read and write permission for the directory and for files it creates inside it.

Default location

If localDiskDirectory is omitted, Gateway uses ${java.io.tmpdir}/myStorage/ (typically /tmp/myStorage/ because Gateway container runs Linux). For production deployments, set localDiskDirectory explicitly to a properly sized, persistent volume rather than relying on the system temp directory.

Amazon S3

The S3 credentials default to managed identity. They will be overwritten if a specific basic credentials (accessKey and secretKey) or session credentials (accessKey, secretKey and sessionToken) are configured. For S3-compatible storage (MinIO, NetApp ONTAP, Dell ECS, Cloudflare R2), set disableChecksums to true. Otherwise, the AWS SDK sends a checksum in a form these backends do not accept, and uploads fail (the backend may report that the checksum header does not match).

Key	Type	Description
accessKey	string	S3 access key
secretKey	string	S3 secret key
sessionToken	string	S3 session token
bucketName	string	S3 bucket name
uri	string	S3 URI
region	string	S3 region
disableChecksums	boolean	Set to true for S3-compatible storage (MinIO, NetApp ONTAP, Dell ECS, Cloudflare R2) so uploads succeed; the default checksum format is not accepted by these backends. Default: false.

Azure Blob Storage

Note that your application will require at least Storage Blob Data Contributor permissions to be able to read/write the data.

Key	Type	Description
tenantId	string	Azure tenant ID
clientId	string	Azure client ID
secret	string	Azure client secret
blobEndpoint	string	Azure Blob Storage endpoint to use
bucketName	string	Bucket (container) name in Blob Storage configured to store in

Large messages

Each individual message that’s above the minimumSizeInBytes threshold will be saved in one file on Amazon S3, with credentials defaulting to managed identity:

curl
Conduktor CLI

curl \
  --request PUT \
  --url 'http://localhost:8888/gateway/v2/interceptor' \
  --header 'Authorization: Basic YWRtaW46Y29uZHVrdG9y' \
  --header 'Content-Type: application/json' \
  --data-raw '{
  "kind": "Interceptor",
  "apiVersion": "gateway/v2",
  "metadata": {
    "name": "myLargeMessageHandlingPlugin"
  },
  "spec": {
    "pluginClass": "io.conduktor.gateway.interceptor.LargeMessageHandlingPlugin",
    "priority": 100,
    "config": {
      "topic": "topic.*",
      "minimumSizeInBytes": 1024,
      "localDiskDirectory": "myStorage/",
      "s3Config": {
        "bucketName": "myBucketName",
        "uri": "http://myexampleuri",
        "region": "us-east-1"
      }
    }
  }
}'

apiVersion: gateway/v2
kind: Interceptor
metadata:
  name: myLargeMessageHandlingPlugin
  scope:
    vCluster: passthrough
spec:
  pluginClass: io.conduktor.gateway.interceptor.LargeMessageHandlingPlugin
  priority: 100
  config:
    topic: topic.*
    minimumSizeInBytes: 1024
    localDiskDirectory: myStorage/
    s3Config:
      bucketName: myBucketName
      uri: http://myexampleuri
      region: us-east-1

Apply with:

conduktor apply -f large-message-handling-interceptor.yaml

With basic credentials:

curl
Conduktor CLI

curl \
  --request PUT \
  --url 'http://localhost:8888/gateway/v2/interceptor' \
  --header 'Authorization: Basic YWRtaW46Y29uZHVrdG9y' \
  --header 'Content-Type: application/json' \
  --data-raw '{
  "kind": "Interceptor",
  "apiVersion": "gateway/v2",
  "metadata": {
    "name": "myLargeMessageHandlingPlugin"
  },
  "spec": {
    "pluginClass": "io.conduktor.gateway.interceptor.LargeMessageHandlingPlugin",
    "priority": 100,
    "config": {
      "topic": "topic.*",
      "minimumSizeInBytes": 1024,
      "localDiskDirectory": "myStorage/",
      "s3Config": {
        "accessKey": "myAccessKey",
        "secretKey": "mySecretKey",
        "bucketName": "myBucketName",
        "uri": "http://myexampleuri",
        "region": "us-east-1"
      }
    }
  }
}'

apiVersion: gateway/v2
kind: Interceptor
metadata:
  name: myLargeMessageHandlingPlugin
  scope:
    vCluster: passthrough
spec:
  pluginClass: io.conduktor.gateway.interceptor.LargeMessageHandlingPlugin
  priority: 100
  config:
    topic: topic.*
    minimumSizeInBytes: 1024
    localDiskDirectory: myStorage/
    s3Config:
      accessKey: myAccessKey
      secretKey: mySecretKey
      bucketName: myBucketName
      uri: http://myexampleuri
      region: us-east-1

Apply with:

conduktor apply -f large-message-handling-basic-creds-interceptor.yaml

With sessionCredentials:

curl
Conduktor CLI

curl \
  --request PUT \
  --url 'http://localhost:8888/gateway/v2/interceptor' \
  --header 'Authorization: Basic YWRtaW46Y29uZHVrdG9y' \
  --header 'Content-Type: application/json' \
  --data-raw '{
  "kind": "Interceptor",
  "apiVersion": "gateway/v2",
  "metadata": {
    "name": "myLargeMessageHandlingPlugin"
  },
  "spec": {
    "pluginClass": "io.conduktor.gateway.interceptor.LargeMessageHandlingPlugin",
    "priority": 100,
    "config": {
      "topic": "topic.*",
      "minimumSizeInBytes": 1024,
      "localDiskDirectory": "myStorage/",
      "s3Config": {
        "accessKey": "myAccessKey",
        "secretKey": "mySecretKey",
        "sessionToken": "mySessionToken",
        "bucketName": "myBucketName",
        "uri": "http://myexampleuri",
        "region": "us-east-1"
      }
    }
  }
}'

apiVersion: gateway/v2
kind: Interceptor
metadata:
  name: myLargeMessageHandlingPlugin
  scope:
    vCluster: passthrough
spec:
  pluginClass: io.conduktor.gateway.interceptor.LargeMessageHandlingPlugin
  priority: 100
  config:
    topic: topic.*
    minimumSizeInBytes: 1024
    localDiskDirectory: myStorage/
    s3Config:
      accessKey: myAccessKey
      secretKey: mySecretKey
      sessionToken: mySessionToken
      bucketName: myBucketName
      uri: http://myexampleuri
      region: us-east-1

Apply with:

conduktor apply -f large-message-handling-session-creds-interceptor.yaml

Combine with encryption

When you combine encryption with LargeMessageHandlingPlugin, priority order matters in both directions. On produce, the encryption Interceptor must run before LargeMessageHandlingPlugin to ensure data offloaded to cloud storage is stored encrypted. On consume, DecryptPlugin must run after LargeMessageHandlingPlugin so the re-fetched ciphertext is decrypted before reaching the consumer.

If LargeMessageHandlingPlugin has a lower priority number than the encryption Interceptor, it offloads records before they’re encrypted. Cloud storage will hold plaintext.

Recommended priority order

EncryptPlugin — lowest priority (for example, 100)
LargeMessageHandlingPlugin — middle priority (for example, 200)
DecryptPlugin — highest priority (for example, 300)

Why this order works

Produce flow: records leave the client, EncryptPlugin runs first (priority 100) and encrypts configured fields or the full payload, then LargeMessageHandlingPlugin runs next (priority 200); if the encrypted record exceeds minimumSizeInBytes, it’s offloaded to cloud storage. The file in cloud storage contains ciphertext.
Consume flow: records leave the broker, LargeMessageHandlingPlugin runs first among the consume-path Interceptors (priority 200) and re-fetches the offloaded ciphertext from cloud storage into the local temp directory, then DecryptPlugin runs next (priority 300) and decrypts the ciphertext before the record reaches the consumer.

See how Interceptors execute for more on priority and execution paths, and the encryption configuration reference for the full encryption Interceptor configuration.

Kafka management

Enable teams

Secure and govern data

Optimize performance

Large Kafka message handling — offload to cloud storage

Configuration

Local disk cache

When files are written to disk

When files are removed

Sizing the volume

Permissions and startup behavior

Default location

Amazon S3

Azure Blob Storage

Large messages

Combine with encryption

Recommended priority order

Why this order works

​Configuration

​Local disk cache

​When files are written to disk

​When files are removed

​Sizing the volume

​Permissions and startup behavior

​Default location

​Amazon S3

​Azure Blob Storage

​Large messages

​Combine with encryption

​Recommended priority order

​Why this order works

Configuration

Local disk cache

When files are written to disk

When files are removed

Sizing the volume

Permissions and startup behavior

Default location

Amazon S3

Azure Blob Storage

Large messages

Combine with encryption

Recommended priority order

Why this order works