...

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. 

...

• Check if the estimated size of the data (bytes) after applying provided compression (if there is one) it needs to serialize is bigger larger than the configured threshold.
  • If it is large than the provided threshold (`large.message.threshold.bytes`):
    • Use the provided BlobStore
    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 alarge-message: trueheader
  • If it’s not large then provided threshold (large.message.threshold.bytes):
    • Do nothing, pass the data as it is.
• If it is
• If it’s not

3. LargeMessageDeserializer

...

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

Proposed Changes

1. Composable Serializer/Deserializer

...

...

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

...