...

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 org.apache.kafka.common.serialization.largemessage.Serializer 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.

...

org.apache.kafka.common.serialization.largemessage.Serializer

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 alarge-message: trueheader
  • If it’s not large then provided threshold (large.message.threshold.bytes):
    • Do nothing, pass the data as it is.

3.

...

org.apache.kafka.common.serialization.largemessage.Deserializer

A configurable Kafka deserializer that performs:

...

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

...

Serializer<byte[]>

Configuration

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

...

Deserializer<byte[]>

Configuration

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;
     }
}

...

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

/**
* An interface for publishing and downloading serialised data to/from payload store.
* The store config will passed down from the original config of the Kafka producer/consumer client. 
*/
public interface PayloadStore implements Configurable, Closeable, Monitorable {
     /**
     * Publish data into the store.
     *
     * @param data data that will be published to the store.
     * @return full path to object in the store.
     * @throw PayloadStoreException in case failed to publish to the store.
     */
     String publish(String topic, byte[] data) throw PayloadStoreException;

    /**
     * Download full data from the store.
     *
     * @param fullPath of the data's reference in the store for example `remote_store/topic_name/<record_random_uuid>`
     * @return content of the object as bytes.
     * @throw PayloadStoreException 
     */
     byte[] download(String fullPath) throw PayloadStoreException;

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

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

/**
* Exception class that represent canexceptions eitherduring beinteraction reliablewith orthe notstore.
* this helps the Payload serializer/desrializerstore to decided either to retry or to crash. 
* OneThe subclassfinal willSerializer beand addedDeserializer iswill PayloadNotFoundExceptionpropagate whichthis is used to indicated if the payload not found 
* This is used by deserializer to skip or notas SerializationException to client.
**/
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);
    }
}

...

Map<String, Object> producerConfig = new HashMap<>();
producerConfig.put("value.serializers", "kafka.serializers.KafkaAvroDeserializer, org.apache.kafka.common.serialization.LargeMessageSerializerSerializer");
producerConfig.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer"); producerConfig.put("large.message.payload.store.class", "myclient.serializers.payload.store.CustomS3Store")
producerConfig.put("large.message.threshold.bytes", 1048576);
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);

...

• 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 configurationcapture SerializationException::getCause and decide what to do if the exception is PayloadException/PayloadNotFoundException.

  • This allows graceful handling of missing payload references

...

• Partial Failure Handling: If payload storage succeeds but Kafka produce fails, the payload will remain in storage until TTL expires. This is acceptable as it only affects storage costs, not correctness. If Kafka message is consumed but payload download fails, the large.message.skip.not.found.error  configuration determines client can capture SerializationException::getCause and determine behavior.

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

• Memory Constrains: Large messages will consume memory during serialization so ensure heap size can accommodate your largest expected message. ConsiderPayloadStoreimplementationsthatsupportcompressiontoreducestoragefootprint.

...