...

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 some folks from Apache Cassandra 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.

...

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

Note: 

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

Public Interfaces

1. Composable Serializer1. 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. 

...

• 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 BlobStore PayloadStore implementation to put publish the large message as blob/object into blob 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.

...

A configurable Kafka deserializer that performs:

• Check if the event is of the Large blob type by large message 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 the provided BlobStore PayloadStore implementation to get download the blob original data from the blob/objectpayload-store.
    • Return blob downpayload as the final value
  • If it doesn't have the large-message header:
    • Do nothing return the Kafka message as it is

...

...

4. BlobStore

4. PayloadStore

/**
* The contract for any PayloadStore implementation. 
* This parent abstract class will validate the initial configurations that any payload store must have, like large.message.payload.store.timeout.ms, large.message.payload.store.retry.count, large.message.payload.store.retry.max.backoff.ms and large.message.payload.store.retry.delay.backoff.ms. 
* And extract them from the provided configs. 
*/

public abstract class BlobStorePayloadStore implements Configurable, Closeable {

    Integer @OverridemaxRetries;
    public void configure(Map<String, ?> configs) {
   Integer timeoutMs;
    Long // configure
maxBackoffMs;
    Long }
delayBackoffMs;
    /**
protected Metrics metrics;     * Upload
 data into object store.
     *@Override
    public * @param data data that will be uploaded into the object.
     * @return {@link BlobResponse}.
     */void configure(Map<String, ?> configs) {
        // configure
    }

     /**
    public abstract* BlobResponsePublish putObject(String topic, byte[] data);

data into the store.
     /**
     * Download@param objectdata fromdata objectthat store.
will be published to the *store.
     * @param@return blobPath id of the object in blobstore{@link PayloadResponse}.
     */
    public *abstract @returnPayloadResponse content of the object as bytes.
 publish(String topic, byte[] data);

    /**/
     * Download publicfull abstractdata byte[] getObject(String blobPath);
}

5. BlobIdGenerator

public interface BlobIdGenerator extends Configurable {
from the store.
     *
     BlobIdGenerator DEFAULT_BLOB_ID_GENERATOR = new BlobIdGenerator() {
    * @param path id of the data's reference in the store
    @Override
 * @return content of the object as publicbytes.
 void configure(Map<String, ?> configs) {*/
    public abstract byte[]   download(String path);

    // Nothing to do with config** 
    * Generate an id }

for the data's reference in the store.  @Override
    * By default the publicid String id(byte[] data) {
            return UUID.randomUUID().toString();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 * generate blob id.
     */
will be published to the store.
    public String id(byte[] data) {
       return UUID.randomUUID().toString();
    }
}

6.

...

PayloadReferenceValue

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

/** PayloadReferenceValue represent the final published payload reference into Kafka. 
* It includes the full payload path on the store as well as the store class used for the publishing. 
* This allow the deserializer to throw a more informative errors if the original payload store class doesn't match the one setup by the consumer client. 
*/
{
  "apiKey": 0,
  "type": "payload-reference",
  "name": "PayloadReferenceValue",
  // Version 0 KIP-1159.
  "validVersions": "0",
  "fields": [
    { "name": "fullPayloadPath", "versions": "0+", "type": "string", "about": "The full path to access payload as string."},
    { "name": "payloadStoreClass", "versions": "0+", "type": "string","about": "The used payload store class path to upload payload."}
  ]
}

7. 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. 
*/
public class PayloadResponse

public class BlobResponse {
    public final int responseCode;
    public final String fullBlobPathpath;
    public final BlobStoreExceptionPayloadStoreException blobStoreExceptionPayloadStoreException;

    /**
     * Construct blobpayload response with response code and blobpayload id.
     */
    public BlobResponsePayloadResponse(int responseCode, String fullBlobPathpath) {
        this(responseCode, fullBlobPathpath, null);
    }

    /**
     * Construct blobpayload response with response code, blobpayload id and exception.
     */
    public BlobResponsePayloadResponse(int responseCode, String fullBlobPathpath, BlobStoreExceptionPayloadStoreException blobStoreExceptionpayloadStoreException) {
        this.responseCode = responseCode;
        this.fullBlobPathfullPayloadPath = fullBlobPathpath;
        this.blobStoreExceptionPayloadStoreException = blobStoreExceptionpayloadStoreException;
    }
}

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.blobpayload.store.class", "CustomS3Store")
 producerConfig.put("s3.bucket", "my-bucket")
producerConfig.put("bootstrap.servers", "localhost:9092");

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

...