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).

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

This KIP is proposing

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

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 is proposing idea of having 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) it needs to serialize is bigger than the configured threshold.
  • Use a BlobStore like implementation to put the large message as blob/object into blob 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
  • Do nothing, pass the data as it is.
• If it is
• If it’s not

3. LargeMessageDeserializer

A configurable Kafka deserializer that performs:

• Check if the event is of the Large blob type by looking for large-message: true header
  • Parse the event and retrieve the ID and the needed useful information on the event.
  • Use a BlobStore implementation to get the blob from the blob/object-store.
  • Pass blob down
  • Do nothing return the Kafka message as it is
• If it is
• If it’s not

Proposed Changes

1. Composable Serializer/Deserializer

Configuration

2. LargeMessageSerializer

Configuration

3. LargeMessageDeserializer

Configuration

4. BlobStore

public abstract class BlobStore implements Configurable, Closeable {

    @Override
    public void configure(Map<String, ?> configs) {
        // configure
    }

    /**
     * Upload data into object store.
     *
     * @param data data that will be uploaded into the object.
     * @return {@link BlobResponse}.
     */
    public abstract BlobResponse putObject(String topic, byte[] data);

    /**
     * Download object from object store.
     *
     * @param blobPath id of the object in blobstore
     * @return content of the object as bytes.
     */
    public abstract byte[] getObject(String blobPath);
}

5. BlobIdGenerator

public interface BlobIdGenerator extends Configurable {
    BlobIdGenerator DEFAULT_BLOB_ID_GENERATOR = new BlobIdGenerator() {
        @Override
        public void configure(Map<String, ?> configs) {
            // Nothing to do with config
        }

        @Override
        public String id(byte[] data) {
            return UUID.randomUUID().toString();
        }
    };

    /**
     * generate blob id.
     */
    String id(byte[] data);
}

6. LargeBlobMessage

public interface LargeMessageFormatter<T> extends Configurable {
    LargeMessageFormatter JSON = new LargeMessageFormatter() {
        // implement a default json one
        // format is { 
        //          "full-blob-path": "<full-path-to-access-blob-as-string>",
        //          "blob-store-class": "<used-blob-store-class-path-to-upload-blob>",
        //          "blob-id-generator-class": "<class-used-to-generate-blob-id>",
        //          "large-message-formatter-class": "<formatter-class-path>"
        //        }
    };

    /**
     * build message bytes from blob store response and data.
     */
    byte[] messageBytes(byte[] data, BlobResponse response);

    /**
     * parse data into LargeBlobMessage.
     */
    T largeBlobMessage(byte[] data);
}

7. BlobResponse

public class BlobResponse {
    public final int responseCode;
    public final String fullBlobPath;
    public final BlobStoreException blobStoreException;

    /**
     * Construct blob response with response code and blob id.
     */
    public BlobResponse(int responseCode, String fullBlobPath) {
        this(responseCode, fullBlobPath, null);
    }

    /**
     * Construct blob response with response code, blob id and exception.
     */
    public BlobResponse(int responseCode, String fullBlobPath, BlobStoreException blobStoreException) {
        this.responseCode = responseCode;
        this.fullBlobPath = fullBlobPath;
        this.blobStoreException = blobStoreException;
    }
}

Example

Map<String, Object> producerConfig = new HashMap<>();
producerConfig.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
producerConfig.put("value.serializers",
        "org.apache.kafka.common.serialization.DoubleSerializer,org.apache.kafka.common.serialization.LargeMessageSerializer");
producerConfig.put("large.message.blob.store.class", "CustomS3Store")
producerConfig.put("bootstrap.servers", "localhost:9092");

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

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