Versions Compared

Key

  • This line was added.
  • This line was removed.
  • Formatting was changed.

...

Configuration

KeyDescriptionTypeDefaultRequired
value.serializersList 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[]>`.StringNoneYes
value.deserializersList 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[]>`.StringsNoneYes

2. LargeMessageSerializer

Configuration

KeyDescriptionTypeDefaultRequired
large.message.payload.store.classImplementation of org.apache.kafka.common.serialization.largemessage.store.PayloadStore .StringNoneYes
large.message.payload.store.timeout.msTimeout for payload store operations. This can't exceedmax.block.msfor Kafka producers ormax.poll.interval.msin Kafka consumer. This is a part of basic configurations for any implementation of org.apache.kafka.common.serialization.largemessage.store.PayloadStoreLong10000No
large.message.payload.store.retry.countNumber of retries for payload store operations. This is a part of basic configurations for any implementation of org.apache.kafka.common.serialization.largemessage.store.PayloadStoreInt5No
large.message.payload.store.retry.max.backoff.msBackoff time between retries for payload store operations. This is a part of basic configurations for any implementation of org.apache.kafka.common.serialization.largemessage.store.PayloadStoreLong10000No
large.message.payload.store.retry.delay.backoff.msDelay time between retries back off for payload store operations. This is a part of basic configurations for any implementation of org.apache.kafka.common.serialization.largemessage.store.PayloadStoreLong100No
large.message.threshold.bytes The maximum size of the message is considered large.Long1MB (Default of max.message.bytes)No





...

Configuration

KeyDescriptionTypeDefaultRequired
large.message.payload.store.classImplementation of org.apache.kafka.common.serialization.largemessage.store.PayloadStore .StringNoneYes
large.message.payload.store.timeout.msTimeout for payload store operations. This can't exceedmax.block.msfor Kafka producers ormax.poll.interval.msin Kafka consumer. This is a part of basic configurations for any implementation of org.apache.kafka.common.serialization.largemessage.store.PayloadStoreLong10000No
large.message.payload.store.retry.countNumber of retries for payload store operations. This is a part of basic configurations for any implementation of org.apache.kafka.common.serialization.largemessage.store.PayloadStoreInt5No
large.message.payload.store.retry.max.backoff.msBackoff time between retries for payload store operations. This is a part of basic configurations for any implementation of org.apache.kafka.common.serialization.largemessage.store.PayloadStoreLong10000No
large.message.payload.store.retry.delay.backoff.msDelay time between retries back off for payload store operations. This is a part of basic configurations for any implementation of org.apache.kafka.common.serialization.largemessage.store.PayloadStoreLong100No
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. This is important when external store ttl is smaller than kafka retention.BooleanFALSENo





4. PayloadStore

Code Block
languagejava
/**
* 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 PayloadStore implements Configurable, Closeable {

    Integer maxRetries;
    Integer timeoutMs;
    Long maxBackoffMs;
    Long delayBackoffMs;
    protected Metrics metrics;     
    
    @Override
    public void configure(Map<String, ?> configs) {
        // configure
    }

     /**
     * 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();
    }
}

...

Code Block
languagejava
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.payload.store.class", "CustomS3Store")
 producerConfig.put("s3.bucket", "my-bucket")
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.

Compatibility, Deprecation, and Migration Plan

...