Status

Current state: "Under Discussion"

Discussion thread: here

JIRA: here

Please keep the discussion on the mailing list rather than commenting on the wiki (wiki discussions get unwieldy fast).

Note: This KIP is a result of working with members of Apache Cassandra community who are working on CEP-44: Kafka integration for Cassandra CDC using Sidecar and one of the limitations they have faced was large message sizes

Motivation

Kafka has a limit for message size which limits some use cases where they might have messages that are larger than message.max.bytes even after enabling compression on the producer side or after applying serialization formats like Apache Avro or Protocol Buffers to reduce payload size. Increasing message size indefinitely is not a viable solution as it can lead to performance degradation, memory issues, and instability of the broker. And by looking at some of the enterprise/cloud offerings of Kafka you can see that on average they can offer 8MB to 10MB as max message size.

Popular Patterns to solve this

At the moment of writing this KIP, there are two famous patterns to handle without increasing message.max.bytes

1. Message Chunking (Splitting and Reassembly message)

Break large messages into smaller chunks, send them sequentially, and reassemble on the consumer side.

How it works

Split the payload into chunks smaller than message.max.bytes.
Assign the same key to all chunks to ensure they land in the same partition, preserving order.
Reassemble chunks on the consumer using metadata (e.g., sequence IDs, total chunk count).

Considerations

Requires custom logic for splitting/reassembling
Consumers must handle out-of-order or missing chunks

Available open-Source implementation:

https://github.com/linkedin/li-apache-kafka-clients#features

2. Reference-Based Messaging

Store the large payload externally and send only a reference (e.g URI, databases key) in the Kafka message.

How it works

Upload the payload to external storage (e.g., S3, HDFS, or a database).
Send a reference (e.g., s3://bucket/key) via Kafka.
Consumers fetch the payload using the reference

Considerations

Reduces Kafka network/storage load.
Introduces dependency on external systems.

Available open-source implementation

There isn't a specific open-source implementation for Reference-Based Messaging. Mostly everyone adopting this has their custom solution

Pattern	Pros	Cons
Chunking	No external storage required	Complex client logic to split and reassemble messages
Reference-Based	Minimizes Kafka load	External system dependency

This KIP is proposing

A serializer in Apache Kafka that implements Reference-Based Messaging as this is the simplest one.
A notion of a Composable serializer where the client can apply a list of serializers before applying a large message serializer

Note:

This KIP will benefit CEP-44: Kafka integration for Cassandra CDC using Sidecar

Public Interfaces

1. Composable Serializer/Deserializer

In some use-cases we might need a way to be able to chain of Kafka serializer/deserializer before applying large message serializer for example need to apply schema or special format like avro or protobuf. We could just create a single LargeMessageSerializer that implemented the Kafka Serializer interface, but we would need to create the different versions (AvroSerializer, ProtobufSerializer) with their support for the schema.
Instead the KIP proposing a composable serializer, that still implements the Kafka Serializer interface, but allows concatenating several serializers to perform what we are looking for here.

2. LargeMessageSerializer

A configurable Kafka serializer that:

Check if the estimated size of the data (bytes) after applying provided compression (if there is one) it needs to serialize is larger than the configured threshold.
- If it is large than the provided threshold (`large.message.threshold.bytes`):
  - Use the provided PayloadStore implementation to publish the large message into payload store, generating an id which is the reference to access this later.
  - Encapsulate that ID into a simple Kafka event using a structured format.
  - Pass the new Kafka Event down
  - Add a large-message: true header
- If it’s not large then provided threshold (large.message.threshold.bytes):
  - Do nothing, pass the data as it is.

3. LargeMessageDeserializer

A configurable Kafka deserializer that performs:

Check if the event is large message by looking for large-message: true header
- If it does have `large-message` header:
  - Parse the event and retrieve the ID and the needed useful information on the event.
  - Use the provided PayloadStore implementation to download the original data from the payload-store.
  - Return payload as the final value
- If it doesn't have the large-message header:
  - Do nothing return the Kafka message as it is

Proposed Changes

1. org.apache.kafka.common.serialization.ComposableSerializer and org.apache.kafka.common.serialization.ComposableDeserializer

Configuration

Key	Description	Type	Default	Required
value.serializers	List of serializers class that implements the `org.apache.kafka.common.serialization.Serializer` interface in order. The first serializer is `Serializer<T>` while the remaining serializers must be `Serializer<byte[]>`.	String	None	No, If not exist the code will fail back to `value.serializer`
value.deserializers	List of deserializer classes that implements the `org.apache.kafka.common.serialization.Deserializer` interface in order. The last deserializer must be `Deserializer<T>` while the rest must be `Deserializer<byte[]>`.	Strings	None	No, If not exist the code will fail back to `value.deserializer`

2. org.apache.kafka.common.serialization.largemessage.LargeMessageSerializer

Configuration

Key	Description	Type	Default	Required
`large.message.payload.store.class`	Implementation of `org.apache.kafka.common.serialization.largemessage.store.PayloadStore` .	String	None	Yes
`large.message.threshold.bytes`	The maximum size of the message is considered large.	Long	1MB (Default of max.message.bytes)	No

3.org.apache.kafka.common.serialization.largemessage.LargeMessageDeserializer

Configuration

Key	Description	Type	Default	Required
`large.message.payload.store.class`	Implementation of `org.apache.kafka.common.serialization.largemessage.store.PayloadStore` .	String	None	Yes
`large.message.skip.not.found.error`	Skip not found error when the payload is not found in the store. This allows the deserializer to skip not-found messages and return empty bytes instead `new byte[0]` . This is important when external store ttl is smaller than kafka retention.	Boolean	FALSE	No

4.org.apache.kafka.common.serialization.largemessage.PayloadStore

/**
* The contract for any PayloadStore implementation. 
* And extract them from the provided configs. 
*/
public interface PayloadStore implements Configurable, Closeable, Monitorable {
     /**
     * Publish data into the store.
     *
     * @param data data that will be published to the store.
     * @return {@link PayloadResponse}.
     */
    public abstract PayloadResponse publish(String topic, byte[] data);

    /**
     * Download full data from the store.
     *
     * @param path id of the data's reference in the store
     * @return content of the object as bytes.
     */
    public abstract byte[] download(String path);

    /** 
    * Generate an id for the data's reference in the store. 
    * By default the id is a random UUID however some stores might need more smarter way to calculate its reference id. 
    * In such a case please override this method. 
    * @param data data that will be published to the store.
    public String id(byte[] data) {
       return UUID.randomUUID().toString();
    }
}

5. org.apache.kafka.common.serialization.largemessage.PayloadResponse

/**
* Response from publish / download from PayloadStore back to the serialization layer
* It contains the final path, response code and the encountered exception if there was any. 
* If PlayloadResponse contains PayloadStoreException with isRetryable flag then it will serialiser will
* retry. 
 */
public class PayloadResponse {
    public final String fullPayloadPath;
    public final PayloadStoreException payloadStoreException;
    /**
     * Construct payload response with response code and payload id.
     */
    public PayloadResponse(String fullPayloadPath) {
        this(fullPayloadPath, null);
    }

    /**
     * Construct payload response with payload id and exception.
     */
    public PayloadResponse(String fullPayloadPath, PayloadStoreException payloadStoreException) {
        this.fullPayloadPath = fullPayloadPath;
        this.payloadStoreException = payloadStoreException;
     }
}

6. org.apache.kafka.common.serialization.largemessage.PayloadStoreException

/**
* Exception class that can either be reliable or not
* this helps the serializer/desrializer to decided either to retry or to crash. 
* One subclass will be added is PayloadNotFoundException which is used to indicated if the payload not found 
* This is used by deserializer to skip or not.
**/
public class PayloadStoreException extends RuntimeException {
    /**
     * Constructor PayloadStoreException with message and throwable.
     */
    public PayloadStoreException(String message, Throwable t) {
        super(message, t);
    }

    /**
     * Constructor PayloadStoreException with message.
     */
    public PayloadStoreException(String message) {
        super(message);
    }

    /**
     * Constructor PayloadStoreException with throwable.
     */
    public PayloadStoreException(Throwable t) {
        super(t);
    }
}

Example

Map<String, Object> producerConfig = new HashMap<>();
producerConfig.put("key.serializers", "kafka.serializers.KafkaAvroDeserializer, org.apache.kafka.common.serialization.StringSerializer");
producerConfig.put("large.message.payload.store.class", "myclient.serializers.payload.store.CustomS3Store")
producerConfig.put("large.message.threshold.bytes", "1MB");
producerConfig.put("s3.bucket", "my-bucket")
producerConfig.put("s3.retry.attempts", "3");
producerConfig.put("s3.connection.timeout.ms", "5000");
producerConfig.put("bootstrap.servers", "localhost:9092");

KafkaProducer<String, Double> producer = new KafkaProducer<>(producerConfig);

Consideration:

TTL Configuration Risk: If the payload store owner doesn't configure an appropriate TTL that aligns with Kafka topic retention, the payload store may grow indefinitely. This occurs because objects remain in storage even after Kafka no longer references them, leading to unnecessary storage costs.
TTL Too Short Risk: If the TTL is set too aggressively (shorter than needed), Kafka references may point to objects that no longer exist in the payload store. When this happens:
- Consumers will encounter NOT_FOUND errors
- To prevent blocking behavior, consumers should enable the large.message.skip.not.found.error configuration
- This allows graceful handling of missing payload references

Recommendation: Set TTL duration to exceed your Kafka topic retention period by a reasonable buffer (e.g., 10-20%) to ensure payload availability throughout the message lifecycle while preventing indefinite storage growth.

Critical Timing Constraints: The total timeout for payload store operations (including all retries) cannot exceed `max.block.ms` for Kafka producers or `max.poll.interval.ms` for Kafka consumers. This is a fundamental requirement for any implementation of org.apache.kafka.common.serialization.largemessage.store.PayloadStore.
- Exceeding these limits will cause producer blocking or consumer rebalancing issues.

Compatibility, Deprecation, and Migration Plan

Old clients just need to set the needed configuration to use this feature

Rejected Alternatives

We have rejected implementing the chunking pattern due to its many potential edge cases and complexity added to the consumer side. A peak of those complexities can be explored more deeply in the LinkedIn presentation.
Implement this as a separate project outside of Apache Kafka as this seems to be a pattern that needs more use cases and it would be better to have this in Apache Kafka as native implementation instead of a separate project