...

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

...

Considerations

• 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

...