> ## Documentation Index > Fetch the complete documentation index at: https://docs.conduktor.io/llms.txt > Use this file to discover all available pages before exploring further. # Large Kafka message handling — offload to cloud storage > Handle oversized Kafka messages by offloading payloads to S3, GCS, or Azure Blob with Conduktor Gateway. Enterprise The Interceptor for handling large messages/batches will save the actual messages produced to Gateway 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) | | Amazon S3 configuration. Mutually exclusive with `azureConfig`. | | azureConfig | [Azure](#azure-blob-storage) | | 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](#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//`. #### 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 batches As of Gateway v3.18.0, the `LargeBatchHandlingPlugin` has been deprecated and will be removed in v3.21.0. Use the [large message handling plugin](#large-messages) instead, which handles individual messages above the size threshold. Each *batch* that's above the `minimumSizeInBytes` threshold will be saved in **one file** on Amazon S3, with credentials defaulting to managed identity: ```bash theme={null} curl \ --request PUT \ --url 'http://localhost:8888/gateway/v2/interceptor' \ --header 'Authorization: Basic YWRtaW46Y29uZHVrdG9y' \ --header 'Content-Type: application/json' \ --data-raw '{ "name": "myLargeBatchHandlingPlugin", "pluginClass": "io.conduktor.gateway.interceptor.LargeBatchHandlingPlugin", "priority": 100, "config": { "topic": "topic.*", "minimumSizeInBytes": 1024, "localDiskDirectory": "myStorage/", "s3Config": { "bucketName": "myBucketName", "uri": "http://myexampleuri", "region": "us-east-1" } } }' ``` ```yaml theme={null} apiVersion: gateway/v2 kind: Interceptor metadata: name: myLargeBatchHandlingPlugin scope: vCluster: passthrough spec: pluginClass: io.conduktor.gateway.interceptor.LargeBatchHandlingPlugin priority: 100 config: topic: topic.* minimumSizeInBytes: 1024 localDiskDirectory: myStorage/ s3Config: bucketName: myBucketName uri: http://myexampleuri region: us-east-1 ``` Apply with: ```bash theme={null} conduktor apply -f large-batch-handling-interceptor.yaml ``` With `basic credentials`: ```bash theme={null} curl \ --request PUT \ --url 'http://localhost:8888/gateway/v2/interceptor' \ --header 'Authorization: Basic YWRtaW46Y29uZHVrdG9y' \ --header 'Content-Type: application/json' \ --data-raw '{ "name": "myLargeBatchHandlingPlugin", "pluginClass": "io.conduktor.gateway.interceptor.LargeBatchHandlingPlugin", "priority": 100, "config": { "topic": "topic.*", "minimumSizeInBytes": 1024, "localDiskDirectory": "myStorage/", "s3Config": { "accessKey": "myAccessKey", "secretKey": "mySecretKey", "bucketName": "myBucketName", "uri": "http://myexampleuri", "region": "us-east-1" } } }' ``` ```yaml theme={null} apiVersion: gateway/v2 kind: Interceptor metadata: name: myLargeBatchHandlingPlugin scope: vCluster: passthrough spec: pluginClass: io.conduktor.gateway.interceptor.LargeBatchHandlingPlugin 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: ```bash theme={null} conduktor apply -f large-batch-handling-basic-creds-interceptor.yaml ``` With `session credentials`: ```bash theme={null} curl \ --request PUT \ --url 'http://localhost:8888/gateway/v2/interceptor' \ --header 'Authorization: Basic YWRtaW46Y29uZHVrdG9y' \ --header 'Content-Type: application/json' \ --data-raw '{ "name": "myLargeBatchHandlingPlugin", "pluginClass": "io.conduktor.gateway.interceptor.LargeBatchHandlingPlugin", "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" } } }' ``` ```yaml theme={null} apiVersion: gateway/v2 kind: Interceptor metadata: name: myLargeBatchHandlingPlugin scope: vCluster: passthrough spec: pluginClass: io.conduktor.gateway.interceptor.LargeBatchHandlingPlugin 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: ```bash theme={null} conduktor apply -f large-batch-handling-session-creds-interceptor.yaml ``` ## 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: ```bash theme={null} curl \ --request PUT \ --url 'http://localhost:8888/gateway/v2/interceptor' \ --header 'Authorization: Basic YWRtaW46Y29uZHVrdG9y' \ --header 'Content-Type: application/json' \ --data-raw '{ "name": "myLargeMessageHandlingPlugin", "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" } } }' ``` ```yaml theme={null} 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: ```bash theme={null} conduktor apply -f large-message-handling-interceptor.yaml ``` With `basic credentials`: ```bash theme={null} curl \ --request PUT \ --url 'http://localhost:8888/gateway/v2/interceptor' \ --header 'Authorization: Basic YWRtaW46Y29uZHVrdG9y' \ --header 'Content-Type: application/json' \ --data-raw '{ "name": "myLargeMessageHandlingPlugin", "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" } } }' ``` ```yaml theme={null} 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: ```bash theme={null} conduktor apply -f large-message-handling-basic-creds-interceptor.yaml ``` With `sessionCredentials`: ```bash theme={null} curl \ --request PUT \ --url 'http://localhost:8888/gateway/v2/interceptor' \ --header 'Authorization: Basic YWRtaW46Y29uZHVrdG9y' \ --header 'Content-Type: application/json' \ --data-raw '{ "name": "myLargeMessageHandlingPlugin", "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" } } }' ``` ```yaml theme={null} 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: ```bash theme={null} 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](/guide/conduktor-concepts/interceptors#how-they-work) for more on priority and execution paths, and [the encryption configuration reference](/guide/tutorials/configure-encryption) for the full encryption Interceptor configuration.