Re: [DISCUSS] KIP-1159: Large message reference based Serializer

[Omnia Ibrahim <o....@gmail.com>]

Hi everyone,
If there is no any more feedback need to be addressed then I will open voting on this KIP early next week.

Regards 
Omnia

> On 28 May 2025, at 11:13, Omnia Ibrahim <o....@gmail.com> wrote:
> 
> Hi Luke, I had another round of updating the KIP. 
> I now made `PayloadStore::publish` returns a simplified version of “PayloadResponse” which will now will simply have the "fullPayloadPath", and “PayloadException”. 
> If PayloadResponse
> -  contains PayloadStoreException with `isRetriable` flag set to true then this will trigger the retry logic 
> - If response code is 404 and large.message.skip.not.found.error then we will skip failing on the download in the deserialiser side. 
> 
> WDYT? 
> 
> Regards 
> Omnia
> 
> 
>> On 22 May 2025, at 15:16, Omnia Ibrahim <o....@gmail.com> wrote:
>> 
>> Hi Luke, sorry for late response. 
>>> IMO, the retry should be the logic inside the
>>> "large.message.payload.store.class"
>>> implementation. If we really want it, I think we need to make it clear in
>>> which circumstance we will retry. For example, if it's an unknown exception
>>> thrown from S3 API, what will we do to it?
>> There was one class definition missing which is PayloadStoreException that is used in PayloadResponse. The PayloadStoreException has flag if it is true then we retry. And then serialised retry on any exception with this flag. We can make publish and download methods throw  PayloadStoreException directly to simplify things WDYT?
>> 
>>> Moving it into the store class makes it much clearer. And it's good to have
>>> a default implementation.
>>> But to me, it's still an implementation-level detail that we don't need to
>>> expose to users to implement it.
>>> Could I know more about when the Id generator will be invoked?
>>> My thought is :
>>> Users can implement a `publish` method to publish like this:
>>> public PayloadResponse publish(String topic, byte[] data) {
>>>  String id = genId();
>>>  // put the id and data to the remote storage
>>>  s3Client.put(id, data, ...);
>>> }
>>> 
>>> So, with the id method in the interface, who will invoke it? Suppose it
>>> will be the serializer/deserializer, but no one passes the generated id to
>>> the publish method, how do we use it?
>> 
>> It will be invoked by the payload class when publish the payload. So it is an implementation details of this class yes. 
>> 
>>> 7. Why do we need "PayloadResponse"? Why can't we return a String or a URL
>>> object?
>> 
>> Originally because I needed something bit of wrapper for re-try and skip logic. But as I mentioned in first point we can make publish and download methods return the ref or payload otherwise they throw PayloadStoreException directly to simplify things WDYT? Other option have publish return only ref as string and download would be the one that could throw PayloadNotFoundException. 
>> 
>>> 8. Could we change the abstract class to interface?
>> We can once we agree on what config we will drop and move to the details of payload provided by the users. 
>> 
>> regards
>> Omnia
>> 
>>> On 19 May 2025, at 09:54, Luke Chen <sh...@gmail.com> wrote:
>>> 
>>> Hi Omnia,
>>> 
>>> Thanks for the explanation and update.
>>> It's better now.
>>> 
>>> Questions:
>>>> 2. It's not clear how the "retry count" comes into play in the KIP. It
>>>> needs more explanation.
>>> My initial thinking is the retry configs are a must for all blob stores, so
>>> we can provide them, and validate them for free for all blob stores so not
>>> every implementation will go through verifying them.
>>> 
>>> IMO, the retry should be the logic inside the
>>> "large.message.payload.store.class"
>>> implementation. If we really want it, I think we need to make it clear in
>>> which circumstance we will retry. For example, if it's an unknown exception
>>> thrown from S3 API, what will we do to it?
>>> 
>>>> 6. About the BlobIdGenerator, why do we need it, could you explain more?
>>>> Again, I thought we only need to replace value to a path, and add the
>>>> "large-message" header, so that when consumer reads this message, it'll
>>>> read the path from value and get the original data via BlobStore. Why do
>>> we
>>>> need this ID generator? I think users should do the object naming when
>>>> putting the object by themselves, not via another interface. WDYT?
>>> In some cases generating the ID might need some smart work for example to
>>> avoid s3 throttling the recommended way on their doc it to create sub paths
>>> under the original bucket, to decide this we might hash the data to find a
>>> suitable sub-path.
>>> Here is an example of how I would generate an path for s3 file
>>> ```
>>> public String id(byte[] data) {
>>>   String subFolder = topic + "-" + Utils.toPositive(Utils.murmur2(data)) %
>>> distributionFactor // distributionFactor is a config for the Id generator
>>> and it represent the max number of sub-folders under the bucket
>>>   return subFolder + “/“ + UUID.randomUUID().toString()
>>> }
>>> ```
>>> Hope this example clarify a bit. However I do agree here it might not need
>>> a class. I have move it to be part of the store class.
>>> 
>>> Moving it into the store class makes it much clearer. And it's good to have
>>> a default implementation.
>>> But to me, it's still an implementation-level detail that we don't need to
>>> expose to users to implement it.
>>> Could I know more about when the Id generator will be invoked?
>>> My thought is :
>>> Users can implement a `publish` method to publish like this:
>>> public PayloadResponse publish(String topic, byte[] data) {
>>>  String id = genId();
>>>  // put the id and data to the remote storage
>>>  s3Client.put(id, data, ...);
>>> }
>>> 
>>> So, with the id method in the interface, who will invoke it? Suppose it
>>> will be the serializer/deserializer, but no one passes the generated id to
>>> the publish method, how do we use it?
>>> 
>>> 7. Why do we need "PayloadResponse"? Why can't we return a String or a URL
>>> object?
>>> 
>>> 8. Could we change the abstract class to interface?
>>> 
>>> Thanks.
>>> Luke
>>> 
>>> 
>>> 
>>> 
>>> On Wed, Apr 30, 2025 at 9:21 PM Omnia Ibrahim <o.g.h.ibrahim@gmail.com <ma...@gmail.com>>
>>> wrote:
>>> 
>>>> Hi Luke,
>>>> 
>>>>>> 3. What does "LargeMessageFormatter" do in the process?
>>>>>> I thought all we want to do is to replace the "large value data" into a
>>>>>> path, and consumers will read the path via blob store class.
>>>>>> All these should be done in serializer/deserializer, so why do we need
>>>> the
>>>>>> formatter?
>>>>> 
>>>>> I wanted to bit of more info than just the path to download, for example
>>>> I want to add stuff like the class path for the original blob store for
>>>> example if consumer is setup with the unmatched blob store to the one used
>>>> during publishing.
>>>>> I have updated the KIP to simplify this by having this always as a
>>>> simple json of path and publisher class which can be represented as
>>>> PayloadReferenceValue. WDYT?
>>>> 
>>>> I thought of another case where having the freedom to form the reference
>>>> might be nice feature, which is DR. Let’s imagine this case where someone
>>>> publish large messages to S3 and reference to Kafka topic then they want to
>>>> have DR. This can be achievable if they have mirrored Kafka topic which
>>>> contains the references but if S3 is unreachable form the DR backup
>>>> location then the reference they have is bit useless. However if the
>>>> message formatter is customisable then dev can implement a complicated
>>>> store that publish to two store locations and the publish both references
>>>> to Kafka as one message and the consumer store can download from either
>>>> buckets that are available. I think keeping the door open to such use-case
>>>> might be good feature but also having such use case might be questionable a
>>>> bit with the latency it will add as we will be publishing to N number of
>>>> stores.
>>>> 
>>>> Regards
>>>> Omnia
>>>> 
>>>>> On 24 Apr 2025, at 17:40, Omnia Ibrahim <o....@gmail.com> wrote:
>>>>> 
>>>>> Hi Luke, thanks for having the time to look into the KIP
>>>>>> 2. It's not clear how the "retry count" comes into play in the KIP. It
>>>>>> needs more explanation.
>>>>> My initial thinking is the retry configs are a must for all blob stores,
>>>> so we can provide them, and validate them for free for all blob stores so
>>>> not every implementation will go through verifying them.
>>>>> 
>>>>>> 3. What does "LargeMessageFormatter" do in the process?
>>>>>> I thought all we want to do is to replace the "large value data" into a
>>>>>> path, and consumers will read the path via blob store class.
>>>>>> All these should be done in serializer/deserializer, so why do we need
>>>> the
>>>>>> formatter?
>>>>> 
>>>>> I wanted to bit of more info than just the path to download, for example
>>>> I want to add stuff like the class path for the original blob store for
>>>> example if consumer is setup with the unmatched blob store to the one used
>>>> during publishing.
>>>>> I have updated the KIP to simplify this by having this always as a
>>>> simple json of path and publisher class which can be represented as
>>>> PayloadReferenceValue. WDYT?
>>>>> 
>>>>>> 4. In the BlobStore, it looks like we presume users will use object
>>>> stores,
>>>>>> which is not good.
>>>>>> Could we make it more generic? Javadoc, method names, …
>>>>> This is a good point, I have updated the method names and Javadoc. I
>>>> also thinking of renaming the class name to PayloadStoreinstead of
>>>> BlobStore as blob store still tide to object store as well. To set some
>>>> context here, I am proposing this after working with some community form
>>>> Apache Cassandra who are working on Cassandra CEP <
>>>> https://cwiki.apache.org/confluence/display/CASSANDRA/CEP-44%3A+Kafka+integration+for+Cassandra+CDC+using+Sidecar#CEP44:KafkaintegrationforCassandraCDCusingSidecar-LargeBlob>-44
>>>> to handle large CDC and the initial thinking was let’s publish any large
>>>> cdc to an object store instead of Kafka this why the naming was suggesting
>>>> “blob store” only.
>>>>> 
>>>>>> 5. It would be good to have some explanation for the purpose of each new
>>>>>> interface/class, and clear javadoc for each method.
>>>>> 
>>>>> Updated the KIP with javadocs
>>>>> 
>>>>>> 6. About the BlobIdGenerator, why do we need it, could you explain more?
>>>>>> Again, I thought we only need to replace value to a path, and add the
>>>>>> "large-message" header, so that when consumer reads this message, it'll
>>>>>> read the path from value and get the original data via BlobStore. Why
>>>> do we
>>>>>> need this ID generator? I think users should do the object naming when
>>>>>> putting the object by themselves, not via another interface. WDYT?
>>>>> In some cases generating the ID might need some smart work for example
>>>> to avoid s3 throttling the recommended way on their doc it to create sub
>>>> paths under the original bucket, to decide this we might hash the data to
>>>> find a suitable sub-path.
>>>>> Here is an example of how I would generate an path for s3 file
>>>>> ```
>>>>> public String id(byte[] data) {
>>>>>   String subFolder = topic + "-" +
>>>> Utils.toPositive(Utils.murmur2(data)) % distributionFactor //
>>>> distributionFactor is a config for the Id generator and it represent the
>>>> max number of sub-folders under the bucket
>>>>>   return subFolder + “/“ + UUID.randomUUID().toString()
>>>>> }
>>>>> ```
>>>>> Hope this example clarify a bit. However I do agree here it might not
>>>> need a class. I have move it to be part of the store class.
>>>>> 
>>>>> Please let me know WDYT of the final shape of the KIP now
>>>>> 
>>>>> Thanks
>>>>> Omnia
>>>>> 
>>>>>> On 24 Apr 2025, at 13:31, Luke Chen <sh...@gmail.com> wrote:
>>>>>> 
>>>>>> Hi Omnia,
>>>>>> 
>>>>>> Thanks for proposing this feature that many users expected.
>>>>>> 
>>>>>> Some comments:
>>>>>> 1. It's quite interesting to see the idea of chained
>>>>>> serializer/deserializer used here. I like it.
>>>>>> 
>>>>>> 2. It's not clear how the "retry count" comes into play in the KIP. It
>>>>>> needs more explanation.
>>>>>> 
>>>>>> 3. What does "LargeMessageFormatter" do in the process?
>>>>>> I thought all we want to do is to replace the "large value data" into a
>>>>>> path, and consumers will read the path via blob store class.
>>>>>> All these should be done in serializer/deserializer, so why do we need
>>>> the
>>>>>> formatter?
>>>>>> 
>>>>>> 4. In the BlobStore, it looks like we presume users will use object
>>>> stores,
>>>>>> which is not good.
>>>>>> Could we make it more generic? Javadoc, method names, ...
>>>>>> 
>>>>>> 5. It would be good to have some explanation for the purpose of each new
>>>>>> interface/class, and clear javadoc for each method.
>>>>>> 
>>>>>> 6. About the BlobIdGenerator, why do we need it, could you explain more?
>>>>>> Again, I thought we only need to replace value to a path, and add the
>>>>>> "large-message" header, so that when consumer reads this message, it'll
>>>>>> read the path from value and get the original data via BlobStore. Why
>>>> do we
>>>>>> need this ID generator? I think users should do the object naming when
>>>>>> putting the object by themselves, not via another interface. WDYT?
>>>>>> 
>>>>>> Thanks.
>>>>>> Luke
>>>>>> 
>>>>>> 
>>>>>> 
>>>>>> 
>>>>>> 
>>>>>> On Thu, Apr 10, 2025 at 9:31 PM Omnia Ibrahim <o....@gmail.com>
>>>>>> wrote:
>>>>>> 
>>>>>>> Hi there I would like to start discussions on
>>>>>>> 
>>>> https://cwiki.apache.org/confluence/display/KAFKA/KIP-1159%3A+Large+message+reference+based+Serializer
>>>>>>> 
>>>>>>> Thanks
>>>>>>> Omnia
>> 
> 

Re: [DISCUSS] KIP-1159: Large message reference based Serializer

[Omnia Ibrahim <o....@gmail.com>]

Hi Fede;
> FV1: Looking at PayloadStore and PayloadResponse, "id" and "path"
> seems to be the same thing, which is a reference to the data in the
> external store. If my interpretation is correct, could we converge to
> using one of the two terms?
 
Path usually refers to the full path stored in Kafka topic as a full reference while an id is for file name for example `s3://my-bucket/my-topic/<random_uuid>` or `redis:cluster1:<key_id>`.

> FV2: I think PayloadResponse javadoc or PayloadStore interface may
> need some updates. It says "Response from publish / download", but
> PayloadStore.download returns a byte array, not a PayloadResponse.
Updated the KIP now and removed PayloadResponse and used the PayloadStoreException as this seems to be enough now after all the previous simplification after the last update. 

> FV3: In PayloadStore interface, public and abstract are implicit
> modifiers, so I think we can remove them.
Updated now 

Thanks 
Omnia
> On 26 Jul 2025, at 16:06, Federico Valeri <fe...@gmail.com> wrote:
> 
> Hi Omnia, nice design. I have few questions:
> 
> FV1: Looking at PayloadStore and PayloadResponse, "id" and "path"
> seems to be the same thing, which is a reference to the data in the
> external store. If my interpretation is correct, could we converge to
> using one of the two terms?
> 
> FV2: I think PayloadResponse javadoc or PayloadStore interface may
> need some updates. It says "Response from publish / download", but
> PayloadStore.download returns a byte array, not a PayloadResponse.
> 
> FV3: In PayloadStore interface, public and abstract are implicit
> modifiers, so I think we can remove them.
> 
> Thanks
> Fede
> 
> On Fri, Jul 25, 2025 at 3:28 PM Omnia Ibrahim <o.g.h.ibrahim@gmail.com <ma...@gmail.com>> wrote:
>> 
>> Hi Jun, thanks for having the time to review this
>> 
>>> JR1. While the KIP is potentially useful, I am wondering who is responsible
>>> for retention for the objects in the payload store. Once a message with a
>>> reference is deleted, the key of the external object is lost and the object
>>> may never be deleted.
>> 
>> The `ttl` in the object store is the responsibility of the owner of this store it should be configured in away that is reasonable with the retention config in Kafka.
>> I have updated the KIP with `Consideration` section.
>>> 
>>> JR2. Configs: For all new configs, it would be useful to list their types.
>> Updated the KIP now
>>> 
>>> JR3. value.serializers: Why is this required? If a user doesn't set it, we
>>> should just use value.serializer, right? Ditto for key.serializers.
>> No you right this was copy/past mistake
>> 
>>> 
>>> JR4. For all new public interfaces such as LargeMessageSerializer,
>>> PayloadStore and PayloadResponse, it would be useful to include the full
>>> package name.
>> Updated the KIP now
>>> 
>>> JR5. large.message.payload.store.retry.max.backoff.ms and
>>> large.message.payload.store.retry.delay.backoff.ms: Is the intention to
>>> implement exponential backoff on retries? If so, it's more consistent if we
>>> can follow the existing naming convention like retry.backoff.max.ms <http://retry.backoff.max.ms/> <http://retry.backoff.max.ms/> and
>>> retry.backoff.ms <http://retry.backoff.ms/> <http://retry.backoff.ms/>.
>> I have removed these to simplify the config more (as Luke suggested initially) and added these to the consideration section.
>> 
>>> 
>>> JR6. large.message.skip.not.found.error : If the reference can't be found,
>>> what value does the deserializer return? Note that null has a special
>>> meaning for tombstone in compacted topics.
>> The deserialiser will return `new byte[0]` not null.
>>> 
>>> JR7. PayloadResponse: Why do we have both responseCode and
>>> PayloadStoreException?
>> We can do without responseCode, the initial though was to report response code form payload store.
>> Update the KIP.
>>> JR8. Why do we need PayloadStore.metrics? Note that we could monitor the
>>> metrics in a plugin through the Monitorable interface.
>> Oh nice, I didn’t know about this interface before. Updated the KIP with this now.
>>> 
>>> JR9. Why do we need the protected field PayloadStoreException.isRetryable?
>> Initial thought here was the serializer can retry the upload. But I have removed all the retry logic from serializer and it will be up to the PayloadStore provider to implement this if they need it.
>>> 
>>> JR10. As Luke mentioned earlier, we could turn PayloadStore to an interface.
>> It is updated now to interface.
>> 
>> Hope the last version of the KIP is more simpler now
>> 
>> Thanks
>> Omnia
>> 
>>> On 23 Jul 2025, at 00:43, Jun Rao <jun@confluent.io.INVALID <ma...@confluent.io.INVALID>> wrote:
>>> 
>>> Thanks for the KIP. A few comments.
>>> 
>>> JR1. While the KIP is potentially useful, I am wondering who is responsible
>>> for retention for the objects in the payload store. Once a message with a
>>> reference is deleted, the key of the external object is lost and the object
>>> may never be deleted.
>>> 
>>> JR2. Configs: For all new configs, it would be useful to list their types.
>>> 
>>> JR3. value.serializers: Why is this required? If a user doesn't set it, we
>>> should just use value.serializer, right? Ditto for key.serializers.
>>> 
>>> JR4. For all new public interfaces such as LargeMessageSerializer,
>>> PayloadStore and PayloadResponse, it would be useful to include the full
>>> package name.
>>> 
>>> JR5. large.message.payload.store.retry.max.backoff.ms and
>>> large.message.payload.store.retry.delay.backoff.ms: Is the intention to
>>> implement exponential backoff on retries? If so, it's more consistent if we
>>> can follow the existing naming convention like retry.backoff.max.ms <http://retry.backoff.max.ms/> <http://retry.backoff.max.ms/> and
>>> retry.backoff.ms <http://retry.backoff.ms/> <http://retry.backoff.ms/>.
>>> 
>>> JR6. large.message.skip.not.found.error : If the reference can't be found,
>>> what value does the deserializer return? Note that null has a special
>>> meaning for tombstone in compacted topics.
>>> 
>>> JR7. PayloadResponse: Why do we have both responseCode and
>>> PayloadStoreException?
>>> 
>>> JR8. Why do we need PayloadStore.metrics? Note that we could monitor the
>>> metrics in a plugin through the Monitorable interface.
>>> 
>>> JR9. Why do we need the protected field PayloadStoreException.isRetryable?
>>> 
>>> JR10. As Luke mentioned earlier, we could turn PayloadStore to an interface.
>>> 
>>> Thanks,

Re: [DISCUSS] KIP-1159: Large message reference based Serializer

[Federico Valeri <fe...@gmail.com>]

Hi Omnia, nice design. I have few questions:

FV1: Looking at PayloadStore and PayloadResponse, "id" and "path"
seems to be the same thing, which is a reference to the data in the
external store. If my interpretation is correct, could we converge to
using one of the two terms?

FV2: I think PayloadResponse javadoc or PayloadStore interface may
need some updates. It says "Response from publish / download", but
PayloadStore.download returns a byte array, not a PayloadResponse.

FV3: In PayloadStore interface, public and abstract are implicit
modifiers, so I think we can remove them.

Thanks
Fede

On Fri, Jul 25, 2025 at 3:28 PM Omnia Ibrahim <o....@gmail.com> wrote:
>
> Hi Jun, thanks for having the time to review this
>
> > JR1. While the KIP is potentially useful, I am wondering who is responsible
> > for retention for the objects in the payload store. Once a message with a
> > reference is deleted, the key of the external object is lost and the object
> > may never be deleted.
>
> The `ttl` in the object store is the responsibility of the owner of this store it should be configured in away that is reasonable with the retention config in Kafka.
> I have updated the KIP with `Consideration` section.
> >
> > JR2. Configs: For all new configs, it would be useful to list their types.
> Updated the KIP now
> >
> > JR3. value.serializers: Why is this required? If a user doesn't set it, we
> > should just use value.serializer, right? Ditto for key.serializers.
> No you right this was copy/past mistake
>
> >
> > JR4. For all new public interfaces such as LargeMessageSerializer,
> > PayloadStore and PayloadResponse, it would be useful to include the full
> > package name.
> Updated the KIP now
> >
> > JR5. large.message.payload.store.retry.max.backoff.ms and
> > large.message.payload.store.retry.delay.backoff.ms: Is the intention to
> > implement exponential backoff on retries? If so, it's more consistent if we
> > can follow the existing naming convention like retry.backoff.max.ms <http://retry.backoff.max.ms/> and
> > retry.backoff.ms <http://retry.backoff.ms/>.
> I have removed these to simplify the config more (as Luke suggested initially) and added these to the consideration section.
>
> >
> > JR6. large.message.skip.not.found.error : If the reference can't be found,
> > what value does the deserializer return? Note that null has a special
> > meaning for tombstone in compacted topics.
> The deserialiser will return `new byte[0]` not null.
> >
> > JR7. PayloadResponse: Why do we have both responseCode and
> > PayloadStoreException?
> We can do without responseCode, the initial though was to report response code form payload store.
> Update the KIP.
> > JR8. Why do we need PayloadStore.metrics? Note that we could monitor the
> > metrics in a plugin through the Monitorable interface.
> Oh nice, I didn’t know about this interface before. Updated the KIP with this now.
> >
> > JR9. Why do we need the protected field PayloadStoreException.isRetryable?
> Initial thought here was the serializer can retry the upload. But I have removed all the retry logic from serializer and it will be up to the PayloadStore provider to implement this if they need it.
> >
> > JR10. As Luke mentioned earlier, we could turn PayloadStore to an interface.
> It is updated now to interface.
>
> Hope the last version of the KIP is more simpler now
>
> Thanks
> Omnia
>
> > On 23 Jul 2025, at 00:43, Jun Rao <ju...@confluent.io.INVALID> wrote:
> >
> > Thanks for the KIP. A few comments.
> >
> > JR1. While the KIP is potentially useful, I am wondering who is responsible
> > for retention for the objects in the payload store. Once a message with a
> > reference is deleted, the key of the external object is lost and the object
> > may never be deleted.
> >
> > JR2. Configs: For all new configs, it would be useful to list their types.
> >
> > JR3. value.serializers: Why is this required? If a user doesn't set it, we
> > should just use value.serializer, right? Ditto for key.serializers.
> >
> > JR4. For all new public interfaces such as LargeMessageSerializer,
> > PayloadStore and PayloadResponse, it would be useful to include the full
> > package name.
> >
> > JR5. large.message.payload.store.retry.max.backoff.ms and
> > large.message.payload.store.retry.delay.backoff.ms: Is the intention to
> > implement exponential backoff on retries? If so, it's more consistent if we
> > can follow the existing naming convention like retry.backoff.max.ms <http://retry.backoff.max.ms/> and
> > retry.backoff.ms <http://retry.backoff.ms/>.
> >
> > JR6. large.message.skip.not.found.error : If the reference can't be found,
> > what value does the deserializer return? Note that null has a special
> > meaning for tombstone in compacted topics.
> >
> > JR7. PayloadResponse: Why do we have both responseCode and
> > PayloadStoreException?
> >
> > JR8. Why do we need PayloadStore.metrics? Note that we could monitor the
> > metrics in a plugin through the Monitorable interface.
> >
> > JR9. Why do we need the protected field PayloadStoreException.isRetryable?
> >
> > JR10. As Luke mentioned earlier, we could turn PayloadStore to an interface.
> >
> > Thanks,
>

Re: [DISCUSS] KIP-1159: Large message reference based Serializer

[Omnia Ibrahim <o....@gmail.com>]

Hi Greg, sorry for late respond as I was away for few days 

> Thanks for pointing this out, I agree that KIP-1163 does not address these
> multi-tenant concerns. I brought up KIP-1163 because I thought it would
> solve some but not all of the large-message concerns.
> However, the technical problems you mention could be addressed under the
> umbrella of "improving Kafka's handling of large messages in-band".
> Investments in resource isolation/cell based architecture tooling/memory
> management will be broadly beneficial to Kafka, improving both
> small-message and large-message workloads.

I do agree that investing in better compute resource isolation would be beneficial for Kafka. However, full isolation in Kafka compute resources isn't an easy task and I don't believe anyone in the community has a complete view of how to achieve this at the moment. In the meanwhile, why not have this serializer until we develop a more sophisticated model for compute isolation?
We have examples of these kinds of interim solutions before - for example, having MM1/MM2 instead of built-in cross-cluster replication

> Could you please add some details for these other use-cases? We already
> have multi-message compression in-protocol (which is often better than
> single-message compression).
> We don't have per-message encryption, but this is already addressed by
> Kroxylicious [1]. I'm not aware of any other binary-to-binary operations
> that would be relevant here.

Beyond large messages, composable serializers address the pattern we see where organizations build custom serialization chains in-house. The specific combinations vary by use case, but the common need is to apply multiple transformations (schema, compression, encryption, audit logging) without creating combinatorial explosion of serializer classes. 
The reference-based messaging pattern itself is the primary driver - composable serializers just provide a clean way to integrate it with existing serialization workflows.

> 
> I also don't know of any organization willing to offer this as external OSS
> using the existing interfaces. And that makes me seriously doubt there's an
> organization willing to offer this as external OSS using
> yet-unmerged-interfaces that duplicate the existing interfaces.

You raise an interesting point about implementation commitments. Looking at Client Interceptors, for example, they were accepted based on interface value, with implementations emerging afterward.

The principle has been that Kafka provides standardized interfaces, and the ecosystem develops implementations. However, I understand the concern about ensuring viability. As this work is linked to Cassandra community (CEP-44) I would expect interface implementation for S3/cloud storage implementations to emerge as part of their CDC integration work. This provides a concrete starting point for the ecosystem.

> AFAIU, there is only one realistic open-source implementation of tiered
> storage, and it's maintained by Aiven. [2]
> For Diskless, Aiven will also maintain an open-source implementation of any
> pluggable interfaces required to make the feature functional.
> 
> Does this proposal also require Aiven to take on the responsibility of
> implementing reference based messaging plugins? If Aiven declines to
> maintain an implementation, is the interface actually beneficial to the
> project?

I'd be curious to understand if there's been a shift in policy regarding implementation commitments for new interfaces, as this would be valuable context for future KIPs.

Regards 
Omnia

> On 19 Aug 2025, at 18:46, Greg Harris <gr...@aiven.io.INVALID> wrote:
> 
> Hi Omnia,
> 
> Thanks for your replies.
> 
> 1.
>> There are other concerns like amplifying all Kafka resources like network
> bandwidth, memory pressure, and I/O operations becoming more unpredictable.
> 
> Thanks for pointing this out, I agree that KIP-1163 does not address these
> multi-tenant concerns. I brought up KIP-1163 because I thought it would
> solve some but not all of the large-message concerns.
> However, the technical problems you mention could be addressed under the
> umbrella of "improving Kafka's handling of large messages in-band".
> Investments in resource isolation/cell based architecture tooling/memory
> management will be broadly beneficial to Kafka, improving both
> small-message and large-message workloads.
> 
> 2.
>> I would prefer this to be in Kafka anyway at least the composable
> serialization - this is beneficial for multiple use-cases and doesn't need
> to be outside of Kafka it should be standard feature
> 
> Could you please add some details for these other use-cases? We already
> have multi-message compression in-protocol (which is often better than
> single-message compression).
> We don't have per-message encryption, but this is already addressed by
> Kroxylicious [1]. I'm not aware of any other binary-to-binary operations
> that would be relevant here.
> 
> 3. On the subject of Kroxylicious, could another rejected alternative be to
> implement reference based messaging as a filter for Kroxylicious?
> 
> 4.
>> I am personally not aware of any one willing to offer this as external
> OSS.
> 
> I also don't know of any organization willing to offer this as external OSS
> using the existing interfaces. And that makes me seriously doubt there's an
> organization willing to offer this as external OSS using
> yet-unmerged-interfaces that duplicate the existing interfaces.
> 
> 5.
>> Also am maybe missing something here but how this is different than
> implementing interface to something like TS or Diskless?
> 
> AFAIU, there is only one realistic open-source implementation of tiered
> storage, and it's maintained by Aiven. [2]
> For Diskless, Aiven will also maintain an open-source implementation of any
> pluggable interfaces required to make the feature functional.
> 
> Does this proposal also require Aiven to take on the responsibility of
> implementing reference based messaging plugins? If Aiven declines to
> maintain an implementation, is the interface actually beneficial to the
> project?
> 
> Thanks,
> Greg
> 
> [1] https://kroxylicious.io/
> [2] https://github.com/Aiven-Open/tiered-storage-for-apache-kafka
> 
> On Tue, Aug 19, 2025 at 9:08 AM Omnia Ibrahim <o.g.h.ibrahim@gmail.com <ma...@gmail.com>>
> wrote:
> 
>> Hi Greg,
>> Thanks for reviewing the KIP!
>> 
>>>> Could you discuss the rejected alternative of improving Kafka's
>> handling of
>>>> large messages in-band? If Kafka is better at handling large messages,
>> then
>>>> it may not be necessary to adopt the complexity of reference-based
>>>> messaging.
>>>> I wonder if these problems truly disappear with referenced based
>> messaging,
>>>> or if they are moved from broker to applications, where the Apache
>> Kafka
>>>> project isn't blamed but also has less ability to address. Client
>>>> applications may have worse memory management or observability than
>> Kafka
>>>> brokers.
>>>> For example, I think that KIP-1163 may benefit large-message workloads,
>>>> because it allows for the time-to-commit a batch to be decoupled from
>> the
>>>> time-to-write the batch.
>> 
>> Regarding the mention of KIP-1163 benefiting large message or Kafka better
>> dealing with large message I would say this would be tricky if cluster is
>> operated/served as multi-tenant.
>> In shared Kafka clusters, large messages from one application can degrade
>> performance for all other topics and clients. This violates the isolation
>> principle that makes Kafka suitable for multi-tenant environments.
>> 
>> While improvements like KIP-1163 help with publishing data into external
>> storage, they don't address the fundamental resource contention issues.
>> There are other concerns like amplifying all Kafka resources like network
>> bandwidth, memory pressure, and I/O operations becoming more unpredictable.
>> 
>> 
>> The client-side approach provides better isolation where storage failures,
>> or payload’s size don't impact Kafka cluster availability and performance
>> for all use-cases, applications can choose appropriate storage for their
>> specific use case which would be complicated to coordinate at the broker
>> level on multi-tenant where each topic needs different external storage.
>> 
>>>> Could there be another rejected alternative to open-source an existing
>>>> reference-based-messaging solution? If there are community members with
>>>> awareness of an existing custom solution, they could also influence
>>>> open-sourcing it.
>> 
>> I am personally not aware of any one willing to offer this as external
>> OSS. I would prefer this to be in Kafka anyway at least the composable
>> serialization - this is beneficial for multiple use-cases and doesn't need
>> to be outside of Kafka it should be standard feature.
>> 
>> 
>>>> AFAIU this proposal still requires implementations of the new
>> interfaces to
>>>> be in a separate project outside of Apache Kafka, because we cannot
>> depend
>>>> on object storage implementations in Apache Kafka.
>>>> Such a project could own the new interfaces itself, either as private
>>>> interfaces or public API it exposes to further downstream projects.
>> This
>>>> will have fewer restrictions on the API design, as a vote on Apache
>> Kafka
>>>> is not necessary to evolve the functionality.
>> 
>> Yes, as we have similar interface in Kafka for example storage layer in
>> Tiered Storage and client interceptors where everyone can point to their
>> own implementation. These examples started with Kafka providing
>> standardized interfaces then different implementations were developed by
>> the community later.
>> 
>>>> Can you clarify what the benefit is for this change to land in Apache
>> Kafka
>>>> specifically? And if there is a group ready to develop and distribute
>>>> open-source implementations of this interface? Otherwise we will end up
>>>> back where we started with only closed-source implementations.
>> 
>> 
>> The "closed-source implementations” on storage layer is not the issue the
>> pattern of applying schemas or different serializers then uploading the
>> payload is the pattern that is everyone doing behind closed doors. Also am
>> maybe missing something here but how this is different than implementing
>> interface to something like TS or Diskless?
>> 
>> 
>> I’ll update the KIP with some of these points.
>> 
>> Thanks
>> Omnia
>> 
>>> On 13 Aug 2025, at 17:28, Greg Harris <gr...@aiven.io.INVALID>
>> wrote:
>>> 
>>> Hi Omnia,
>>> 
>>> Thanks for the KIP!
>>> 
>>> I had some questions about the rejected alternatives:
>>> 
>>> 1. The motivation for the KIP includes this:
>>>> Increasing message size indefinitely is not a viable solution as it can
>>> lead to performance degradation, memory issues, and instability of the
>>> broker.
>>> 
>>> Could you discuss the rejected alternative of improving Kafka's handling
>> of
>>> large messages in-band? If Kafka is better at handling large messages,
>> then
>>> it may not be necessary to adopt the complexity of reference-based
>>> messaging.
>>> I wonder if these problems truly disappear with referenced based
>> messaging,
>>> or if they are moved from broker to applications, where the Apache Kafka
>>> project isn't blamed but also has less ability to address. Client
>>> applications may have worse memory management or observability than Kafka
>>> brokers.
>>> For example, I think that KIP-1163 may benefit large-message workloads,
>>> because it allows for the time-to-commit a batch to be decoupled from the
>>> time-to-write the batch.
>>> 
>>> 2. The "popular patterns" section includes this:
>>>> Available open-source implementation
>>>> There isn't a specific open-source implementation for Reference-Based
>>> Messaging. Mostly everyone adopting this has their custom solution
>>> 
>>> Could there be another rejected alternative to open-source an existing
>>> reference-based-messaging solution? If there are community members with
>>> awareness of an existing custom solution, they could also influence
>>> open-sourcing it.
>>> 
>>> 3. This is listed as a rejected alternative:
>>>> Implement this as a separate project outside of Apache Kafka as this
>>> seems to be a pattern that needs more use cases and it would be better to
>>> have this in Apache Kafka as native implementation instead of a separate
>>> project.
>>> 
>>> AFAIU this proposal still requires implementations of the new interfaces
>> to
>>> be in a separate project outside of Apache Kafka, because we cannot
>> depend
>>> on object storage implementations in Apache Kafka.
>>> Such a project could own the new interfaces itself, either as private
>>> interfaces or public API it exposes to further downstream projects. This
>>> will have fewer restrictions on the API design, as a vote on Apache Kafka
>>> is not necessary to evolve the functionality.
>>> 
>>> Can you clarify what the benefit is for this change to land in Apache
>> Kafka
>>> specifically? And if there is a group ready to develop and distribute
>>> open-source implementations of this interface? Otherwise we will end up
>>> back where we started with only closed-source implementations.
>>> 
>>> Thanks,
>>> Greg
>>> 
>>> On Wed, Aug 13, 2025 at 5:27 AM Omnia Ibrahim <o.g.h.ibrahim@gmail.com
>> <ma...@gmail.com>>
>>> wrote:
>>> 
>>>> Hi Jun
>>>>> 
>>>>> JR20. The inconsistency is still there. It's probably simpler to
>>>>> consolidate on
>>>>> org.apache.kafka.common.serialization.largemessage.PayloadStore.
>>>> Updated the kip now to address these
>>>>> JR21. Could you document how the payload id string is encoded? For
>>>> example,
>>>>> is it always UTF8?
>>>> I updated the java doc, the default implementation do provide UTF-8
>>>> however if the developer want to support any other they will need to
>>>> address this in the implementation of the store.
>>>>> 
>>>>> JR22. "Since compacted topics can retain data indefinitely, users must
>>>>> choose between setting a business-appropriate TTL or accepting
>> indefinite
>>>>> storage costs." Setting a TTL is not ideal since it can break the
>>>> consumer
>>>>> application. So, we probably don't want to recommend it to the users.
>>>>> Another option is to run a cleaning job by consuming the topic and if
>> it
>>>>> observes a null payload, delete the object based on the key in the
>>>> record.
>>>> 
>>>> Updated the recommendations
>>>>> JR24. In the following,
>> org.apache.kafka.common.serialization.Serializer
>>>>> should include the largemessage namespace.
>>>>> producerConfig.put("value.serializers",
>>>>> "kafka.serializers.KafkaAvroDeserializer,
>>>>> org.apache.kafka.common.serialization.Serializer");
>>>> 
>>>> Updated now
>>>> 
>>>> Thanks
>>>> Omnia
>>>> 
>>>>> On 12 Aug 2025, at 18:09, Jun Rao <ju...@confluent.io.INVALID> wrote:
>>>>> 
>>>>> Hi, Omnia,
>>>>> 
>>>>> Thanks for the reply. A few more comments.
>>>>> 
>>>>> JR20. The inconsistency is still there. It's probably simpler to
>>>>> consolidate on
>>>>> org.apache.kafka.common.serialization.largemessage.PayloadStore.
>>>>> 
>>>>> JR21. Could you document how the payload id string is encoded? For
>>>> example,
>>>>> is it always UTF8?
>>>>> 
>>>>> JR22. "Since compacted topics can retain data indefinitely, users must
>>>>> choose between setting a business-appropriate TTL or accepting
>> indefinite
>>>>> storage costs." Setting a TTL is not ideal since it can break the
>>>> consumer
>>>>> application. So, we probably don't want to recommend it to the users.
>>>>> Another option is to run a cleaning job by consuming the topic and if
>> it
>>>>> observes a null payload, delete the object based on the key in the
>>>> record.
>>>>> 
>>>>> JR23. We probably don't need to force the PayloadStore implementer to
>>>>> implement Monitorable. We could document that if it's desirable to
>>>> monitor
>>>>> a PayloadStore plugin, one can implement the Monitorable interface.
>>>>> 
>>>>> JR24. In the following,
>> org.apache.kafka.common.serialization.Serializer
>>>>> should include the largemessage namespace.
>>>>> producerConfig.put("value.serializers",
>>>>> "kafka.serializers.KafkaAvroDeserializer,
>>>>> org.apache.kafka.common.serialization.Serializer");
>>>>> 
>>>>> Jun
>>>>> 
>>>>> On Tue, Aug 12, 2025 at 3:27 AM Omnia Ibrahim <o.g.h.ibrahim@gmail.com
>>>> <ma...@gmail.com>>
>>>>> wrote:
>>>>> 
>>>>>> Hi Jun,
>>>>>> 
>>>>>>> JR20. Could you fix the inconsistency?
>>>>>>> org.apache.kafka.common.serialization.largemessage.PayloadStore vs
>>>>>>> org.apache.kafka.common.serialization.largemessage.store.PayloadStore
>>>>>> Updated
>>>>>>> 
>>>>>>> JR21. "Encapsulate that ID into a simple Kafka event using a
>> structured
>>>>>>> format.". Could you define the structured format explicitly?
>>>>>> 
>>>>>> This was from the initial design we had Id format but after discussing
>>>>>> this with Luke I ended up simplifying this by removing the format and
>> I
>>>>>> guess I forgot this line. I have updated the kip now.
>>>>>> 
>>>>>>> JR22. TTL for object stores: Could the new serializer be used on a
>>>>>>> compacted topic? If so, how should a user configure the TTL since the
>>>>>> Kafka
>>>>>>> retention is infinite.
>>>>>> 
>>>>>> Great question about compacted topics! There are few usage of
>> compacted
>>>>>> topic to consider:
>>>>>> 1. Typical Compacted Topic Usage: The compacted topics are commonly
>> used
>>>>>> as cache/state layers where the latest value for each key represents
>> the
>>>>>> current state. These use cases typically involve smaller records since
>>>> they
>>>>>> need to be loaded efficiently for state reconstruction. Large message
>>>>>> serialization would be unusual for this pattern.
>>>>>> 
>>>>>> 2. User-Defined Compacted Topics with Large Payloads: for applications
>>>>>> that do use compacted topics with large payloads, the PayloadStore
>>>>>> implementation should handle this by:
>>>>>>      a. Consistent ID Generation: Use deterministic IDs based on the
>>>>>> Kafka key (rather than random UUIDs) so that when a payload id is
>>>> updated,
>>>>>> it overwrites the same payload store object instead of creating new
>>>> ones.
>>>>>> 
>>>>>>      b. TTL Strategy: Since compacted topics can retain data
>>>>>> indefinitely, users have two options:
>>>>>>              - Set a business-appropriate TTL (basically "we know our
>>>>>> cache data becomes stale after 30 days”)
>>>>>>              - Configure no TTL and accept indefinite storage costs
>> as
>>>>>> a trade-off for the architectural benefits
>>>>>> 
>>>>>> 3. Compact + Delete Policy: Topics with
>> `cleanup.policy=compact,delete`
>>>>>> will eventually remove old data, so standard TTL approaches work
>>>> normally.
>>>>>> The key insight is that TTL configuration depends on the business
>>>>>> requirements and usage patterns rather than just the Kafka retention
>>>>>> policy. The PayloadStore implementation should provide flexibility for
>>>>>> users to make this trade-off consciously.
>>>>>> 
>>>>>> I have update the consideration section with this.
>>>>>> 
>>>>>> Thanks
>>>>>> Omnia
>>>>>> 
>>>>>>> On 11 Aug 2025, at 23:32, Jun Rao <ju...@confluent.io.INVALID> wrote:
>>>>>>> 
>>>>>>> Hi, Omnia,
>>>>>>> 
>>>>>>> Thanks for the reply. A few more comments.
>>>>>>> 
>>>>>>> JR20. Could you fix the inconsistency?
>>>>>>> org.apache.kafka.common.serialization.largemessage.PayloadStore vs
>>>>>>> org.apache.kafka.common.serialization.largemessage.store.PayloadStore
>>>>>>> 
>>>>>>> JR21. "Encapsulate that ID into a simple Kafka event using a
>> structured
>>>>>>> format.". Could you define the structured format explicitly?
>>>>>>> 
>>>>>>> JR22. TTL for object stores: Could the new serializer be used on a
>>>>>>> compacted topic? If so, how should a user configure the TTL since the
>>>>>> Kafka
>>>>>>> retention is infinite.
>>>>>>> 
>>>>>>> Jun
>>>>>>> 
>>>>>>> On Mon, Aug 11, 2025 at 10:08 AM Omnia Ibrahim <
>>>> o.g.h.ibrahim@gmail.com
>>>>>> <ma...@gmail.com>>
>>>>>>> wrote:
>>>>>>> 
>>>>>>>> Hi Jun
>>>>>>>>> JR11. Do you think that we need to add key.serializers and
>>>>>>>>> key.deserializers or do you think covering large messages in value
>> is
>>>>>>>>> enough?
>>>>>>>> The large message issue usually is a value issue, I never saw a key
>>>> that
>>>>>>>> is bigger than broker message size. If I would add key.serializers
>> and
>>>>>>>> key.deserializers it would be for consistency and maybe there are
>>>>>> use-cases
>>>>>>>> where developers want to apply multiple serializations in order on
>> the
>>>>>> key
>>>>>>>> as well outside the context of the large message support.
>>>>>>>> I updated the KIP to add these two configs now.
>>>>>>>> 
>>>>>>>>> JR12. We can load ComposableSerializer automatically if
>>>>>> value.serializers
>>>>>>>>> or value.deserializers are specified. But, it seems that we could
>>>> keep
>>>>>>>>> ComposableSerializer as an internal implementation. For example,
>>>>>>>>> ProducerInterceptors is automatically loaded when multiple
>>>> interceptors
>>>>>>>> are
>>>>>>>>> specified and is an internal class.
>>>>>>>> I updated the kip to highlight that the changes is updating
>>>>>> ProducerConfig
>>>>>>>> and ConsumerConfig to have new config for serializers/deserializers
>>>> and
>>>>>> not
>>>>>>>> the actual class this is implementation details and not public
>>>>>> interfaces.
>>>>>>>> 
>>>>>>>>> JR17. We could estimate the size after compression, but the
>> estimator
>>>>>> is
>>>>>>>>> not 100% accurate. It seems that it's simpler to just use the
>>>> original
>>>>>>>>> message size.
>>>>>>>> We can keep it as original message size, I was thinking if it is
>> good
>>>>>>>> enough for max.request.size it might be good enough for this. I
>>>> updated
>>>>>> the
>>>>>>>> KIP anyway to simplify it and keep it to the original size.
>>>>>>>> 
>>>>>>>> Hope the final version addressed all feedbacks and we can resume
>> with
>>>>>> the
>>>>>>>> voting
>>>>>>>> 
>>>>>>>> Thanks
>>>>>>>> Omnia
>>>>>>>> 
>>>>>>>>> On 8 Aug 2025, at 22:10, Jun Rao <ju...@confluent.io.INVALID> wrote:
>>>>>>>>> 
>>>>>>>>> Hi, Omnia,
>>>>>>>>> 
>>>>>>>>> Thanks for the reply.
>>>>>>>>> 
>>>>>>>>> JR11. Do you think that we need to add key.serializers and
>>>>>>>>> key.deserializers or do you think covering large messages in value
>> is
>>>>>>>>> enough?
>>>>>>>>> 
>>>>>>>>> JR12. We can load ComposableSerializer automatically if
>>>>>> value.serializers
>>>>>>>>> or value.deserializers are specified. But, it seems that we could
>>>> keep
>>>>>>>>> ComposableSerializer as an internal implementation. For example,
>>>>>>>>> ProducerInterceptors is automatically loaded when multiple
>>>> interceptors
>>>>>>>> are
>>>>>>>>> specified and is an internal class.
>>>>>>>>> 
>>>>>>>>> JR17. We could estimate the size after compression, but the
>> estimator
>>>>>> is
>>>>>>>>> not 100% accurate. It seems that it's simpler to just use the
>>>> original
>>>>>>>>> message size.
>>>>>>>>> 
>>>>>>>>> Jun
>>>>>>>>> 
>>>>>>>>> On Mon, Aug 4, 2025 at 10:33 AM Omnia Ibrahim <
>>>> o.g.h.ibrahim@gmail.com
>>>>>>>> <ma...@gmail.com>>
>>>>>>>>> wrote:
>>>>>>>>> 
>>>>>>>>>> 
>>>>>>>>>> Hi Jun
>>>>>>>>>>> JR11. value.serializers and value.deserializers: Should they be
>> of
>>>>>> type
>>>>>>>>>>> List? Also, where are key.serializers and key.deserializers?
>>>>>>>>>>> 
>>>>>>>>>> Updated now
>>>>>>>>>>> JR12. Do we still need ComposableSerializer and
>>>>>> ComposableDeserializer?
>>>>>>>>>> The initial thinking here was if `value.serializers or
>>>>>>>> value.deserializers
>>>>>>>>>> ` exists the client will load  ComposableSerializer or
>>>>>>>>>> ComposableDeserializer  automatically and use them. Unfortunately
>>>> this
>>>>>>>>>> would need us to define these serializers.
>>>>>>>>>> 
>>>>>>>>>> The other option is to update `Plugin<Serializer<V>>
>>>>>>>>>> valueSerializerPlugin` and KafkaProducer constructor to accept
>>>>>>>>>> List<Serializer<V>> and move the logic of the ComposableSerializer
>>>>>> into
>>>>>>>>>> KafkaProducer::doSend which when we do serialization (same with
>>>>>>>>>> KafkaConsumer). This option hide the logic and reduce exposure for
>>>>>>>> client
>>>>>>>>>> to these.
>>>>>>>>>> WDYT?
>>>>>>>>>> 
>>>>>>>>>>> 
>>>>>>>>>>> JR13. large.message.payload.store.class : should it be of type
>>>> class?
>>>>>>>>>> Updated
>>>>>>>>>>> 
>>>>>>>>>>> JR14.
>>>>>>>>>>> 
>>>>>>>>>> 
>>>>>>>> 
>>>>>> 
>>>> 
>> org.apache.kafka.common.serialization.largemessage.LargeMessageSerializer :
>>>>>>>>>>> The name seems redundant since largemessage appears twice.
>>>>>>>>>> Updated
>>>>>>>>>>> 
>>>>>>>>>>> JR15. PayloadResponse: It still mentions response code. It
>> mentions
>>>>>>>>>>> "isRetryable flag", which no longer exists in
>>>> PayloadStoreException.
>>>>>>>>>> There
>>>>>>>>>>> are typos in "then it will serialiser will”.
>>>>>>>>>> The KIP is updated now
>>>>>>>>>> 
>>>>>>>>>>> JR16. Regarding returning new byte[0] if
>>>>>>>>>> large.message.skip.not.found.error
>>>>>>>>>>> is true, this will likely fail the next deserializer and the
>>>>>>>> application
>>>>>>>>>>> won't have the right context of the error. It's probably better
>> to
>>>>>> just
>>>>>>>>>>> propagate the specific exception and let the caller handle it.
>>>>>>>>>> You right this will cause issue if there is another deserializer
>>>>>> waiting
>>>>>>>>>> for the data. I have updated the KIP with this.
>>>>>>>>>> 
>>>>>>>>>>> 
>>>>>>>>>>> JR17. LargeMessageSerializer:  "Check if the estimated size of
>> the
>>>>>> data
>>>>>>>>>>> (bytes) after applying provided compression (if there is one)"
>>>>>>>>>> Compression
>>>>>>>>>>> actually happens after serialization and is done on a batch of
>>>>>> records.
>>>>>>>>>> Yes the compression itself happened after but we also have
>>>>>>>>>> `estimateSizeInBytesUpperBound` which my understanding is this
>> take
>>>>>> the
>>>>>>>>>> compression type into the account as well
>>>>>>>>>>> 
>>>>>>>>>>> 
>>>>>>>>>>> JR18. Could you define the type T for LargeMessageSerializer?
>>>>>>>>>> Update the KIP this T would be byte[]
>>>>>>>>>> 
>>>>>>>>>> Thanks
>>>>>>>>>> Omnia
>>>>>>>>>>> 
>>>>>>>>>>> Jun
>>>>>>>>>>> 
>>>>>>>>>>> On Fri, Jul 25, 2025 at 6:28 AM Omnia Ibrahim <
>>>>>> o.g.h.ibrahim@gmail.com
>>>>>>>>>> <ma...@gmail.com>>
>>>>>>>>>>> wrote:
>>>>>>>>>>> 
>>>>>>>>>>>> Hi Jun, thanks for having the time to review this
>>>>>>>>>>>> 
>>>>>>>>>>>>> JR1. While the KIP is potentially useful, I am wondering who is
>>>>>>>>>>>> responsible
>>>>>>>>>>>>> for retention for the objects in the payload store. Once a
>>>> message
>>>>>>>>>> with a
>>>>>>>>>>>>> reference is deleted, the key of the external object is lost
>> and
>>>>>> the
>>>>>>>>>>>> object
>>>>>>>>>>>>> may never be deleted.
>>>>>>>>>>>> 
>>>>>>>>>>>> The `ttl` in the object store is the responsibility of the owner
>>>> of
>>>>>>>> this
>>>>>>>>>>>> store it should be configured in away that is reasonable with
>> the
>>>>>>>>>> retention
>>>>>>>>>>>> config in Kafka.
>>>>>>>>>>>> I have updated the KIP with `Consideration` section.
>>>>>>>>>>>>> 
>>>>>>>>>>>>> JR2. Configs: For all new configs, it would be useful to list
>>>> their
>>>>>>>>>>>> types.
>>>>>>>>>>>> Updated the KIP now
>>>>>>>>>>>>> 
>>>>>>>>>>>>> JR3. value.serializers: Why is this required? If a user doesn't
>>>> set
>>>>>>>> it,
>>>>>>>>>>>> we
>>>>>>>>>>>>> should just use value.serializer, right? Ditto for
>>>> key.serializers.
>>>>>>>>>>>> No you right this was copy/past mistake
>>>>>>>>>>>> 
>>>>>>>>>>>>> 
>>>>>>>>>>>>> JR4. For all new public interfaces such as
>>>> LargeMessageSerializer,
>>>>>>>>>>>>> PayloadStore and PayloadResponse, it would be useful to include
>>>> the
>>>>>>>>>> full
>>>>>>>>>>>>> package name.
>>>>>>>>>>>> Updated the KIP now
>>>>>>>>>>>>> 
>>>>>>>>>>>>> JR5. large.message.payload.store.retry.max.backoff.ms and
>>>>>>>>>>>>> large.message.payload.store.retry.delay.backoff.ms: Is the
>>>>>> intention
>>>>>>>>>> to
>>>>>>>>>>>>> implement exponential backoff on retries? If so, it's more
>>>>>> consistent
>>>>>>>>>> if
>>>>>>>>>>>> we
>>>>>>>>>>>>> can follow the existing naming convention like
>>>>>> retry.backoff.max.ms
>>>>>>>> <
>>>>>>>>>>>> http://retry.backoff.max.ms/> and
>>>>>>>>>>>>> retry.backoff.ms <http://retry.backoff.ms/> <
>>>>>>>> http://retry.backoff.ms/> <http://retry.backoff.ms/
>>>>>>>>>>> .
>>>>>>>>>>>> I have removed these to simplify the config more (as Luke
>>>> suggested
>>>>>>>>>>>> initially) and added these to the consideration section.
>>>>>>>>>>>> 
>>>>>>>>>>>>> 
>>>>>>>>>>>>> JR6. large.message.skip.not.found.error : If the reference
>> can't
>>>> be
>>>>>>>>>>>> found,
>>>>>>>>>>>>> what value does the deserializer return? Note that null has a
>>>>>> special
>>>>>>>>>>>>> meaning for tombstone in compacted topics.
>>>>>>>>>>>> The deserialiser will return `new byte[0]` not null.
>>>>>>>>>>>>> 
>>>>>>>>>>>>> JR7. PayloadResponse: Why do we have both responseCode and
>>>>>>>>>>>>> PayloadStoreException?
>>>>>>>>>>>> We can do without responseCode, the initial though was to report
>>>>>>>>>> response
>>>>>>>>>>>> code form payload store.
>>>>>>>>>>>> Update the KIP.
>>>>>>>>>>>>> JR8. Why do we need PayloadStore.metrics? Note that we could
>>>>>> monitor
>>>>>>>>>> the
>>>>>>>>>>>>> metrics in a plugin through the Monitorable interface.
>>>>>>>>>>>> Oh nice, I didn’t know about this interface before. Updated the
>>>> KIP
>>>>>>>> with
>>>>>>>>>>>> this now.
>>>>>>>>>>>>> 
>>>>>>>>>>>>> JR9. Why do we need the protected field
>>>>>>>>>>>> PayloadStoreException.isRetryable?
>>>>>>>>>>>> Initial thought here was the serializer can retry the upload.
>> But
>>>> I
>>>>>>>> have
>>>>>>>>>>>> removed all the retry logic from serializer and it will be up to
>>>> the
>>>>>>>>>>>> PayloadStore provider to implement this if they need it.
>>>>>>>>>>>>> 
>>>>>>>>>>>>> JR10. As Luke mentioned earlier, we could turn PayloadStore to
>> an
>>>>>>>>>>>> interface.
>>>>>>>>>>>> It is updated now to interface.
>>>>>>>>>>>> 
>>>>>>>>>>>> Hope the last version of the KIP is more simpler now
>>>>>>>>>>>> 
>>>>>>>>>>>> Thanks
>>>>>>>>>>>> Omnia
>>>>>>>>>>>> 
>>>>>>>>>>>>> On 23 Jul 2025, at 00:43, Jun Rao <jun@confluent.io.INVALID
>>>>>> <ma...@confluent.io.INVALID> <mailto:
>>>>>>>> jun@confluent.io.INVALID <ma...@confluent.io.INVALID> <mailto:
>>>> jun@confluent.io.INVALID>> <mailto:
>>>>>>>>>> jun@confluent.io.INVALID <ma...@confluent.io.INVALID>
>> <ma...@confluent.io.INVALID> <mailto:
>>>> jun@confluent.io.INVALID <ma...@confluent.io.INVALID> <ma...@confluent.io.INVALID>> <mailto:
>>>>>> jun@confluent.io.INVALID <ma...@confluent.io.INVALID> <ma...@confluent.io.INVALID> <mailto:
>> jun@confluent.io.INVALID <ma...@confluent.io.INVALID>>>>> wrote:
>>>>>>>>>>>>> 
>>>>>>>>>>>>> Thanks for the KIP. A few comments.
>>>>>>>>>>>>> 
>>>>>>>>>>>>> JR1. While the KIP is potentially useful, I am wondering who is
>>>>>>>>>>>> responsible
>>>>>>>>>>>>> for retention for the objects in the payload store. Once a
>>>> message
>>>>>>>>>> with a
>>>>>>>>>>>>> reference is deleted, the key of the external object is lost
>> and
>>>>>> the
>>>>>>>>>>>> object
>>>>>>>>>>>>> may never be deleted.
>>>>>>>>>>>>> 
>>>>>>>>>>>>> JR2. Configs: For all new configs, it would be useful to list
>>>> their
>>>>>>>>>>>> types.
>>>>>>>>>>>>> 
>>>>>>>>>>>>> JR3. value.serializers: Why is this required? If a user doesn't
>>>> set
>>>>>>>> it,
>>>>>>>>>>>> we
>>>>>>>>>>>>> should just use value.serializer, right? Ditto for
>>>> key.serializers.
>>>>>>>>>>>>> 
>>>>>>>>>>>>> JR4. For all new public interfaces such as
>>>> LargeMessageSerializer,
>>>>>>>>>>>>> PayloadStore and PayloadResponse, it would be useful to include
>>>> the
>>>>>>>>>> full
>>>>>>>>>>>>> package name.
>>>>>>>>>>>>> 
>>>>>>>>>>>>> JR5. large.message.payload.store.retry.max.backoff.ms and
>>>>>>>>>>>>> large.message.payload.store.retry.delay.backoff.ms: Is the
>>>>>> intention
>>>>>>>>>> to
>>>>>>>>>>>>> implement exponential backoff on retries? If so, it's more
>>>>>> consistent
>>>>>>>>>> if
>>>>>>>>>>>> we
>>>>>>>>>>>>> can follow the existing naming convention like
>>>>>> retry.backoff.max.ms <http://retry.backoff.max.ms/> <http://retry.backoff.max.ms/> <
>> http://retry.backoff.max.ms/> <
>>>> http://retry.backoff.max.ms/>
>>>>>>>> <http://retry.backoff.max.ms/> <
>>>>>>>>>> http://retry.backoff.max.ms/> <
>>>>>>>>>>>> http://retry.backoff.max.ms/> and
>>>>>>>>>>>>> retry.backoff.ms <http://retry.backoff.ms/> <http://retry.backoff.ms/> <
>> http://retry.backoff.ms/> <
>>>> http://retry.backoff.ms/> <
>>>>>> http://retry.backoff.ms/> <
>>>>>>>> http://retry.backoff.ms/> <http://retry.backoff.ms/
>>>>>>>>>>> .
>>>>>>>>>>>>> 
>>>>>>>>>>>>> JR6. large.message.skip.not.found.error : If the reference
>> can't
>>>> be
>>>>>>>>>>>> found,
>>>>>>>>>>>>> what value does the deserializer return? Note that null has a
>>>>>> special
>>>>>>>>>>>>> meaning for tombstone in compacted topics.
>>>>>>>>>>>>> 
>>>>>>>>>>>>> JR7. PayloadResponse: Why do we have both responseCode and
>>>>>>>>>>>>> PayloadStoreException?
>>>>>>>>>>>>> 
>>>>>>>>>>>>> JR8. Why do we need PayloadStore.metrics? Note that we could
>>>>>> monitor
>>>>>>>>>> the
>>>>>>>>>>>>> metrics in a plugin through the Monitorable interface.
>>>>>>>>>>>>> 
>>>>>>>>>>>>> JR9. Why do we need the protected field
>>>>>>>>>>>> PayloadStoreException.isRetryable?
>>>>>>>>>>>>> 
>>>>>>>>>>>>> JR10. As Luke mentioned earlier, we could turn PayloadStore to
>> an
>>>>>>>>>>>> interface.
>>>>>>>>>>>>> 
>>>>>>>>>>>>> Thanks,

Re: [DISCUSS] KIP-1159: Large message reference based Serializer

[Greg Harris <gr...@aiven.io.INVALID>]

Hi Omnia,

Thanks for your replies.

1.
> There are other concerns like amplifying all Kafka resources like network
bandwidth, memory pressure, and I/O operations becoming more unpredictable.

Thanks for pointing this out, I agree that KIP-1163 does not address these
multi-tenant concerns. I brought up KIP-1163 because I thought it would
solve some but not all of the large-message concerns.
However, the technical problems you mention could be addressed under the
umbrella of "improving Kafka's handling of large messages in-band".
Investments in resource isolation/cell based architecture tooling/memory
management will be broadly beneficial to Kafka, improving both
small-message and large-message workloads.

2.
> I would prefer this to be in Kafka anyway at least the composable
serialization - this is beneficial for multiple use-cases and doesn't need
to be outside of Kafka it should be standard feature

Could you please add some details for these other use-cases? We already
have multi-message compression in-protocol (which is often better than
single-message compression).
We don't have per-message encryption, but this is already addressed by
Kroxylicious [1]. I'm not aware of any other binary-to-binary operations
that would be relevant here.

3. On the subject of Kroxylicious, could another rejected alternative be to
implement reference based messaging as a filter for Kroxylicious?

4.
> I am personally not aware of any one willing to offer this as external
OSS.

I also don't know of any organization willing to offer this as external OSS
using the existing interfaces. And that makes me seriously doubt there's an
organization willing to offer this as external OSS using
yet-unmerged-interfaces that duplicate the existing interfaces.

5.
> Also am maybe missing something here but how this is different than
implementing interface to something like TS or Diskless?

AFAIU, there is only one realistic open-source implementation of tiered
storage, and it's maintained by Aiven. [2]
For Diskless, Aiven will also maintain an open-source implementation of any
pluggable interfaces required to make the feature functional.

Does this proposal also require Aiven to take on the responsibility of
implementing reference based messaging plugins? If Aiven declines to
maintain an implementation, is the interface actually beneficial to the
project?

Thanks,
Greg

[1] https://kroxylicious.io/
[2] https://github.com/Aiven-Open/tiered-storage-for-apache-kafka

On Tue, Aug 19, 2025 at 9:08 AM Omnia Ibrahim <o....@gmail.com>
wrote:

> Hi Greg,
> Thanks for reviewing the KIP!
>
> > > Could you discuss the rejected alternative of improving Kafka's
> handling of
> > > large messages in-band? If Kafka is better at handling large messages,
> then
> > > it may not be necessary to adopt the complexity of reference-based
> > > messaging.
> > > I wonder if these problems truly disappear with referenced based
> messaging,
> > > or if they are moved from broker to applications, where the Apache
> Kafka
> > > project isn't blamed but also has less ability to address. Client
> > > applications may have worse memory management or observability than
> Kafka
> > > brokers.
> > > For example, I think that KIP-1163 may benefit large-message workloads,
> > > because it allows for the time-to-commit a batch to be decoupled from
> the
> > > time-to-write the batch.
>
> Regarding the mention of KIP-1163 benefiting large message or Kafka better
> dealing with large message I would say this would be tricky if cluster is
> operated/served as multi-tenant.
> In shared Kafka clusters, large messages from one application can degrade
> performance for all other topics and clients. This violates the isolation
> principle that makes Kafka suitable for multi-tenant environments.
>
> While improvements like KIP-1163 help with publishing data into external
> storage, they don't address the fundamental resource contention issues.
> There are other concerns like amplifying all Kafka resources like network
> bandwidth, memory pressure, and I/O operations becoming more unpredictable.
>
>
> The client-side approach provides better isolation where storage failures,
> or payload’s size don't impact Kafka cluster availability and performance
> for all use-cases, applications can choose appropriate storage for their
> specific use case which would be complicated to coordinate at the broker
> level on multi-tenant where each topic needs different external storage.
>
> > > Could there be another rejected alternative to open-source an existing
> > > reference-based-messaging solution? If there are community members with
> > > awareness of an existing custom solution, they could also influence
> > > open-sourcing it.
>
> I am personally not aware of any one willing to offer this as external
> OSS. I would prefer this to be in Kafka anyway at least the composable
> serialization - this is beneficial for multiple use-cases and doesn't need
> to be outside of Kafka it should be standard feature.
>
>
> > > AFAIU this proposal still requires implementations of the new
> interfaces to
> > > be in a separate project outside of Apache Kafka, because we cannot
> depend
> > > on object storage implementations in Apache Kafka.
> > > Such a project could own the new interfaces itself, either as private
> > > interfaces or public API it exposes to further downstream projects.
> This
> > > will have fewer restrictions on the API design, as a vote on Apache
> Kafka
> > > is not necessary to evolve the functionality.
>
> Yes, as we have similar interface in Kafka for example storage layer in
> Tiered Storage and client interceptors where everyone can point to their
> own implementation. These examples started with Kafka providing
> standardized interfaces then different implementations were developed by
> the community later.
>
> > > Can you clarify what the benefit is for this change to land in Apache
> Kafka
> > > specifically? And if there is a group ready to develop and distribute
> > > open-source implementations of this interface? Otherwise we will end up
> > > back where we started with only closed-source implementations.
>
>
> The "closed-source implementations” on storage layer is not the issue the
> pattern of applying schemas or different serializers then uploading the
> payload is the pattern that is everyone doing behind closed doors. Also am
> maybe missing something here but how this is different than implementing
> interface to something like TS or Diskless?
>
>
> I’ll update the KIP with some of these points.
>
> Thanks
> Omnia
>
> > On 13 Aug 2025, at 17:28, Greg Harris <gr...@aiven.io.INVALID>
> wrote:
> >
> > Hi Omnia,
> >
> > Thanks for the KIP!
> >
> > I had some questions about the rejected alternatives:
> >
> > 1. The motivation for the KIP includes this:
> >> Increasing message size indefinitely is not a viable solution as it can
> > lead to performance degradation, memory issues, and instability of the
> > broker.
> >
> > Could you discuss the rejected alternative of improving Kafka's handling
> of
> > large messages in-band? If Kafka is better at handling large messages,
> then
> > it may not be necessary to adopt the complexity of reference-based
> > messaging.
> > I wonder if these problems truly disappear with referenced based
> messaging,
> > or if they are moved from broker to applications, where the Apache Kafka
> > project isn't blamed but also has less ability to address. Client
> > applications may have worse memory management or observability than Kafka
> > brokers.
> > For example, I think that KIP-1163 may benefit large-message workloads,
> > because it allows for the time-to-commit a batch to be decoupled from the
> > time-to-write the batch.
> >
> > 2. The "popular patterns" section includes this:
> >> Available open-source implementation
> >> There isn't a specific open-source implementation for Reference-Based
> > Messaging. Mostly everyone adopting this has their custom solution
> >
> > Could there be another rejected alternative to open-source an existing
> > reference-based-messaging solution? If there are community members with
> > awareness of an existing custom solution, they could also influence
> > open-sourcing it.
> >
> > 3. This is listed as a rejected alternative:
> >> Implement this as a separate project outside of Apache Kafka as this
> > seems to be a pattern that needs more use cases and it would be better to
> > have this in Apache Kafka as native implementation instead of a separate
> > project.
> >
> > AFAIU this proposal still requires implementations of the new interfaces
> to
> > be in a separate project outside of Apache Kafka, because we cannot
> depend
> > on object storage implementations in Apache Kafka.
> > Such a project could own the new interfaces itself, either as private
> > interfaces or public API it exposes to further downstream projects. This
> > will have fewer restrictions on the API design, as a vote on Apache Kafka
> > is not necessary to evolve the functionality.
> >
> > Can you clarify what the benefit is for this change to land in Apache
> Kafka
> > specifically? And if there is a group ready to develop and distribute
> > open-source implementations of this interface? Otherwise we will end up
> > back where we started with only closed-source implementations.
> >
> > Thanks,
> > Greg
> >
> > On Wed, Aug 13, 2025 at 5:27 AM Omnia Ibrahim <o.g.h.ibrahim@gmail.com
> <ma...@gmail.com>>
> > wrote:
> >
> >> Hi Jun
> >>>
> >>> JR20. The inconsistency is still there. It's probably simpler to
> >>> consolidate on
> >>> org.apache.kafka.common.serialization.largemessage.PayloadStore.
> >> Updated the kip now to address these
> >>> JR21. Could you document how the payload id string is encoded? For
> >> example,
> >>> is it always UTF8?
> >> I updated the java doc, the default implementation do provide UTF-8
> >> however if the developer want to support any other they will need to
> >> address this in the implementation of the store.
> >>>
> >>> JR22. "Since compacted topics can retain data indefinitely, users must
> >>> choose between setting a business-appropriate TTL or accepting
> indefinite
> >>> storage costs." Setting a TTL is not ideal since it can break the
> >> consumer
> >>> application. So, we probably don't want to recommend it to the users.
> >>> Another option is to run a cleaning job by consuming the topic and if
> it
> >>> observes a null payload, delete the object based on the key in the
> >> record.
> >>
> >> Updated the recommendations
> >>> JR24. In the following,
> org.apache.kafka.common.serialization.Serializer
> >>> should include the largemessage namespace.
> >>> producerConfig.put("value.serializers",
> >>> "kafka.serializers.KafkaAvroDeserializer,
> >>> org.apache.kafka.common.serialization.Serializer");
> >>
> >> Updated now
> >>
> >> Thanks
> >> Omnia
> >>
> >>> On 12 Aug 2025, at 18:09, Jun Rao <ju...@confluent.io.INVALID> wrote:
> >>>
> >>> Hi, Omnia,
> >>>
> >>> Thanks for the reply. A few more comments.
> >>>
> >>> JR20. The inconsistency is still there. It's probably simpler to
> >>> consolidate on
> >>> org.apache.kafka.common.serialization.largemessage.PayloadStore.
> >>>
> >>> JR21. Could you document how the payload id string is encoded? For
> >> example,
> >>> is it always UTF8?
> >>>
> >>> JR22. "Since compacted topics can retain data indefinitely, users must
> >>> choose between setting a business-appropriate TTL or accepting
> indefinite
> >>> storage costs." Setting a TTL is not ideal since it can break the
> >> consumer
> >>> application. So, we probably don't want to recommend it to the users.
> >>> Another option is to run a cleaning job by consuming the topic and if
> it
> >>> observes a null payload, delete the object based on the key in the
> >> record.
> >>>
> >>> JR23. We probably don't need to force the PayloadStore implementer to
> >>> implement Monitorable. We could document that if it's desirable to
> >> monitor
> >>> a PayloadStore plugin, one can implement the Monitorable interface.
> >>>
> >>> JR24. In the following,
> org.apache.kafka.common.serialization.Serializer
> >>> should include the largemessage namespace.
> >>> producerConfig.put("value.serializers",
> >>> "kafka.serializers.KafkaAvroDeserializer,
> >>> org.apache.kafka.common.serialization.Serializer");
> >>>
> >>> Jun
> >>>
> >>> On Tue, Aug 12, 2025 at 3:27 AM Omnia Ibrahim <o.g.h.ibrahim@gmail.com
> >> <ma...@gmail.com>>
> >>> wrote:
> >>>
> >>>> Hi Jun,
> >>>>
> >>>>> JR20. Could you fix the inconsistency?
> >>>>> org.apache.kafka.common.serialization.largemessage.PayloadStore vs
> >>>>> org.apache.kafka.common.serialization.largemessage.store.PayloadStore
> >>>> Updated
> >>>>>
> >>>>> JR21. "Encapsulate that ID into a simple Kafka event using a
> structured
> >>>>> format.". Could you define the structured format explicitly?
> >>>>
> >>>> This was from the initial design we had Id format but after discussing
> >>>> this with Luke I ended up simplifying this by removing the format and
> I
> >>>> guess I forgot this line. I have updated the kip now.
> >>>>
> >>>>> JR22. TTL for object stores: Could the new serializer be used on a
> >>>>> compacted topic? If so, how should a user configure the TTL since the
> >>>> Kafka
> >>>>> retention is infinite.
> >>>>
> >>>> Great question about compacted topics! There are few usage of
> compacted
> >>>> topic to consider:
> >>>> 1. Typical Compacted Topic Usage: The compacted topics are commonly
> used
> >>>> as cache/state layers where the latest value for each key represents
> the
> >>>> current state. These use cases typically involve smaller records since
> >> they
> >>>> need to be loaded efficiently for state reconstruction. Large message
> >>>> serialization would be unusual for this pattern.
> >>>>
> >>>> 2. User-Defined Compacted Topics with Large Payloads: for applications
> >>>> that do use compacted topics with large payloads, the PayloadStore
> >>>> implementation should handle this by:
> >>>>       a. Consistent ID Generation: Use deterministic IDs based on the
> >>>> Kafka key (rather than random UUIDs) so that when a payload id is
> >> updated,
> >>>> it overwrites the same payload store object instead of creating new
> >> ones.
> >>>>
> >>>>       b. TTL Strategy: Since compacted topics can retain data
> >>>> indefinitely, users have two options:
> >>>>               - Set a business-appropriate TTL (basically "we know our
> >>>> cache data becomes stale after 30 days”)
> >>>>               - Configure no TTL and accept indefinite storage costs
> as
> >>>> a trade-off for the architectural benefits
> >>>>
> >>>> 3. Compact + Delete Policy: Topics with
> `cleanup.policy=compact,delete`
> >>>> will eventually remove old data, so standard TTL approaches work
> >> normally.
> >>>> The key insight is that TTL configuration depends on the business
> >>>> requirements and usage patterns rather than just the Kafka retention
> >>>> policy. The PayloadStore implementation should provide flexibility for
> >>>> users to make this trade-off consciously.
> >>>>
> >>>> I have update the consideration section with this.
> >>>>
> >>>> Thanks
> >>>> Omnia
> >>>>
> >>>>> On 11 Aug 2025, at 23:32, Jun Rao <ju...@confluent.io.INVALID> wrote:
> >>>>>
> >>>>> Hi, Omnia,
> >>>>>
> >>>>> Thanks for the reply. A few more comments.
> >>>>>
> >>>>> JR20. Could you fix the inconsistency?
> >>>>> org.apache.kafka.common.serialization.largemessage.PayloadStore vs
> >>>>> org.apache.kafka.common.serialization.largemessage.store.PayloadStore
> >>>>>
> >>>>> JR21. "Encapsulate that ID into a simple Kafka event using a
> structured
> >>>>> format.". Could you define the structured format explicitly?
> >>>>>
> >>>>> JR22. TTL for object stores: Could the new serializer be used on a
> >>>>> compacted topic? If so, how should a user configure the TTL since the
> >>>> Kafka
> >>>>> retention is infinite.
> >>>>>
> >>>>> Jun
> >>>>>
> >>>>> On Mon, Aug 11, 2025 at 10:08 AM Omnia Ibrahim <
> >> o.g.h.ibrahim@gmail.com
> >>>> <ma...@gmail.com>>
> >>>>> wrote:
> >>>>>
> >>>>>> Hi Jun
> >>>>>>> JR11. Do you think that we need to add key.serializers and
> >>>>>>> key.deserializers or do you think covering large messages in value
> is
> >>>>>>> enough?
> >>>>>> The large message issue usually is a value issue, I never saw a key
> >> that
> >>>>>> is bigger than broker message size. If I would add key.serializers
> and
> >>>>>> key.deserializers it would be for consistency and maybe there are
> >>>> use-cases
> >>>>>> where developers want to apply multiple serializations in order on
> the
> >>>> key
> >>>>>> as well outside the context of the large message support.
> >>>>>> I updated the KIP to add these two configs now.
> >>>>>>
> >>>>>>> JR12. We can load ComposableSerializer automatically if
> >>>> value.serializers
> >>>>>>> or value.deserializers are specified. But, it seems that we could
> >> keep
> >>>>>>> ComposableSerializer as an internal implementation. For example,
> >>>>>>> ProducerInterceptors is automatically loaded when multiple
> >> interceptors
> >>>>>> are
> >>>>>>> specified and is an internal class.
> >>>>>> I updated the kip to highlight that the changes is updating
> >>>> ProducerConfig
> >>>>>> and ConsumerConfig to have new config for serializers/deserializers
> >> and
> >>>> not
> >>>>>> the actual class this is implementation details and not public
> >>>> interfaces.
> >>>>>>
> >>>>>>> JR17. We could estimate the size after compression, but the
> estimator
> >>>> is
> >>>>>>> not 100% accurate. It seems that it's simpler to just use the
> >> original
> >>>>>>> message size.
> >>>>>> We can keep it as original message size, I was thinking if it is
> good
> >>>>>> enough for max.request.size it might be good enough for this. I
> >> updated
> >>>> the
> >>>>>> KIP anyway to simplify it and keep it to the original size.
> >>>>>>
> >>>>>> Hope the final version addressed all feedbacks and we can resume
> with
> >>>> the
> >>>>>> voting
> >>>>>>
> >>>>>> Thanks
> >>>>>> Omnia
> >>>>>>
> >>>>>>> On 8 Aug 2025, at 22:10, Jun Rao <ju...@confluent.io.INVALID> wrote:
> >>>>>>>
> >>>>>>> Hi, Omnia,
> >>>>>>>
> >>>>>>> Thanks for the reply.
> >>>>>>>
> >>>>>>> JR11. Do you think that we need to add key.serializers and
> >>>>>>> key.deserializers or do you think covering large messages in value
> is
> >>>>>>> enough?
> >>>>>>>
> >>>>>>> JR12. We can load ComposableSerializer automatically if
> >>>> value.serializers
> >>>>>>> or value.deserializers are specified. But, it seems that we could
> >> keep
> >>>>>>> ComposableSerializer as an internal implementation. For example,
> >>>>>>> ProducerInterceptors is automatically loaded when multiple
> >> interceptors
> >>>>>> are
> >>>>>>> specified and is an internal class.
> >>>>>>>
> >>>>>>> JR17. We could estimate the size after compression, but the
> estimator
> >>>> is
> >>>>>>> not 100% accurate. It seems that it's simpler to just use the
> >> original
> >>>>>>> message size.
> >>>>>>>
> >>>>>>> Jun
> >>>>>>>
> >>>>>>> On Mon, Aug 4, 2025 at 10:33 AM Omnia Ibrahim <
> >> o.g.h.ibrahim@gmail.com
> >>>>>> <ma...@gmail.com>>
> >>>>>>> wrote:
> >>>>>>>
> >>>>>>>>
> >>>>>>>> Hi Jun
> >>>>>>>>> JR11. value.serializers and value.deserializers: Should they be
> of
> >>>> type
> >>>>>>>>> List? Also, where are key.serializers and key.deserializers?
> >>>>>>>>>
> >>>>>>>> Updated now
> >>>>>>>>> JR12. Do we still need ComposableSerializer and
> >>>> ComposableDeserializer?
> >>>>>>>> The initial thinking here was if `value.serializers or
> >>>>>> value.deserializers
> >>>>>>>> ` exists the client will load  ComposableSerializer or
> >>>>>>>> ComposableDeserializer  automatically and use them. Unfortunately
> >> this
> >>>>>>>> would need us to define these serializers.
> >>>>>>>>
> >>>>>>>> The other option is to update `Plugin<Serializer<V>>
> >>>>>>>> valueSerializerPlugin` and KafkaProducer constructor to accept
> >>>>>>>> List<Serializer<V>> and move the logic of the ComposableSerializer
> >>>> into
> >>>>>>>> KafkaProducer::doSend which when we do serialization (same with
> >>>>>>>> KafkaConsumer). This option hide the logic and reduce exposure for
> >>>>>> client
> >>>>>>>> to these.
> >>>>>>>> WDYT?
> >>>>>>>>
> >>>>>>>>>
> >>>>>>>>> JR13. large.message.payload.store.class : should it be of type
> >> class?
> >>>>>>>> Updated
> >>>>>>>>>
> >>>>>>>>> JR14.
> >>>>>>>>>
> >>>>>>>>
> >>>>>>
> >>>>
> >>
> org.apache.kafka.common.serialization.largemessage.LargeMessageSerializer :
> >>>>>>>>> The name seems redundant since largemessage appears twice.
> >>>>>>>> Updated
> >>>>>>>>>
> >>>>>>>>> JR15. PayloadResponse: It still mentions response code. It
> mentions
> >>>>>>>>> "isRetryable flag", which no longer exists in
> >> PayloadStoreException.
> >>>>>>>> There
> >>>>>>>>> are typos in "then it will serialiser will”.
> >>>>>>>> The KIP is updated now
> >>>>>>>>
> >>>>>>>>> JR16. Regarding returning new byte[0] if
> >>>>>>>> large.message.skip.not.found.error
> >>>>>>>>> is true, this will likely fail the next deserializer and the
> >>>>>> application
> >>>>>>>>> won't have the right context of the error. It's probably better
> to
> >>>> just
> >>>>>>>>> propagate the specific exception and let the caller handle it.
> >>>>>>>> You right this will cause issue if there is another deserializer
> >>>> waiting
> >>>>>>>> for the data. I have updated the KIP with this.
> >>>>>>>>
> >>>>>>>>>
> >>>>>>>>> JR17. LargeMessageSerializer:  "Check if the estimated size of
> the
> >>>> data
> >>>>>>>>> (bytes) after applying provided compression (if there is one)"
> >>>>>>>> Compression
> >>>>>>>>> actually happens after serialization and is done on a batch of
> >>>> records.
> >>>>>>>> Yes the compression itself happened after but we also have
> >>>>>>>> `estimateSizeInBytesUpperBound` which my understanding is this
> take
> >>>> the
> >>>>>>>> compression type into the account as well
> >>>>>>>>>
> >>>>>>>>>
> >>>>>>>>> JR18. Could you define the type T for LargeMessageSerializer?
> >>>>>>>> Update the KIP this T would be byte[]
> >>>>>>>>
> >>>>>>>> Thanks
> >>>>>>>> Omnia
> >>>>>>>>>
> >>>>>>>>> Jun
> >>>>>>>>>
> >>>>>>>>> On Fri, Jul 25, 2025 at 6:28 AM Omnia Ibrahim <
> >>>> o.g.h.ibrahim@gmail.com
> >>>>>>>> <ma...@gmail.com>>
> >>>>>>>>> wrote:
> >>>>>>>>>
> >>>>>>>>>> Hi Jun, thanks for having the time to review this
> >>>>>>>>>>
> >>>>>>>>>>> JR1. While the KIP is potentially useful, I am wondering who is
> >>>>>>>>>> responsible
> >>>>>>>>>>> for retention for the objects in the payload store. Once a
> >> message
> >>>>>>>> with a
> >>>>>>>>>>> reference is deleted, the key of the external object is lost
> and
> >>>> the
> >>>>>>>>>> object
> >>>>>>>>>>> may never be deleted.
> >>>>>>>>>>
> >>>>>>>>>> The `ttl` in the object store is the responsibility of the owner
> >> of
> >>>>>> this
> >>>>>>>>>> store it should be configured in away that is reasonable with
> the
> >>>>>>>> retention
> >>>>>>>>>> config in Kafka.
> >>>>>>>>>> I have updated the KIP with `Consideration` section.
> >>>>>>>>>>>
> >>>>>>>>>>> JR2. Configs: For all new configs, it would be useful to list
> >> their
> >>>>>>>>>> types.
> >>>>>>>>>> Updated the KIP now
> >>>>>>>>>>>
> >>>>>>>>>>> JR3. value.serializers: Why is this required? If a user doesn't
> >> set
> >>>>>> it,
> >>>>>>>>>> we
> >>>>>>>>>>> should just use value.serializer, right? Ditto for
> >> key.serializers.
> >>>>>>>>>> No you right this was copy/past mistake
> >>>>>>>>>>
> >>>>>>>>>>>
> >>>>>>>>>>> JR4. For all new public interfaces such as
> >> LargeMessageSerializer,
> >>>>>>>>>>> PayloadStore and PayloadResponse, it would be useful to include
> >> the
> >>>>>>>> full
> >>>>>>>>>>> package name.
> >>>>>>>>>> Updated the KIP now
> >>>>>>>>>>>
> >>>>>>>>>>> JR5. large.message.payload.store.retry.max.backoff.ms and
> >>>>>>>>>>> large.message.payload.store.retry.delay.backoff.ms: Is the
> >>>> intention
> >>>>>>>> to
> >>>>>>>>>>> implement exponential backoff on retries? If so, it's more
> >>>> consistent
> >>>>>>>> if
> >>>>>>>>>> we
> >>>>>>>>>>> can follow the existing naming convention like
> >>>> retry.backoff.max.ms
> >>>>>> <
> >>>>>>>>>> http://retry.backoff.max.ms/> and
> >>>>>>>>>>> retry.backoff.ms <http://retry.backoff.ms/> <
> >>>>>> http://retry.backoff.ms/> <http://retry.backoff.ms/
> >>>>>>>>> .
> >>>>>>>>>> I have removed these to simplify the config more (as Luke
> >> suggested
> >>>>>>>>>> initially) and added these to the consideration section.
> >>>>>>>>>>
> >>>>>>>>>>>
> >>>>>>>>>>> JR6. large.message.skip.not.found.error : If the reference
> can't
> >> be
> >>>>>>>>>> found,
> >>>>>>>>>>> what value does the deserializer return? Note that null has a
> >>>> special
> >>>>>>>>>>> meaning for tombstone in compacted topics.
> >>>>>>>>>> The deserialiser will return `new byte[0]` not null.
> >>>>>>>>>>>
> >>>>>>>>>>> JR7. PayloadResponse: Why do we have both responseCode and
> >>>>>>>>>>> PayloadStoreException?
> >>>>>>>>>> We can do without responseCode, the initial though was to report
> >>>>>>>> response
> >>>>>>>>>> code form payload store.
> >>>>>>>>>> Update the KIP.
> >>>>>>>>>>> JR8. Why do we need PayloadStore.metrics? Note that we could
> >>>> monitor
> >>>>>>>> the
> >>>>>>>>>>> metrics in a plugin through the Monitorable interface.
> >>>>>>>>>> Oh nice, I didn’t know about this interface before. Updated the
> >> KIP
> >>>>>> with
> >>>>>>>>>> this now.
> >>>>>>>>>>>
> >>>>>>>>>>> JR9. Why do we need the protected field
> >>>>>>>>>> PayloadStoreException.isRetryable?
> >>>>>>>>>> Initial thought here was the serializer can retry the upload.
> But
> >> I
> >>>>>> have
> >>>>>>>>>> removed all the retry logic from serializer and it will be up to
> >> the
> >>>>>>>>>> PayloadStore provider to implement this if they need it.
> >>>>>>>>>>>
> >>>>>>>>>>> JR10. As Luke mentioned earlier, we could turn PayloadStore to
> an
> >>>>>>>>>> interface.
> >>>>>>>>>> It is updated now to interface.
> >>>>>>>>>>
> >>>>>>>>>> Hope the last version of the KIP is more simpler now
> >>>>>>>>>>
> >>>>>>>>>> Thanks
> >>>>>>>>>> Omnia
> >>>>>>>>>>
> >>>>>>>>>>> On 23 Jul 2025, at 00:43, Jun Rao <jun@confluent.io.INVALID
> >>>> <ma...@confluent.io.INVALID> <mailto:
> >>>>>> jun@confluent.io.INVALID <ma...@confluent.io.INVALID> <mailto:
> >> jun@confluent.io.INVALID>> <mailto:
> >>>>>>>> jun@confluent.io.INVALID <ma...@confluent.io.INVALID>
> <ma...@confluent.io.INVALID> <mailto:
> >> jun@confluent.io.INVALID <ma...@confluent.io.INVALID>> <mailto:
> >>>> jun@confluent.io.INVALID <ma...@confluent.io.INVALID> <mailto:
> jun@confluent.io.INVALID>>>> wrote:
> >>>>>>>>>>>
> >>>>>>>>>>> Thanks for the KIP. A few comments.
> >>>>>>>>>>>
> >>>>>>>>>>> JR1. While the KIP is potentially useful, I am wondering who is
> >>>>>>>>>> responsible
> >>>>>>>>>>> for retention for the objects in the payload store. Once a
> >> message
> >>>>>>>> with a
> >>>>>>>>>>> reference is deleted, the key of the external object is lost
> and
> >>>> the
> >>>>>>>>>> object
> >>>>>>>>>>> may never be deleted.
> >>>>>>>>>>>
> >>>>>>>>>>> JR2. Configs: For all new configs, it would be useful to list
> >> their
> >>>>>>>>>> types.
> >>>>>>>>>>>
> >>>>>>>>>>> JR3. value.serializers: Why is this required? If a user doesn't
> >> set
> >>>>>> it,
> >>>>>>>>>> we
> >>>>>>>>>>> should just use value.serializer, right? Ditto for
> >> key.serializers.
> >>>>>>>>>>>
> >>>>>>>>>>> JR4. For all new public interfaces such as
> >> LargeMessageSerializer,
> >>>>>>>>>>> PayloadStore and PayloadResponse, it would be useful to include
> >> the
> >>>>>>>> full
> >>>>>>>>>>> package name.
> >>>>>>>>>>>
> >>>>>>>>>>> JR5. large.message.payload.store.retry.max.backoff.ms and
> >>>>>>>>>>> large.message.payload.store.retry.delay.backoff.ms: Is the
> >>>> intention
> >>>>>>>> to
> >>>>>>>>>>> implement exponential backoff on retries? If so, it's more
> >>>> consistent
> >>>>>>>> if
> >>>>>>>>>> we
> >>>>>>>>>>> can follow the existing naming convention like
> >>>> retry.backoff.max.ms <http://retry.backoff.max.ms/> <
> http://retry.backoff.max.ms/> <
> >> http://retry.backoff.max.ms/>
> >>>>>> <http://retry.backoff.max.ms/> <
> >>>>>>>> http://retry.backoff.max.ms/> <
> >>>>>>>>>> http://retry.backoff.max.ms/> and
> >>>>>>>>>>> retry.backoff.ms <http://retry.backoff.ms/> <
> http://retry.backoff.ms/> <
> >> http://retry.backoff.ms/> <
> >>>> http://retry.backoff.ms/> <
> >>>>>> http://retry.backoff.ms/> <http://retry.backoff.ms/
> >>>>>>>>> .
> >>>>>>>>>>>
> >>>>>>>>>>> JR6. large.message.skip.not.found.error : If the reference
> can't
> >> be
> >>>>>>>>>> found,
> >>>>>>>>>>> what value does the deserializer return? Note that null has a
> >>>> special
> >>>>>>>>>>> meaning for tombstone in compacted topics.
> >>>>>>>>>>>
> >>>>>>>>>>> JR7. PayloadResponse: Why do we have both responseCode and
> >>>>>>>>>>> PayloadStoreException?
> >>>>>>>>>>>
> >>>>>>>>>>> JR8. Why do we need PayloadStore.metrics? Note that we could
> >>>> monitor
> >>>>>>>> the
> >>>>>>>>>>> metrics in a plugin through the Monitorable interface.
> >>>>>>>>>>>
> >>>>>>>>>>> JR9. Why do we need the protected field
> >>>>>>>>>> PayloadStoreException.isRetryable?
> >>>>>>>>>>>
> >>>>>>>>>>> JR10. As Luke mentioned earlier, we could turn PayloadStore to
> an
> >>>>>>>>>> interface.
> >>>>>>>>>>>
> >>>>>>>>>>> Thanks,
>
>

Re: [DISCUSS] KIP-1159: Large message reference based Serializer

[Omnia Ibrahim <o....@gmail.com>]

Hi Greg, 
Thanks for reviewing the KIP!

> > Could you discuss the rejected alternative of improving Kafka's handling of
> > large messages in-band? If Kafka is better at handling large messages, then
> > it may not be necessary to adopt the complexity of reference-based
> > messaging.
> > I wonder if these problems truly disappear with referenced based messaging,
> > or if they are moved from broker to applications, where the Apache Kafka
> > project isn't blamed but also has less ability to address. Client
> > applications may have worse memory management or observability than Kafka
> > brokers.
> > For example, I think that KIP-1163 may benefit large-message workloads,
> > because it allows for the time-to-commit a batch to be decoupled from the
> > time-to-write the batch.

Regarding the mention of KIP-1163 benefiting large message or Kafka better dealing with large message I would say this would be tricky if cluster is operated/served as multi-tenant.
In shared Kafka clusters, large messages from one application can degrade performance for all other topics and clients. This violates the isolation principle that makes Kafka suitable for multi-tenant environments.

While improvements like KIP-1163 help with publishing data into external storage, they don't address the fundamental resource contention issues. There are other concerns like amplifying all Kafka resources like network bandwidth, memory pressure, and I/O operations becoming more unpredictable.


The client-side approach provides better isolation where storage failures, or payload’s size don't impact Kafka cluster availability and performance for all use-cases, applications can choose appropriate storage for their specific use case which would be complicated to coordinate at the broker level on multi-tenant where each topic needs different external storage.

> > Could there be another rejected alternative to open-source an existing
> > reference-based-messaging solution? If there are community members with
> > awareness of an existing custom solution, they could also influence
> > open-sourcing it.

I am personally not aware of any one willing to offer this as external OSS. I would prefer this to be in Kafka anyway at least the composable serialization - this is beneficial for multiple use-cases and doesn't need to be outside of Kafka it should be standard feature.


> > AFAIU this proposal still requires implementations of the new interfaces to
> > be in a separate project outside of Apache Kafka, because we cannot depend
> > on object storage implementations in Apache Kafka.
> > Such a project could own the new interfaces itself, either as private
> > interfaces or public API it exposes to further downstream projects. This
> > will have fewer restrictions on the API design, as a vote on Apache Kafka
> > is not necessary to evolve the functionality.

Yes, as we have similar interface in Kafka for example storage layer in Tiered Storage and client interceptors where everyone can point to their own implementation. These examples started with Kafka providing standardized interfaces then different implementations were developed by the community later.

> > Can you clarify what the benefit is for this change to land in Apache Kafka
> > specifically? And if there is a group ready to develop and distribute
> > open-source implementations of this interface? Otherwise we will end up
> > back where we started with only closed-source implementations.


The "closed-source implementations” on storage layer is not the issue the pattern of applying schemas or different serializers then uploading the payload is the pattern that is everyone doing behind closed doors. Also am maybe missing something here but how this is different than implementing interface to something like TS or Diskless?
 

I’ll update the KIP with some of these points. 

Thanks
Omnia

> On 13 Aug 2025, at 17:28, Greg Harris <gr...@aiven.io.INVALID> wrote:
> 
> Hi Omnia,
> 
> Thanks for the KIP!
> 
> I had some questions about the rejected alternatives:
> 
> 1. The motivation for the KIP includes this:
>> Increasing message size indefinitely is not a viable solution as it can
> lead to performance degradation, memory issues, and instability of the
> broker.
> 
> Could you discuss the rejected alternative of improving Kafka's handling of
> large messages in-band? If Kafka is better at handling large messages, then
> it may not be necessary to adopt the complexity of reference-based
> messaging.
> I wonder if these problems truly disappear with referenced based messaging,
> or if they are moved from broker to applications, where the Apache Kafka
> project isn't blamed but also has less ability to address. Client
> applications may have worse memory management or observability than Kafka
> brokers.
> For example, I think that KIP-1163 may benefit large-message workloads,
> because it allows for the time-to-commit a batch to be decoupled from the
> time-to-write the batch.
> 
> 2. The "popular patterns" section includes this:
>> Available open-source implementation
>> There isn't a specific open-source implementation for Reference-Based
> Messaging. Mostly everyone adopting this has their custom solution
> 
> Could there be another rejected alternative to open-source an existing
> reference-based-messaging solution? If there are community members with
> awareness of an existing custom solution, they could also influence
> open-sourcing it.
> 
> 3. This is listed as a rejected alternative:
>> Implement this as a separate project outside of Apache Kafka as this
> seems to be a pattern that needs more use cases and it would be better to
> have this in Apache Kafka as native implementation instead of a separate
> project.
> 
> AFAIU this proposal still requires implementations of the new interfaces to
> be in a separate project outside of Apache Kafka, because we cannot depend
> on object storage implementations in Apache Kafka.
> Such a project could own the new interfaces itself, either as private
> interfaces or public API it exposes to further downstream projects. This
> will have fewer restrictions on the API design, as a vote on Apache Kafka
> is not necessary to evolve the functionality.
> 
> Can you clarify what the benefit is for this change to land in Apache Kafka
> specifically? And if there is a group ready to develop and distribute
> open-source implementations of this interface? Otherwise we will end up
> back where we started with only closed-source implementations.
> 
> Thanks,
> Greg
> 
> On Wed, Aug 13, 2025 at 5:27 AM Omnia Ibrahim <o.g.h.ibrahim@gmail.com <ma...@gmail.com>>
> wrote:
> 
>> Hi Jun
>>> 
>>> JR20. The inconsistency is still there. It's probably simpler to
>>> consolidate on
>>> org.apache.kafka.common.serialization.largemessage.PayloadStore.
>> Updated the kip now to address these
>>> JR21. Could you document how the payload id string is encoded? For
>> example,
>>> is it always UTF8?
>> I updated the java doc, the default implementation do provide UTF-8
>> however if the developer want to support any other they will need to
>> address this in the implementation of the store.
>>> 
>>> JR22. "Since compacted topics can retain data indefinitely, users must
>>> choose between setting a business-appropriate TTL or accepting indefinite
>>> storage costs." Setting a TTL is not ideal since it can break the
>> consumer
>>> application. So, we probably don't want to recommend it to the users.
>>> Another option is to run a cleaning job by consuming the topic and if it
>>> observes a null payload, delete the object based on the key in the
>> record.
>> 
>> Updated the recommendations
>>> JR24. In the following, org.apache.kafka.common.serialization.Serializer
>>> should include the largemessage namespace.
>>> producerConfig.put("value.serializers",
>>> "kafka.serializers.KafkaAvroDeserializer,
>>> org.apache.kafka.common.serialization.Serializer");
>> 
>> Updated now
>> 
>> Thanks
>> Omnia
>> 
>>> On 12 Aug 2025, at 18:09, Jun Rao <ju...@confluent.io.INVALID> wrote:
>>> 
>>> Hi, Omnia,
>>> 
>>> Thanks for the reply. A few more comments.
>>> 
>>> JR20. The inconsistency is still there. It's probably simpler to
>>> consolidate on
>>> org.apache.kafka.common.serialization.largemessage.PayloadStore.
>>> 
>>> JR21. Could you document how the payload id string is encoded? For
>> example,
>>> is it always UTF8?
>>> 
>>> JR22. "Since compacted topics can retain data indefinitely, users must
>>> choose between setting a business-appropriate TTL or accepting indefinite
>>> storage costs." Setting a TTL is not ideal since it can break the
>> consumer
>>> application. So, we probably don't want to recommend it to the users.
>>> Another option is to run a cleaning job by consuming the topic and if it
>>> observes a null payload, delete the object based on the key in the
>> record.
>>> 
>>> JR23. We probably don't need to force the PayloadStore implementer to
>>> implement Monitorable. We could document that if it's desirable to
>> monitor
>>> a PayloadStore plugin, one can implement the Monitorable interface.
>>> 
>>> JR24. In the following, org.apache.kafka.common.serialization.Serializer
>>> should include the largemessage namespace.
>>> producerConfig.put("value.serializers",
>>> "kafka.serializers.KafkaAvroDeserializer,
>>> org.apache.kafka.common.serialization.Serializer");
>>> 
>>> Jun
>>> 
>>> On Tue, Aug 12, 2025 at 3:27 AM Omnia Ibrahim <o.g.h.ibrahim@gmail.com
>> <ma...@gmail.com>>
>>> wrote:
>>> 
>>>> Hi Jun,
>>>> 
>>>>> JR20. Could you fix the inconsistency?
>>>>> org.apache.kafka.common.serialization.largemessage.PayloadStore vs
>>>>> org.apache.kafka.common.serialization.largemessage.store.PayloadStore
>>>> Updated
>>>>> 
>>>>> JR21. "Encapsulate that ID into a simple Kafka event using a structured
>>>>> format.". Could you define the structured format explicitly?
>>>> 
>>>> This was from the initial design we had Id format but after discussing
>>>> this with Luke I ended up simplifying this by removing the format and I
>>>> guess I forgot this line. I have updated the kip now.
>>>> 
>>>>> JR22. TTL for object stores: Could the new serializer be used on a
>>>>> compacted topic? If so, how should a user configure the TTL since the
>>>> Kafka
>>>>> retention is infinite.
>>>> 
>>>> Great question about compacted topics! There are few usage of compacted
>>>> topic to consider:
>>>> 1. Typical Compacted Topic Usage: The compacted topics are commonly used
>>>> as cache/state layers where the latest value for each key represents the
>>>> current state. These use cases typically involve smaller records since
>> they
>>>> need to be loaded efficiently for state reconstruction. Large message
>>>> serialization would be unusual for this pattern.
>>>> 
>>>> 2. User-Defined Compacted Topics with Large Payloads: for applications
>>>> that do use compacted topics with large payloads, the PayloadStore
>>>> implementation should handle this by:
>>>>       a. Consistent ID Generation: Use deterministic IDs based on the
>>>> Kafka key (rather than random UUIDs) so that when a payload id is
>> updated,
>>>> it overwrites the same payload store object instead of creating new
>> ones.
>>>> 
>>>>       b. TTL Strategy: Since compacted topics can retain data
>>>> indefinitely, users have two options:
>>>>               - Set a business-appropriate TTL (basically "we know our
>>>> cache data becomes stale after 30 days”)
>>>>               - Configure no TTL and accept indefinite storage costs as
>>>> a trade-off for the architectural benefits
>>>> 
>>>> 3. Compact + Delete Policy: Topics with `cleanup.policy=compact,delete`
>>>> will eventually remove old data, so standard TTL approaches work
>> normally.
>>>> The key insight is that TTL configuration depends on the business
>>>> requirements and usage patterns rather than just the Kafka retention
>>>> policy. The PayloadStore implementation should provide flexibility for
>>>> users to make this trade-off consciously.
>>>> 
>>>> I have update the consideration section with this.
>>>> 
>>>> Thanks
>>>> Omnia
>>>> 
>>>>> On 11 Aug 2025, at 23:32, Jun Rao <ju...@confluent.io.INVALID> wrote:
>>>>> 
>>>>> Hi, Omnia,
>>>>> 
>>>>> Thanks for the reply. A few more comments.
>>>>> 
>>>>> JR20. Could you fix the inconsistency?
>>>>> org.apache.kafka.common.serialization.largemessage.PayloadStore vs
>>>>> org.apache.kafka.common.serialization.largemessage.store.PayloadStore
>>>>> 
>>>>> JR21. "Encapsulate that ID into a simple Kafka event using a structured
>>>>> format.". Could you define the structured format explicitly?
>>>>> 
>>>>> JR22. TTL for object stores: Could the new serializer be used on a
>>>>> compacted topic? If so, how should a user configure the TTL since the
>>>> Kafka
>>>>> retention is infinite.
>>>>> 
>>>>> Jun
>>>>> 
>>>>> On Mon, Aug 11, 2025 at 10:08 AM Omnia Ibrahim <
>> o.g.h.ibrahim@gmail.com
>>>> <ma...@gmail.com>>
>>>>> wrote:
>>>>> 
>>>>>> Hi Jun
>>>>>>> JR11. Do you think that we need to add key.serializers and
>>>>>>> key.deserializers or do you think covering large messages in value is
>>>>>>> enough?
>>>>>> The large message issue usually is a value issue, I never saw a key
>> that
>>>>>> is bigger than broker message size. If I would add key.serializers and
>>>>>> key.deserializers it would be for consistency and maybe there are
>>>> use-cases
>>>>>> where developers want to apply multiple serializations in order on the
>>>> key
>>>>>> as well outside the context of the large message support.
>>>>>> I updated the KIP to add these two configs now.
>>>>>> 
>>>>>>> JR12. We can load ComposableSerializer automatically if
>>>> value.serializers
>>>>>>> or value.deserializers are specified. But, it seems that we could
>> keep
>>>>>>> ComposableSerializer as an internal implementation. For example,
>>>>>>> ProducerInterceptors is automatically loaded when multiple
>> interceptors
>>>>>> are
>>>>>>> specified and is an internal class.
>>>>>> I updated the kip to highlight that the changes is updating
>>>> ProducerConfig
>>>>>> and ConsumerConfig to have new config for serializers/deserializers
>> and
>>>> not
>>>>>> the actual class this is implementation details and not public
>>>> interfaces.
>>>>>> 
>>>>>>> JR17. We could estimate the size after compression, but the estimator
>>>> is
>>>>>>> not 100% accurate. It seems that it's simpler to just use the
>> original
>>>>>>> message size.
>>>>>> We can keep it as original message size, I was thinking if it is good
>>>>>> enough for max.request.size it might be good enough for this. I
>> updated
>>>> the
>>>>>> KIP anyway to simplify it and keep it to the original size.
>>>>>> 
>>>>>> Hope the final version addressed all feedbacks and we can resume with
>>>> the
>>>>>> voting
>>>>>> 
>>>>>> Thanks
>>>>>> Omnia
>>>>>> 
>>>>>>> On 8 Aug 2025, at 22:10, Jun Rao <ju...@confluent.io.INVALID> wrote:
>>>>>>> 
>>>>>>> Hi, Omnia,
>>>>>>> 
>>>>>>> Thanks for the reply.
>>>>>>> 
>>>>>>> JR11. Do you think that we need to add key.serializers and
>>>>>>> key.deserializers or do you think covering large messages in value is
>>>>>>> enough?
>>>>>>> 
>>>>>>> JR12. We can load ComposableSerializer automatically if
>>>> value.serializers
>>>>>>> or value.deserializers are specified. But, it seems that we could
>> keep
>>>>>>> ComposableSerializer as an internal implementation. For example,
>>>>>>> ProducerInterceptors is automatically loaded when multiple
>> interceptors
>>>>>> are
>>>>>>> specified and is an internal class.
>>>>>>> 
>>>>>>> JR17. We could estimate the size after compression, but the estimator
>>>> is
>>>>>>> not 100% accurate. It seems that it's simpler to just use the
>> original
>>>>>>> message size.
>>>>>>> 
>>>>>>> Jun
>>>>>>> 
>>>>>>> On Mon, Aug 4, 2025 at 10:33 AM Omnia Ibrahim <
>> o.g.h.ibrahim@gmail.com
>>>>>> <ma...@gmail.com>>
>>>>>>> wrote:
>>>>>>> 
>>>>>>>> 
>>>>>>>> Hi Jun
>>>>>>>>> JR11. value.serializers and value.deserializers: Should they be of
>>>> type
>>>>>>>>> List? Also, where are key.serializers and key.deserializers?
>>>>>>>>> 
>>>>>>>> Updated now
>>>>>>>>> JR12. Do we still need ComposableSerializer and
>>>> ComposableDeserializer?
>>>>>>>> The initial thinking here was if `value.serializers or
>>>>>> value.deserializers
>>>>>>>> ` exists the client will load  ComposableSerializer or
>>>>>>>> ComposableDeserializer  automatically and use them. Unfortunately
>> this
>>>>>>>> would need us to define these serializers.
>>>>>>>> 
>>>>>>>> The other option is to update `Plugin<Serializer<V>>
>>>>>>>> valueSerializerPlugin` and KafkaProducer constructor to accept
>>>>>>>> List<Serializer<V>> and move the logic of the ComposableSerializer
>>>> into
>>>>>>>> KafkaProducer::doSend which when we do serialization (same with
>>>>>>>> KafkaConsumer). This option hide the logic and reduce exposure for
>>>>>> client
>>>>>>>> to these.
>>>>>>>> WDYT?
>>>>>>>> 
>>>>>>>>> 
>>>>>>>>> JR13. large.message.payload.store.class : should it be of type
>> class?
>>>>>>>> Updated
>>>>>>>>> 
>>>>>>>>> JR14.
>>>>>>>>> 
>>>>>>>> 
>>>>>> 
>>>> 
>> org.apache.kafka.common.serialization.largemessage.LargeMessageSerializer :
>>>>>>>>> The name seems redundant since largemessage appears twice.
>>>>>>>> Updated
>>>>>>>>> 
>>>>>>>>> JR15. PayloadResponse: It still mentions response code. It mentions
>>>>>>>>> "isRetryable flag", which no longer exists in
>> PayloadStoreException.
>>>>>>>> There
>>>>>>>>> are typos in "then it will serialiser will”.
>>>>>>>> The KIP is updated now
>>>>>>>> 
>>>>>>>>> JR16. Regarding returning new byte[0] if
>>>>>>>> large.message.skip.not.found.error
>>>>>>>>> is true, this will likely fail the next deserializer and the
>>>>>> application
>>>>>>>>> won't have the right context of the error. It's probably better to
>>>> just
>>>>>>>>> propagate the specific exception and let the caller handle it.
>>>>>>>> You right this will cause issue if there is another deserializer
>>>> waiting
>>>>>>>> for the data. I have updated the KIP with this.
>>>>>>>> 
>>>>>>>>> 
>>>>>>>>> JR17. LargeMessageSerializer:  "Check if the estimated size of the
>>>> data
>>>>>>>>> (bytes) after applying provided compression (if there is one)"
>>>>>>>> Compression
>>>>>>>>> actually happens after serialization and is done on a batch of
>>>> records.
>>>>>>>> Yes the compression itself happened after but we also have
>>>>>>>> `estimateSizeInBytesUpperBound` which my understanding is this take
>>>> the
>>>>>>>> compression type into the account as well
>>>>>>>>> 
>>>>>>>>> 
>>>>>>>>> JR18. Could you define the type T for LargeMessageSerializer?
>>>>>>>> Update the KIP this T would be byte[]
>>>>>>>> 
>>>>>>>> Thanks
>>>>>>>> Omnia
>>>>>>>>> 
>>>>>>>>> Jun
>>>>>>>>> 
>>>>>>>>> On Fri, Jul 25, 2025 at 6:28 AM Omnia Ibrahim <
>>>> o.g.h.ibrahim@gmail.com
>>>>>>>> <ma...@gmail.com>>
>>>>>>>>> wrote:
>>>>>>>>> 
>>>>>>>>>> Hi Jun, thanks for having the time to review this
>>>>>>>>>> 
>>>>>>>>>>> JR1. While the KIP is potentially useful, I am wondering who is
>>>>>>>>>> responsible
>>>>>>>>>>> for retention for the objects in the payload store. Once a
>> message
>>>>>>>> with a
>>>>>>>>>>> reference is deleted, the key of the external object is lost and
>>>> the
>>>>>>>>>> object
>>>>>>>>>>> may never be deleted.
>>>>>>>>>> 
>>>>>>>>>> The `ttl` in the object store is the responsibility of the owner
>> of
>>>>>> this
>>>>>>>>>> store it should be configured in away that is reasonable with the
>>>>>>>> retention
>>>>>>>>>> config in Kafka.
>>>>>>>>>> I have updated the KIP with `Consideration` section.
>>>>>>>>>>> 
>>>>>>>>>>> JR2. Configs: For all new configs, it would be useful to list
>> their
>>>>>>>>>> types.
>>>>>>>>>> Updated the KIP now
>>>>>>>>>>> 
>>>>>>>>>>> JR3. value.serializers: Why is this required? If a user doesn't
>> set
>>>>>> it,
>>>>>>>>>> we
>>>>>>>>>>> should just use value.serializer, right? Ditto for
>> key.serializers.
>>>>>>>>>> No you right this was copy/past mistake
>>>>>>>>>> 
>>>>>>>>>>> 
>>>>>>>>>>> JR4. For all new public interfaces such as
>> LargeMessageSerializer,
>>>>>>>>>>> PayloadStore and PayloadResponse, it would be useful to include
>> the
>>>>>>>> full
>>>>>>>>>>> package name.
>>>>>>>>>> Updated the KIP now
>>>>>>>>>>> 
>>>>>>>>>>> JR5. large.message.payload.store.retry.max.backoff.ms and
>>>>>>>>>>> large.message.payload.store.retry.delay.backoff.ms: Is the
>>>> intention
>>>>>>>> to
>>>>>>>>>>> implement exponential backoff on retries? If so, it's more
>>>> consistent
>>>>>>>> if
>>>>>>>>>> we
>>>>>>>>>>> can follow the existing naming convention like
>>>> retry.backoff.max.ms
>>>>>> <
>>>>>>>>>> http://retry.backoff.max.ms/> and
>>>>>>>>>>> retry.backoff.ms <http://retry.backoff.ms/> <
>>>>>> http://retry.backoff.ms/> <http://retry.backoff.ms/
>>>>>>>>> .
>>>>>>>>>> I have removed these to simplify the config more (as Luke
>> suggested
>>>>>>>>>> initially) and added these to the consideration section.
>>>>>>>>>> 
>>>>>>>>>>> 
>>>>>>>>>>> JR6. large.message.skip.not.found.error : If the reference can't
>> be
>>>>>>>>>> found,
>>>>>>>>>>> what value does the deserializer return? Note that null has a
>>>> special
>>>>>>>>>>> meaning for tombstone in compacted topics.
>>>>>>>>>> The deserialiser will return `new byte[0]` not null.
>>>>>>>>>>> 
>>>>>>>>>>> JR7. PayloadResponse: Why do we have both responseCode and
>>>>>>>>>>> PayloadStoreException?
>>>>>>>>>> We can do without responseCode, the initial though was to report
>>>>>>>> response
>>>>>>>>>> code form payload store.
>>>>>>>>>> Update the KIP.
>>>>>>>>>>> JR8. Why do we need PayloadStore.metrics? Note that we could
>>>> monitor
>>>>>>>> the
>>>>>>>>>>> metrics in a plugin through the Monitorable interface.
>>>>>>>>>> Oh nice, I didn’t know about this interface before. Updated the
>> KIP
>>>>>> with
>>>>>>>>>> this now.
>>>>>>>>>>> 
>>>>>>>>>>> JR9. Why do we need the protected field
>>>>>>>>>> PayloadStoreException.isRetryable?
>>>>>>>>>> Initial thought here was the serializer can retry the upload. But
>> I
>>>>>> have
>>>>>>>>>> removed all the retry logic from serializer and it will be up to
>> the
>>>>>>>>>> PayloadStore provider to implement this if they need it.
>>>>>>>>>>> 
>>>>>>>>>>> JR10. As Luke mentioned earlier, we could turn PayloadStore to an
>>>>>>>>>> interface.
>>>>>>>>>> It is updated now to interface.
>>>>>>>>>> 
>>>>>>>>>> Hope the last version of the KIP is more simpler now
>>>>>>>>>> 
>>>>>>>>>> Thanks
>>>>>>>>>> Omnia
>>>>>>>>>> 
>>>>>>>>>>> On 23 Jul 2025, at 00:43, Jun Rao <jun@confluent.io.INVALID
>>>> <ma...@confluent.io.INVALID> <mailto:
>>>>>> jun@confluent.io.INVALID <ma...@confluent.io.INVALID> <mailto:
>> jun@confluent.io.INVALID>> <mailto:
>>>>>>>> jun@confluent.io.INVALID <ma...@confluent.io.INVALID> <ma...@confluent.io.INVALID> <mailto:
>> jun@confluent.io.INVALID <ma...@confluent.io.INVALID>> <mailto:
>>>> jun@confluent.io.INVALID <ma...@confluent.io.INVALID> <ma...@confluent.io.INVALID>>>> wrote:
>>>>>>>>>>> 
>>>>>>>>>>> Thanks for the KIP. A few comments.
>>>>>>>>>>> 
>>>>>>>>>>> JR1. While the KIP is potentially useful, I am wondering who is
>>>>>>>>>> responsible
>>>>>>>>>>> for retention for the objects in the payload store. Once a
>> message
>>>>>>>> with a
>>>>>>>>>>> reference is deleted, the key of the external object is lost and
>>>> the
>>>>>>>>>> object
>>>>>>>>>>> may never be deleted.
>>>>>>>>>>> 
>>>>>>>>>>> JR2. Configs: For all new configs, it would be useful to list
>> their
>>>>>>>>>> types.
>>>>>>>>>>> 
>>>>>>>>>>> JR3. value.serializers: Why is this required? If a user doesn't
>> set
>>>>>> it,
>>>>>>>>>> we
>>>>>>>>>>> should just use value.serializer, right? Ditto for
>> key.serializers.
>>>>>>>>>>> 
>>>>>>>>>>> JR4. For all new public interfaces such as
>> LargeMessageSerializer,
>>>>>>>>>>> PayloadStore and PayloadResponse, it would be useful to include
>> the
>>>>>>>> full
>>>>>>>>>>> package name.
>>>>>>>>>>> 
>>>>>>>>>>> JR5. large.message.payload.store.retry.max.backoff.ms and
>>>>>>>>>>> large.message.payload.store.retry.delay.backoff.ms: Is the
>>>> intention
>>>>>>>> to
>>>>>>>>>>> implement exponential backoff on retries? If so, it's more
>>>> consistent
>>>>>>>> if
>>>>>>>>>> we
>>>>>>>>>>> can follow the existing naming convention like
>>>> retry.backoff.max.ms <http://retry.backoff.max.ms/> <http://retry.backoff.max.ms/> <
>> http://retry.backoff.max.ms/>
>>>>>> <http://retry.backoff.max.ms/> <
>>>>>>>> http://retry.backoff.max.ms/> <
>>>>>>>>>> http://retry.backoff.max.ms/> and
>>>>>>>>>>> retry.backoff.ms <http://retry.backoff.ms/> <http://retry.backoff.ms/> <
>> http://retry.backoff.ms/> <
>>>> http://retry.backoff.ms/> <
>>>>>> http://retry.backoff.ms/> <http://retry.backoff.ms/
>>>>>>>>> .
>>>>>>>>>>> 
>>>>>>>>>>> JR6. large.message.skip.not.found.error : If the reference can't
>> be
>>>>>>>>>> found,
>>>>>>>>>>> what value does the deserializer return? Note that null has a
>>>> special
>>>>>>>>>>> meaning for tombstone in compacted topics.
>>>>>>>>>>> 
>>>>>>>>>>> JR7. PayloadResponse: Why do we have both responseCode and
>>>>>>>>>>> PayloadStoreException?
>>>>>>>>>>> 
>>>>>>>>>>> JR8. Why do we need PayloadStore.metrics? Note that we could
>>>> monitor
>>>>>>>> the
>>>>>>>>>>> metrics in a plugin through the Monitorable interface.
>>>>>>>>>>> 
>>>>>>>>>>> JR9. Why do we need the protected field
>>>>>>>>>> PayloadStoreException.isRetryable?
>>>>>>>>>>> 
>>>>>>>>>>> JR10. As Luke mentioned earlier, we could turn PayloadStore to an
>>>>>>>>>> interface.
>>>>>>>>>>> 
>>>>>>>>>>> Thanks,

Re: [DISCUSS] KIP-1159: Large message reference based Serializer

[Greg Harris <gr...@aiven.io.INVALID>]

Hi Omnia,

Thanks for the KIP!

I had some questions about the rejected alternatives:

1. The motivation for the KIP includes this:
> Increasing message size indefinitely is not a viable solution as it can
lead to performance degradation, memory issues, and instability of the
broker.

Could you discuss the rejected alternative of improving Kafka's handling of
large messages in-band? If Kafka is better at handling large messages, then
it may not be necessary to adopt the complexity of reference-based
messaging.
I wonder if these problems truly disappear with referenced based messaging,
or if they are moved from broker to applications, where the Apache Kafka
project isn't blamed but also has less ability to address. Client
applications may have worse memory management or observability than Kafka
brokers.
For example, I think that KIP-1163 may benefit large-message workloads,
because it allows for the time-to-commit a batch to be decoupled from the
time-to-write the batch.

2. The "popular patterns" section includes this:
> Available open-source implementation
> There isn't a specific open-source implementation for Reference-Based
Messaging. Mostly everyone adopting this has their custom solution

Could there be another rejected alternative to open-source an existing
reference-based-messaging solution? If there are community members with
awareness of an existing custom solution, they could also influence
open-sourcing it.

3. This is listed as a rejected alternative:
> Implement this as a separate project outside of Apache Kafka as this
seems to be a pattern that needs more use cases and it would be better to
have this in Apache Kafka as native implementation instead of a separate
project.

AFAIU this proposal still requires implementations of the new interfaces to
be in a separate project outside of Apache Kafka, because we cannot depend
on object storage implementations in Apache Kafka.
Such a project could own the new interfaces itself, either as private
interfaces or public API it exposes to further downstream projects. This
will have fewer restrictions on the API design, as a vote on Apache Kafka
is not necessary to evolve the functionality.

Can you clarify what the benefit is for this change to land in Apache Kafka
specifically? And if there is a group ready to develop and distribute
open-source implementations of this interface? Otherwise we will end up
back where we started with only closed-source implementations.

Thanks,
Greg

On Wed, Aug 13, 2025 at 5:27 AM Omnia Ibrahim <o....@gmail.com>
wrote:

> Hi Jun
> >
> > JR20. The inconsistency is still there. It's probably simpler to
> > consolidate on
> > org.apache.kafka.common.serialization.largemessage.PayloadStore.
> Updated the kip now to address these
> > JR21. Could you document how the payload id string is encoded? For
> example,
> > is it always UTF8?
> I updated the java doc, the default implementation do provide UTF-8
> however if the developer want to support any other they will need to
> address this in the implementation of the store.
> >
> > JR22. "Since compacted topics can retain data indefinitely, users must
> > choose between setting a business-appropriate TTL or accepting indefinite
> > storage costs." Setting a TTL is not ideal since it can break the
> consumer
> > application. So, we probably don't want to recommend it to the users.
> > Another option is to run a cleaning job by consuming the topic and if it
> > observes a null payload, delete the object based on the key in the
> record.
>
> Updated the recommendations
> > JR24. In the following, org.apache.kafka.common.serialization.Serializer
> > should include the largemessage namespace.
> > producerConfig.put("value.serializers",
> > "kafka.serializers.KafkaAvroDeserializer,
> > org.apache.kafka.common.serialization.Serializer");
>
> Updated now
>
> Thanks
> Omnia
>
> > On 12 Aug 2025, at 18:09, Jun Rao <ju...@confluent.io.INVALID> wrote:
> >
> > Hi, Omnia,
> >
> > Thanks for the reply. A few more comments.
> >
> > JR20. The inconsistency is still there. It's probably simpler to
> > consolidate on
> > org.apache.kafka.common.serialization.largemessage.PayloadStore.
> >
> > JR21. Could you document how the payload id string is encoded? For
> example,
> > is it always UTF8?
> >
> > JR22. "Since compacted topics can retain data indefinitely, users must
> > choose between setting a business-appropriate TTL or accepting indefinite
> > storage costs." Setting a TTL is not ideal since it can break the
> consumer
> > application. So, we probably don't want to recommend it to the users.
> > Another option is to run a cleaning job by consuming the topic and if it
> > observes a null payload, delete the object based on the key in the
> record.
> >
> > JR23. We probably don't need to force the PayloadStore implementer to
> > implement Monitorable. We could document that if it's desirable to
> monitor
> > a PayloadStore plugin, one can implement the Monitorable interface.
> >
> > JR24. In the following, org.apache.kafka.common.serialization.Serializer
> > should include the largemessage namespace.
> > producerConfig.put("value.serializers",
> > "kafka.serializers.KafkaAvroDeserializer,
> > org.apache.kafka.common.serialization.Serializer");
> >
> > Jun
> >
> > On Tue, Aug 12, 2025 at 3:27 AM Omnia Ibrahim <o.g.h.ibrahim@gmail.com
> <ma...@gmail.com>>
> > wrote:
> >
> >> Hi Jun,
> >>
> >>> JR20. Could you fix the inconsistency?
> >>> org.apache.kafka.common.serialization.largemessage.PayloadStore vs
> >>> org.apache.kafka.common.serialization.largemessage.store.PayloadStore
> >> Updated
> >>>
> >>> JR21. "Encapsulate that ID into a simple Kafka event using a structured
> >>> format.". Could you define the structured format explicitly?
> >>
> >> This was from the initial design we had Id format but after discussing
> >> this with Luke I ended up simplifying this by removing the format and I
> >> guess I forgot this line. I have updated the kip now.
> >>
> >>> JR22. TTL for object stores: Could the new serializer be used on a
> >>> compacted topic? If so, how should a user configure the TTL since the
> >> Kafka
> >>> retention is infinite.
> >>
> >> Great question about compacted topics! There are few usage of compacted
> >> topic to consider:
> >> 1. Typical Compacted Topic Usage: The compacted topics are commonly used
> >> as cache/state layers where the latest value for each key represents the
> >> current state. These use cases typically involve smaller records since
> they
> >> need to be loaded efficiently for state reconstruction. Large message
> >> serialization would be unusual for this pattern.
> >>
> >> 2. User-Defined Compacted Topics with Large Payloads: for applications
> >> that do use compacted topics with large payloads, the PayloadStore
> >> implementation should handle this by:
> >>        a. Consistent ID Generation: Use deterministic IDs based on the
> >> Kafka key (rather than random UUIDs) so that when a payload id is
> updated,
> >> it overwrites the same payload store object instead of creating new
> ones.
> >>
> >>        b. TTL Strategy: Since compacted topics can retain data
> >> indefinitely, users have two options:
> >>                - Set a business-appropriate TTL (basically "we know our
> >> cache data becomes stale after 30 days”)
> >>                - Configure no TTL and accept indefinite storage costs as
> >> a trade-off for the architectural benefits
> >>
> >> 3. Compact + Delete Policy: Topics with `cleanup.policy=compact,delete`
> >> will eventually remove old data, so standard TTL approaches work
> normally.
> >> The key insight is that TTL configuration depends on the business
> >> requirements and usage patterns rather than just the Kafka retention
> >> policy. The PayloadStore implementation should provide flexibility for
> >> users to make this trade-off consciously.
> >>
> >> I have update the consideration section with this.
> >>
> >> Thanks
> >> Omnia
> >>
> >>> On 11 Aug 2025, at 23:32, Jun Rao <ju...@confluent.io.INVALID> wrote:
> >>>
> >>> Hi, Omnia,
> >>>
> >>> Thanks for the reply. A few more comments.
> >>>
> >>> JR20. Could you fix the inconsistency?
> >>> org.apache.kafka.common.serialization.largemessage.PayloadStore vs
> >>> org.apache.kafka.common.serialization.largemessage.store.PayloadStore
> >>>
> >>> JR21. "Encapsulate that ID into a simple Kafka event using a structured
> >>> format.". Could you define the structured format explicitly?
> >>>
> >>> JR22. TTL for object stores: Could the new serializer be used on a
> >>> compacted topic? If so, how should a user configure the TTL since the
> >> Kafka
> >>> retention is infinite.
> >>>
> >>> Jun
> >>>
> >>> On Mon, Aug 11, 2025 at 10:08 AM Omnia Ibrahim <
> o.g.h.ibrahim@gmail.com
> >> <ma...@gmail.com>>
> >>> wrote:
> >>>
> >>>> Hi Jun
> >>>>> JR11. Do you think that we need to add key.serializers and
> >>>>> key.deserializers or do you think covering large messages in value is
> >>>>> enough?
> >>>> The large message issue usually is a value issue, I never saw a key
> that
> >>>> is bigger than broker message size. If I would add key.serializers and
> >>>> key.deserializers it would be for consistency and maybe there are
> >> use-cases
> >>>> where developers want to apply multiple serializations in order on the
> >> key
> >>>> as well outside the context of the large message support.
> >>>> I updated the KIP to add these two configs now.
> >>>>
> >>>>> JR12. We can load ComposableSerializer automatically if
> >> value.serializers
> >>>>> or value.deserializers are specified. But, it seems that we could
> keep
> >>>>> ComposableSerializer as an internal implementation. For example,
> >>>>> ProducerInterceptors is automatically loaded when multiple
> interceptors
> >>>> are
> >>>>> specified and is an internal class.
> >>>> I updated the kip to highlight that the changes is updating
> >> ProducerConfig
> >>>> and ConsumerConfig to have new config for serializers/deserializers
> and
> >> not
> >>>> the actual class this is implementation details and not public
> >> interfaces.
> >>>>
> >>>>> JR17. We could estimate the size after compression, but the estimator
> >> is
> >>>>> not 100% accurate. It seems that it's simpler to just use the
> original
> >>>>> message size.
> >>>> We can keep it as original message size, I was thinking if it is good
> >>>> enough for max.request.size it might be good enough for this. I
> updated
> >> the
> >>>> KIP anyway to simplify it and keep it to the original size.
> >>>>
> >>>> Hope the final version addressed all feedbacks and we can resume with
> >> the
> >>>> voting
> >>>>
> >>>> Thanks
> >>>> Omnia
> >>>>
> >>>>> On 8 Aug 2025, at 22:10, Jun Rao <ju...@confluent.io.INVALID> wrote:
> >>>>>
> >>>>> Hi, Omnia,
> >>>>>
> >>>>> Thanks for the reply.
> >>>>>
> >>>>> JR11. Do you think that we need to add key.serializers and
> >>>>> key.deserializers or do you think covering large messages in value is
> >>>>> enough?
> >>>>>
> >>>>> JR12. We can load ComposableSerializer automatically if
> >> value.serializers
> >>>>> or value.deserializers are specified. But, it seems that we could
> keep
> >>>>> ComposableSerializer as an internal implementation. For example,
> >>>>> ProducerInterceptors is automatically loaded when multiple
> interceptors
> >>>> are
> >>>>> specified and is an internal class.
> >>>>>
> >>>>> JR17. We could estimate the size after compression, but the estimator
> >> is
> >>>>> not 100% accurate. It seems that it's simpler to just use the
> original
> >>>>> message size.
> >>>>>
> >>>>> Jun
> >>>>>
> >>>>> On Mon, Aug 4, 2025 at 10:33 AM Omnia Ibrahim <
> o.g.h.ibrahim@gmail.com
> >>>> <ma...@gmail.com>>
> >>>>> wrote:
> >>>>>
> >>>>>>
> >>>>>> Hi Jun
> >>>>>>> JR11. value.serializers and value.deserializers: Should they be of
> >> type
> >>>>>>> List? Also, where are key.serializers and key.deserializers?
> >>>>>>>
> >>>>>> Updated now
> >>>>>>> JR12. Do we still need ComposableSerializer and
> >> ComposableDeserializer?
> >>>>>> The initial thinking here was if `value.serializers or
> >>>> value.deserializers
> >>>>>> ` exists the client will load  ComposableSerializer or
> >>>>>> ComposableDeserializer  automatically and use them. Unfortunately
> this
> >>>>>> would need us to define these serializers.
> >>>>>>
> >>>>>> The other option is to update `Plugin<Serializer<V>>
> >>>>>> valueSerializerPlugin` and KafkaProducer constructor to accept
> >>>>>> List<Serializer<V>> and move the logic of the ComposableSerializer
> >> into
> >>>>>> KafkaProducer::doSend which when we do serialization (same with
> >>>>>> KafkaConsumer). This option hide the logic and reduce exposure for
> >>>> client
> >>>>>> to these.
> >>>>>> WDYT?
> >>>>>>
> >>>>>>>
> >>>>>>> JR13. large.message.payload.store.class : should it be of type
> class?
> >>>>>> Updated
> >>>>>>>
> >>>>>>> JR14.
> >>>>>>>
> >>>>>>
> >>>>
> >>
> org.apache.kafka.common.serialization.largemessage.LargeMessageSerializer :
> >>>>>>> The name seems redundant since largemessage appears twice.
> >>>>>> Updated
> >>>>>>>
> >>>>>>> JR15. PayloadResponse: It still mentions response code. It mentions
> >>>>>>> "isRetryable flag", which no longer exists in
> PayloadStoreException.
> >>>>>> There
> >>>>>>> are typos in "then it will serialiser will”.
> >>>>>> The KIP is updated now
> >>>>>>
> >>>>>>> JR16. Regarding returning new byte[0] if
> >>>>>> large.message.skip.not.found.error
> >>>>>>> is true, this will likely fail the next deserializer and the
> >>>> application
> >>>>>>> won't have the right context of the error. It's probably better to
> >> just
> >>>>>>> propagate the specific exception and let the caller handle it.
> >>>>>> You right this will cause issue if there is another deserializer
> >> waiting
> >>>>>> for the data. I have updated the KIP with this.
> >>>>>>
> >>>>>>>
> >>>>>>> JR17. LargeMessageSerializer:  "Check if the estimated size of the
> >> data
> >>>>>>> (bytes) after applying provided compression (if there is one)"
> >>>>>> Compression
> >>>>>>> actually happens after serialization and is done on a batch of
> >> records.
> >>>>>> Yes the compression itself happened after but we also have
> >>>>>> `estimateSizeInBytesUpperBound` which my understanding is this take
> >> the
> >>>>>> compression type into the account as well
> >>>>>>>
> >>>>>>>
> >>>>>>> JR18. Could you define the type T for LargeMessageSerializer?
> >>>>>> Update the KIP this T would be byte[]
> >>>>>>
> >>>>>> Thanks
> >>>>>> Omnia
> >>>>>>>
> >>>>>>> Jun
> >>>>>>>
> >>>>>>> On Fri, Jul 25, 2025 at 6:28 AM Omnia Ibrahim <
> >> o.g.h.ibrahim@gmail.com
> >>>>>> <ma...@gmail.com>>
> >>>>>>> wrote:
> >>>>>>>
> >>>>>>>> Hi Jun, thanks for having the time to review this
> >>>>>>>>
> >>>>>>>>> JR1. While the KIP is potentially useful, I am wondering who is
> >>>>>>>> responsible
> >>>>>>>>> for retention for the objects in the payload store. Once a
> message
> >>>>>> with a
> >>>>>>>>> reference is deleted, the key of the external object is lost and
> >> the
> >>>>>>>> object
> >>>>>>>>> may never be deleted.
> >>>>>>>>
> >>>>>>>> The `ttl` in the object store is the responsibility of the owner
> of
> >>>> this
> >>>>>>>> store it should be configured in away that is reasonable with the
> >>>>>> retention
> >>>>>>>> config in Kafka.
> >>>>>>>> I have updated the KIP with `Consideration` section.
> >>>>>>>>>
> >>>>>>>>> JR2. Configs: For all new configs, it would be useful to list
> their
> >>>>>>>> types.
> >>>>>>>> Updated the KIP now
> >>>>>>>>>
> >>>>>>>>> JR3. value.serializers: Why is this required? If a user doesn't
> set
> >>>> it,
> >>>>>>>> we
> >>>>>>>>> should just use value.serializer, right? Ditto for
> key.serializers.
> >>>>>>>> No you right this was copy/past mistake
> >>>>>>>>
> >>>>>>>>>
> >>>>>>>>> JR4. For all new public interfaces such as
> LargeMessageSerializer,
> >>>>>>>>> PayloadStore and PayloadResponse, it would be useful to include
> the
> >>>>>> full
> >>>>>>>>> package name.
> >>>>>>>> Updated the KIP now
> >>>>>>>>>
> >>>>>>>>> JR5. large.message.payload.store.retry.max.backoff.ms and
> >>>>>>>>> large.message.payload.store.retry.delay.backoff.ms: Is the
> >> intention
> >>>>>> to
> >>>>>>>>> implement exponential backoff on retries? If so, it's more
> >> consistent
> >>>>>> if
> >>>>>>>> we
> >>>>>>>>> can follow the existing naming convention like
> >> retry.backoff.max.ms
> >>>> <
> >>>>>>>> http://retry.backoff.max.ms/> and
> >>>>>>>>> retry.backoff.ms <http://retry.backoff.ms/> <
> >>>> http://retry.backoff.ms/> <http://retry.backoff.ms/
> >>>>>>> .
> >>>>>>>> I have removed these to simplify the config more (as Luke
> suggested
> >>>>>>>> initially) and added these to the consideration section.
> >>>>>>>>
> >>>>>>>>>
> >>>>>>>>> JR6. large.message.skip.not.found.error : If the reference can't
> be
> >>>>>>>> found,
> >>>>>>>>> what value does the deserializer return? Note that null has a
> >> special
> >>>>>>>>> meaning for tombstone in compacted topics.
> >>>>>>>> The deserialiser will return `new byte[0]` not null.
> >>>>>>>>>
> >>>>>>>>> JR7. PayloadResponse: Why do we have both responseCode and
> >>>>>>>>> PayloadStoreException?
> >>>>>>>> We can do without responseCode, the initial though was to report
> >>>>>> response
> >>>>>>>> code form payload store.
> >>>>>>>> Update the KIP.
> >>>>>>>>> JR8. Why do we need PayloadStore.metrics? Note that we could
> >> monitor
> >>>>>> the
> >>>>>>>>> metrics in a plugin through the Monitorable interface.
> >>>>>>>> Oh nice, I didn’t know about this interface before. Updated the
> KIP
> >>>> with
> >>>>>>>> this now.
> >>>>>>>>>
> >>>>>>>>> JR9. Why do we need the protected field
> >>>>>>>> PayloadStoreException.isRetryable?
> >>>>>>>> Initial thought here was the serializer can retry the upload. But
> I
> >>>> have
> >>>>>>>> removed all the retry logic from serializer and it will be up to
> the
> >>>>>>>> PayloadStore provider to implement this if they need it.
> >>>>>>>>>
> >>>>>>>>> JR10. As Luke mentioned earlier, we could turn PayloadStore to an
> >>>>>>>> interface.
> >>>>>>>> It is updated now to interface.
> >>>>>>>>
> >>>>>>>> Hope the last version of the KIP is more simpler now
> >>>>>>>>
> >>>>>>>> Thanks
> >>>>>>>> Omnia
> >>>>>>>>
> >>>>>>>>> On 23 Jul 2025, at 00:43, Jun Rao <jun@confluent.io.INVALID
> >> <ma...@confluent.io.INVALID> <mailto:
> >>>> jun@confluent.io.INVALID <ma...@confluent.io.INVALID> <mailto:
> jun@confluent.io.INVALID>> <mailto:
> >>>>>> jun@confluent.io.INVALID <ma...@confluent.io.INVALID> <mailto:
> jun@confluent.io.INVALID> <mailto:
> >> jun@confluent.io.INVALID <ma...@confluent.io.INVALID>>>> wrote:
> >>>>>>>>>
> >>>>>>>>> Thanks for the KIP. A few comments.
> >>>>>>>>>
> >>>>>>>>> JR1. While the KIP is potentially useful, I am wondering who is
> >>>>>>>> responsible
> >>>>>>>>> for retention for the objects in the payload store. Once a
> message
> >>>>>> with a
> >>>>>>>>> reference is deleted, the key of the external object is lost and
> >> the
> >>>>>>>> object
> >>>>>>>>> may never be deleted.
> >>>>>>>>>
> >>>>>>>>> JR2. Configs: For all new configs, it would be useful to list
> their
> >>>>>>>> types.
> >>>>>>>>>
> >>>>>>>>> JR3. value.serializers: Why is this required? If a user doesn't
> set
> >>>> it,
> >>>>>>>> we
> >>>>>>>>> should just use value.serializer, right? Ditto for
> key.serializers.
> >>>>>>>>>
> >>>>>>>>> JR4. For all new public interfaces such as
> LargeMessageSerializer,
> >>>>>>>>> PayloadStore and PayloadResponse, it would be useful to include
> the
> >>>>>> full
> >>>>>>>>> package name.
> >>>>>>>>>
> >>>>>>>>> JR5. large.message.payload.store.retry.max.backoff.ms and
> >>>>>>>>> large.message.payload.store.retry.delay.backoff.ms: Is the
> >> intention
> >>>>>> to
> >>>>>>>>> implement exponential backoff on retries? If so, it's more
> >> consistent
> >>>>>> if
> >>>>>>>> we
> >>>>>>>>> can follow the existing naming convention like
> >> retry.backoff.max.ms <http://retry.backoff.max.ms/> <
> http://retry.backoff.max.ms/>
> >>>> <http://retry.backoff.max.ms/> <
> >>>>>> http://retry.backoff.max.ms/> <
> >>>>>>>> http://retry.backoff.max.ms/> and
> >>>>>>>>> retry.backoff.ms <http://retry.backoff.ms/> <
> http://retry.backoff.ms/> <
> >> http://retry.backoff.ms/> <
> >>>> http://retry.backoff.ms/> <http://retry.backoff.ms/
> >>>>>>> .
> >>>>>>>>>
> >>>>>>>>> JR6. large.message.skip.not.found.error : If the reference can't
> be
> >>>>>>>> found,
> >>>>>>>>> what value does the deserializer return? Note that null has a
> >> special
> >>>>>>>>> meaning for tombstone in compacted topics.
> >>>>>>>>>
> >>>>>>>>> JR7. PayloadResponse: Why do we have both responseCode and
> >>>>>>>>> PayloadStoreException?
> >>>>>>>>>
> >>>>>>>>> JR8. Why do we need PayloadStore.metrics? Note that we could
> >> monitor
> >>>>>> the
> >>>>>>>>> metrics in a plugin through the Monitorable interface.
> >>>>>>>>>
> >>>>>>>>> JR9. Why do we need the protected field
> >>>>>>>> PayloadStoreException.isRetryable?
> >>>>>>>>>
> >>>>>>>>> JR10. As Luke mentioned earlier, we could turn PayloadStore to an
> >>>>>>>> interface.
> >>>>>>>>>
> >>>>>>>>> Thanks,
>
>

Re: [DISCUSS] KIP-1159: Large message reference based Serializer

[Omnia Ibrahim <o....@gmail.com>]

Hi Jun
> 
> JR20. The inconsistency is still there. It's probably simpler to
> consolidate on
> org.apache.kafka.common.serialization.largemessage.PayloadStore.
Updated the kip now to address these
> JR21. Could you document how the payload id string is encoded? For example,
> is it always UTF8?
I updated the java doc, the default implementation do provide UTF-8 however if the developer want to support any other they will need to address this in the implementation of the store. 
> 
> JR22. "Since compacted topics can retain data indefinitely, users must
> choose between setting a business-appropriate TTL or accepting indefinite
> storage costs." Setting a TTL is not ideal since it can break the consumer
> application. So, we probably don't want to recommend it to the users.
> Another option is to run a cleaning job by consuming the topic and if it
> observes a null payload, delete the object based on the key in the record.

Updated the recommendations 
> JR24. In the following, org.apache.kafka.common.serialization.Serializer
> should include the largemessage namespace.
> producerConfig.put("value.serializers",
> "kafka.serializers.KafkaAvroDeserializer,
> org.apache.kafka.common.serialization.Serializer");

Updated now

Thanks 
Omnia

> On 12 Aug 2025, at 18:09, Jun Rao <ju...@confluent.io.INVALID> wrote:
> 
> Hi, Omnia,
> 
> Thanks for the reply. A few more comments.
> 
> JR20. The inconsistency is still there. It's probably simpler to
> consolidate on
> org.apache.kafka.common.serialization.largemessage.PayloadStore.
> 
> JR21. Could you document how the payload id string is encoded? For example,
> is it always UTF8?
> 
> JR22. "Since compacted topics can retain data indefinitely, users must
> choose between setting a business-appropriate TTL or accepting indefinite
> storage costs." Setting a TTL is not ideal since it can break the consumer
> application. So, we probably don't want to recommend it to the users.
> Another option is to run a cleaning job by consuming the topic and if it
> observes a null payload, delete the object based on the key in the record.
> 
> JR23. We probably don't need to force the PayloadStore implementer to
> implement Monitorable. We could document that if it's desirable to monitor
> a PayloadStore plugin, one can implement the Monitorable interface.
> 
> JR24. In the following, org.apache.kafka.common.serialization.Serializer
> should include the largemessage namespace.
> producerConfig.put("value.serializers",
> "kafka.serializers.KafkaAvroDeserializer,
> org.apache.kafka.common.serialization.Serializer");
> 
> Jun
> 
> On Tue, Aug 12, 2025 at 3:27 AM Omnia Ibrahim <o.g.h.ibrahim@gmail.com <ma...@gmail.com>>
> wrote:
> 
>> Hi Jun,
>> 
>>> JR20. Could you fix the inconsistency?
>>> org.apache.kafka.common.serialization.largemessage.PayloadStore vs
>>> org.apache.kafka.common.serialization.largemessage.store.PayloadStore
>> Updated
>>> 
>>> JR21. "Encapsulate that ID into a simple Kafka event using a structured
>>> format.". Could you define the structured format explicitly?
>> 
>> This was from the initial design we had Id format but after discussing
>> this with Luke I ended up simplifying this by removing the format and I
>> guess I forgot this line. I have updated the kip now.
>> 
>>> JR22. TTL for object stores: Could the new serializer be used on a
>>> compacted topic? If so, how should a user configure the TTL since the
>> Kafka
>>> retention is infinite.
>> 
>> Great question about compacted topics! There are few usage of compacted
>> topic to consider:
>> 1. Typical Compacted Topic Usage: The compacted topics are commonly used
>> as cache/state layers where the latest value for each key represents the
>> current state. These use cases typically involve smaller records since they
>> need to be loaded efficiently for state reconstruction. Large message
>> serialization would be unusual for this pattern.
>> 
>> 2. User-Defined Compacted Topics with Large Payloads: for applications
>> that do use compacted topics with large payloads, the PayloadStore
>> implementation should handle this by:
>>        a. Consistent ID Generation: Use deterministic IDs based on the
>> Kafka key (rather than random UUIDs) so that when a payload id is updated,
>> it overwrites the same payload store object instead of creating new ones.
>> 
>>        b. TTL Strategy: Since compacted topics can retain data
>> indefinitely, users have two options:
>>                - Set a business-appropriate TTL (basically "we know our
>> cache data becomes stale after 30 days”)
>>                - Configure no TTL and accept indefinite storage costs as
>> a trade-off for the architectural benefits
>> 
>> 3. Compact + Delete Policy: Topics with `cleanup.policy=compact,delete`
>> will eventually remove old data, so standard TTL approaches work normally.
>> The key insight is that TTL configuration depends on the business
>> requirements and usage patterns rather than just the Kafka retention
>> policy. The PayloadStore implementation should provide flexibility for
>> users to make this trade-off consciously.
>> 
>> I have update the consideration section with this.
>> 
>> Thanks
>> Omnia
>> 
>>> On 11 Aug 2025, at 23:32, Jun Rao <ju...@confluent.io.INVALID> wrote:
>>> 
>>> Hi, Omnia,
>>> 
>>> Thanks for the reply. A few more comments.
>>> 
>>> JR20. Could you fix the inconsistency?
>>> org.apache.kafka.common.serialization.largemessage.PayloadStore vs
>>> org.apache.kafka.common.serialization.largemessage.store.PayloadStore
>>> 
>>> JR21. "Encapsulate that ID into a simple Kafka event using a structured
>>> format.". Could you define the structured format explicitly?
>>> 
>>> JR22. TTL for object stores: Could the new serializer be used on a
>>> compacted topic? If so, how should a user configure the TTL since the
>> Kafka
>>> retention is infinite.
>>> 
>>> Jun
>>> 
>>> On Mon, Aug 11, 2025 at 10:08 AM Omnia Ibrahim <o.g.h.ibrahim@gmail.com
>> <ma...@gmail.com>>
>>> wrote:
>>> 
>>>> Hi Jun
>>>>> JR11. Do you think that we need to add key.serializers and
>>>>> key.deserializers or do you think covering large messages in value is
>>>>> enough?
>>>> The large message issue usually is a value issue, I never saw a key that
>>>> is bigger than broker message size. If I would add key.serializers and
>>>> key.deserializers it would be for consistency and maybe there are
>> use-cases
>>>> where developers want to apply multiple serializations in order on the
>> key
>>>> as well outside the context of the large message support.
>>>> I updated the KIP to add these two configs now.
>>>> 
>>>>> JR12. We can load ComposableSerializer automatically if
>> value.serializers
>>>>> or value.deserializers are specified. But, it seems that we could keep
>>>>> ComposableSerializer as an internal implementation. For example,
>>>>> ProducerInterceptors is automatically loaded when multiple interceptors
>>>> are
>>>>> specified and is an internal class.
>>>> I updated the kip to highlight that the changes is updating
>> ProducerConfig
>>>> and ConsumerConfig to have new config for serializers/deserializers and
>> not
>>>> the actual class this is implementation details and not public
>> interfaces.
>>>> 
>>>>> JR17. We could estimate the size after compression, but the estimator
>> is
>>>>> not 100% accurate. It seems that it's simpler to just use the original
>>>>> message size.
>>>> We can keep it as original message size, I was thinking if it is good
>>>> enough for max.request.size it might be good enough for this. I updated
>> the
>>>> KIP anyway to simplify it and keep it to the original size.
>>>> 
>>>> Hope the final version addressed all feedbacks and we can resume with
>> the
>>>> voting
>>>> 
>>>> Thanks
>>>> Omnia
>>>> 
>>>>> On 8 Aug 2025, at 22:10, Jun Rao <ju...@confluent.io.INVALID> wrote:
>>>>> 
>>>>> Hi, Omnia,
>>>>> 
>>>>> Thanks for the reply.
>>>>> 
>>>>> JR11. Do you think that we need to add key.serializers and
>>>>> key.deserializers or do you think covering large messages in value is
>>>>> enough?
>>>>> 
>>>>> JR12. We can load ComposableSerializer automatically if
>> value.serializers
>>>>> or value.deserializers are specified. But, it seems that we could keep
>>>>> ComposableSerializer as an internal implementation. For example,
>>>>> ProducerInterceptors is automatically loaded when multiple interceptors
>>>> are
>>>>> specified and is an internal class.
>>>>> 
>>>>> JR17. We could estimate the size after compression, but the estimator
>> is
>>>>> not 100% accurate. It seems that it's simpler to just use the original
>>>>> message size.
>>>>> 
>>>>> Jun
>>>>> 
>>>>> On Mon, Aug 4, 2025 at 10:33 AM Omnia Ibrahim <o.g.h.ibrahim@gmail.com
>>>> <ma...@gmail.com>>
>>>>> wrote:
>>>>> 
>>>>>> 
>>>>>> Hi Jun
>>>>>>> JR11. value.serializers and value.deserializers: Should they be of
>> type
>>>>>>> List? Also, where are key.serializers and key.deserializers?
>>>>>>> 
>>>>>> Updated now
>>>>>>> JR12. Do we still need ComposableSerializer and
>> ComposableDeserializer?
>>>>>> The initial thinking here was if `value.serializers or
>>>> value.deserializers
>>>>>> ` exists the client will load  ComposableSerializer or
>>>>>> ComposableDeserializer  automatically and use them. Unfortunately this
>>>>>> would need us to define these serializers.
>>>>>> 
>>>>>> The other option is to update `Plugin<Serializer<V>>
>>>>>> valueSerializerPlugin` and KafkaProducer constructor to accept
>>>>>> List<Serializer<V>> and move the logic of the ComposableSerializer
>> into
>>>>>> KafkaProducer::doSend which when we do serialization (same with
>>>>>> KafkaConsumer). This option hide the logic and reduce exposure for
>>>> client
>>>>>> to these.
>>>>>> WDYT?
>>>>>> 
>>>>>>> 
>>>>>>> JR13. large.message.payload.store.class : should it be of type class?
>>>>>> Updated
>>>>>>> 
>>>>>>> JR14.
>>>>>>> 
>>>>>> 
>>>> 
>> org.apache.kafka.common.serialization.largemessage.LargeMessageSerializer :
>>>>>>> The name seems redundant since largemessage appears twice.
>>>>>> Updated
>>>>>>> 
>>>>>>> JR15. PayloadResponse: It still mentions response code. It mentions
>>>>>>> "isRetryable flag", which no longer exists in PayloadStoreException.
>>>>>> There
>>>>>>> are typos in "then it will serialiser will”.
>>>>>> The KIP is updated now
>>>>>> 
>>>>>>> JR16. Regarding returning new byte[0] if
>>>>>> large.message.skip.not.found.error
>>>>>>> is true, this will likely fail the next deserializer and the
>>>> application
>>>>>>> won't have the right context of the error. It's probably better to
>> just
>>>>>>> propagate the specific exception and let the caller handle it.
>>>>>> You right this will cause issue if there is another deserializer
>> waiting
>>>>>> for the data. I have updated the KIP with this.
>>>>>> 
>>>>>>> 
>>>>>>> JR17. LargeMessageSerializer:  "Check if the estimated size of the
>> data
>>>>>>> (bytes) after applying provided compression (if there is one)"
>>>>>> Compression
>>>>>>> actually happens after serialization and is done on a batch of
>> records.
>>>>>> Yes the compression itself happened after but we also have
>>>>>> `estimateSizeInBytesUpperBound` which my understanding is this take
>> the
>>>>>> compression type into the account as well
>>>>>>> 
>>>>>>> 
>>>>>>> JR18. Could you define the type T for LargeMessageSerializer?
>>>>>> Update the KIP this T would be byte[]
>>>>>> 
>>>>>> Thanks
>>>>>> Omnia
>>>>>>> 
>>>>>>> Jun
>>>>>>> 
>>>>>>> On Fri, Jul 25, 2025 at 6:28 AM Omnia Ibrahim <
>> o.g.h.ibrahim@gmail.com
>>>>>> <ma...@gmail.com>>
>>>>>>> wrote:
>>>>>>> 
>>>>>>>> Hi Jun, thanks for having the time to review this
>>>>>>>> 
>>>>>>>>> JR1. While the KIP is potentially useful, I am wondering who is
>>>>>>>> responsible
>>>>>>>>> for retention for the objects in the payload store. Once a message
>>>>>> with a
>>>>>>>>> reference is deleted, the key of the external object is lost and
>> the
>>>>>>>> object
>>>>>>>>> may never be deleted.
>>>>>>>> 
>>>>>>>> The `ttl` in the object store is the responsibility of the owner of
>>>> this
>>>>>>>> store it should be configured in away that is reasonable with the
>>>>>> retention
>>>>>>>> config in Kafka.
>>>>>>>> I have updated the KIP with `Consideration` section.
>>>>>>>>> 
>>>>>>>>> JR2. Configs: For all new configs, it would be useful to list their
>>>>>>>> types.
>>>>>>>> Updated the KIP now
>>>>>>>>> 
>>>>>>>>> JR3. value.serializers: Why is this required? If a user doesn't set
>>>> it,
>>>>>>>> we
>>>>>>>>> should just use value.serializer, right? Ditto for key.serializers.
>>>>>>>> No you right this was copy/past mistake
>>>>>>>> 
>>>>>>>>> 
>>>>>>>>> JR4. For all new public interfaces such as LargeMessageSerializer,
>>>>>>>>> PayloadStore and PayloadResponse, it would be useful to include the
>>>>>> full
>>>>>>>>> package name.
>>>>>>>> Updated the KIP now
>>>>>>>>> 
>>>>>>>>> JR5. large.message.payload.store.retry.max.backoff.ms and
>>>>>>>>> large.message.payload.store.retry.delay.backoff.ms: Is the
>> intention
>>>>>> to
>>>>>>>>> implement exponential backoff on retries? If so, it's more
>> consistent
>>>>>> if
>>>>>>>> we
>>>>>>>>> can follow the existing naming convention like
>> retry.backoff.max.ms
>>>> <
>>>>>>>> http://retry.backoff.max.ms/> and
>>>>>>>>> retry.backoff.ms <http://retry.backoff.ms/> <
>>>> http://retry.backoff.ms/> <http://retry.backoff.ms/
>>>>>>> .
>>>>>>>> I have removed these to simplify the config more (as Luke suggested
>>>>>>>> initially) and added these to the consideration section.
>>>>>>>> 
>>>>>>>>> 
>>>>>>>>> JR6. large.message.skip.not.found.error : If the reference can't be
>>>>>>>> found,
>>>>>>>>> what value does the deserializer return? Note that null has a
>> special
>>>>>>>>> meaning for tombstone in compacted topics.
>>>>>>>> The deserialiser will return `new byte[0]` not null.
>>>>>>>>> 
>>>>>>>>> JR7. PayloadResponse: Why do we have both responseCode and
>>>>>>>>> PayloadStoreException?
>>>>>>>> We can do without responseCode, the initial though was to report
>>>>>> response
>>>>>>>> code form payload store.
>>>>>>>> Update the KIP.
>>>>>>>>> JR8. Why do we need PayloadStore.metrics? Note that we could
>> monitor
>>>>>> the
>>>>>>>>> metrics in a plugin through the Monitorable interface.
>>>>>>>> Oh nice, I didn’t know about this interface before. Updated the KIP
>>>> with
>>>>>>>> this now.
>>>>>>>>> 
>>>>>>>>> JR9. Why do we need the protected field
>>>>>>>> PayloadStoreException.isRetryable?
>>>>>>>> Initial thought here was the serializer can retry the upload. But I
>>>> have
>>>>>>>> removed all the retry logic from serializer and it will be up to the
>>>>>>>> PayloadStore provider to implement this if they need it.
>>>>>>>>> 
>>>>>>>>> JR10. As Luke mentioned earlier, we could turn PayloadStore to an
>>>>>>>> interface.
>>>>>>>> It is updated now to interface.
>>>>>>>> 
>>>>>>>> Hope the last version of the KIP is more simpler now
>>>>>>>> 
>>>>>>>> Thanks
>>>>>>>> Omnia
>>>>>>>> 
>>>>>>>>> On 23 Jul 2025, at 00:43, Jun Rao <jun@confluent.io.INVALID
>> <ma...@confluent.io.INVALID> <mailto:
>>>> jun@confluent.io.INVALID <ma...@confluent.io.INVALID> <ma...@confluent.io.INVALID>> <mailto:
>>>>>> jun@confluent.io.INVALID <ma...@confluent.io.INVALID> <ma...@confluent.io.INVALID> <mailto:
>> jun@confluent.io.INVALID <ma...@confluent.io.INVALID>>>> wrote:
>>>>>>>>> 
>>>>>>>>> Thanks for the KIP. A few comments.
>>>>>>>>> 
>>>>>>>>> JR1. While the KIP is potentially useful, I am wondering who is
>>>>>>>> responsible
>>>>>>>>> for retention for the objects in the payload store. Once a message
>>>>>> with a
>>>>>>>>> reference is deleted, the key of the external object is lost and
>> the
>>>>>>>> object
>>>>>>>>> may never be deleted.
>>>>>>>>> 
>>>>>>>>> JR2. Configs: For all new configs, it would be useful to list their
>>>>>>>> types.
>>>>>>>>> 
>>>>>>>>> JR3. value.serializers: Why is this required? If a user doesn't set
>>>> it,
>>>>>>>> we
>>>>>>>>> should just use value.serializer, right? Ditto for key.serializers.
>>>>>>>>> 
>>>>>>>>> JR4. For all new public interfaces such as LargeMessageSerializer,
>>>>>>>>> PayloadStore and PayloadResponse, it would be useful to include the
>>>>>> full
>>>>>>>>> package name.
>>>>>>>>> 
>>>>>>>>> JR5. large.message.payload.store.retry.max.backoff.ms and
>>>>>>>>> large.message.payload.store.retry.delay.backoff.ms: Is the
>> intention
>>>>>> to
>>>>>>>>> implement exponential backoff on retries? If so, it's more
>> consistent
>>>>>> if
>>>>>>>> we
>>>>>>>>> can follow the existing naming convention like
>> retry.backoff.max.ms <http://retry.backoff.max.ms/> <http://retry.backoff.max.ms/>
>>>> <http://retry.backoff.max.ms/> <
>>>>>> http://retry.backoff.max.ms/> <
>>>>>>>> http://retry.backoff.max.ms/> and
>>>>>>>>> retry.backoff.ms <http://retry.backoff.ms/> <http://retry.backoff.ms/> <
>> http://retry.backoff.ms/> <
>>>> http://retry.backoff.ms/> <http://retry.backoff.ms/
>>>>>>> .
>>>>>>>>> 
>>>>>>>>> JR6. large.message.skip.not.found.error : If the reference can't be
>>>>>>>> found,
>>>>>>>>> what value does the deserializer return? Note that null has a
>> special
>>>>>>>>> meaning for tombstone in compacted topics.
>>>>>>>>> 
>>>>>>>>> JR7. PayloadResponse: Why do we have both responseCode and
>>>>>>>>> PayloadStoreException?
>>>>>>>>> 
>>>>>>>>> JR8. Why do we need PayloadStore.metrics? Note that we could
>> monitor
>>>>>> the
>>>>>>>>> metrics in a plugin through the Monitorable interface.
>>>>>>>>> 
>>>>>>>>> JR9. Why do we need the protected field
>>>>>>>> PayloadStoreException.isRetryable?
>>>>>>>>> 
>>>>>>>>> JR10. As Luke mentioned earlier, we could turn PayloadStore to an
>>>>>>>> interface.
>>>>>>>>> 
>>>>>>>>> Thanks,

Re: [DISCUSS] KIP-1159: Large message reference based Serializer

[Jun Rao <ju...@confluent.io.INVALID>]

Hi, Omnia,

Thanks for the reply. A few more comments.

JR20. The inconsistency is still there. It's probably simpler to
consolidate on
org.apache.kafka.common.serialization.largemessage.PayloadStore.

JR21. Could you document how the payload id string is encoded? For example,
is it always UTF8?

JR22. "Since compacted topics can retain data indefinitely, users must
choose between setting a business-appropriate TTL or accepting indefinite
storage costs." Setting a TTL is not ideal since it can break the consumer
application. So, we probably don't want to recommend it to the users.
Another option is to run a cleaning job by consuming the topic and if it
observes a null payload, delete the object based on the key in the record.

JR23. We probably don't need to force the PayloadStore implementer to
implement Monitorable. We could document that if it's desirable to monitor
a PayloadStore plugin, one can implement the Monitorable interface.

JR24. In the following, org.apache.kafka.common.serialization.Serializer
should include the largemessage namespace.
producerConfig.put("value.serializers",
"kafka.serializers.KafkaAvroDeserializer,
org.apache.kafka.common.serialization.Serializer");

Jun

On Tue, Aug 12, 2025 at 3:27 AM Omnia Ibrahim <o....@gmail.com>
wrote:

> Hi Jun,
>
> > JR20. Could you fix the inconsistency?
> > org.apache.kafka.common.serialization.largemessage.PayloadStore vs
> > org.apache.kafka.common.serialization.largemessage.store.PayloadStore
> Updated
> >
> > JR21. "Encapsulate that ID into a simple Kafka event using a structured
> > format.". Could you define the structured format explicitly?
>
> This was from the initial design we had Id format but after discussing
> this with Luke I ended up simplifying this by removing the format and I
> guess I forgot this line. I have updated the kip now.
>
> > JR22. TTL for object stores: Could the new serializer be used on a
> > compacted topic? If so, how should a user configure the TTL since the
> Kafka
> > retention is infinite.
>
> Great question about compacted topics! There are few usage of compacted
> topic to consider:
> 1. Typical Compacted Topic Usage: The compacted topics are commonly used
> as cache/state layers where the latest value for each key represents the
> current state. These use cases typically involve smaller records since they
> need to be loaded efficiently for state reconstruction. Large message
> serialization would be unusual for this pattern.
>
> 2. User-Defined Compacted Topics with Large Payloads: for applications
> that do use compacted topics with large payloads, the PayloadStore
> implementation should handle this by:
>         a. Consistent ID Generation: Use deterministic IDs based on the
> Kafka key (rather than random UUIDs) so that when a payload id is updated,
> it overwrites the same payload store object instead of creating new ones.
>
>         b. TTL Strategy: Since compacted topics can retain data
> indefinitely, users have two options:
>                 - Set a business-appropriate TTL (basically "we know our
> cache data becomes stale after 30 days”)
>                 - Configure no TTL and accept indefinite storage costs as
> a trade-off for the architectural benefits
>
> 3. Compact + Delete Policy: Topics with `cleanup.policy=compact,delete`
> will eventually remove old data, so standard TTL approaches work normally.
> The key insight is that TTL configuration depends on the business
> requirements and usage patterns rather than just the Kafka retention
> policy. The PayloadStore implementation should provide flexibility for
> users to make this trade-off consciously.
>
> I have update the consideration section with this.
>
> Thanks
> Omnia
>
> > On 11 Aug 2025, at 23:32, Jun Rao <ju...@confluent.io.INVALID> wrote:
> >
> > Hi, Omnia,
> >
> > Thanks for the reply. A few more comments.
> >
> > JR20. Could you fix the inconsistency?
> > org.apache.kafka.common.serialization.largemessage.PayloadStore vs
> > org.apache.kafka.common.serialization.largemessage.store.PayloadStore
> >
> > JR21. "Encapsulate that ID into a simple Kafka event using a structured
> > format.". Could you define the structured format explicitly?
> >
> > JR22. TTL for object stores: Could the new serializer be used on a
> > compacted topic? If so, how should a user configure the TTL since the
> Kafka
> > retention is infinite.
> >
> > Jun
> >
> > On Mon, Aug 11, 2025 at 10:08 AM Omnia Ibrahim <o.g.h.ibrahim@gmail.com
> <ma...@gmail.com>>
> > wrote:
> >
> >> Hi Jun
> >>> JR11. Do you think that we need to add key.serializers and
> >>> key.deserializers or do you think covering large messages in value is
> >>> enough?
> >> The large message issue usually is a value issue, I never saw a key that
> >> is bigger than broker message size. If I would add key.serializers and
> >> key.deserializers it would be for consistency and maybe there are
> use-cases
> >> where developers want to apply multiple serializations in order on the
> key
> >> as well outside the context of the large message support.
> >> I updated the KIP to add these two configs now.
> >>
> >>> JR12. We can load ComposableSerializer automatically if
> value.serializers
> >>> or value.deserializers are specified. But, it seems that we could keep
> >>> ComposableSerializer as an internal implementation. For example,
> >>> ProducerInterceptors is automatically loaded when multiple interceptors
> >> are
> >>> specified and is an internal class.
> >> I updated the kip to highlight that the changes is updating
> ProducerConfig
> >> and ConsumerConfig to have new config for serializers/deserializers and
> not
> >> the actual class this is implementation details and not public
> interfaces.
> >>
> >>> JR17. We could estimate the size after compression, but the estimator
> is
> >>> not 100% accurate. It seems that it's simpler to just use the original
> >>> message size.
> >> We can keep it as original message size, I was thinking if it is good
> >> enough for max.request.size it might be good enough for this. I updated
> the
> >> KIP anyway to simplify it and keep it to the original size.
> >>
> >> Hope the final version addressed all feedbacks and we can resume with
> the
> >> voting
> >>
> >> Thanks
> >> Omnia
> >>
> >>> On 8 Aug 2025, at 22:10, Jun Rao <ju...@confluent.io.INVALID> wrote:
> >>>
> >>> Hi, Omnia,
> >>>
> >>> Thanks for the reply.
> >>>
> >>> JR11. Do you think that we need to add key.serializers and
> >>> key.deserializers or do you think covering large messages in value is
> >>> enough?
> >>>
> >>> JR12. We can load ComposableSerializer automatically if
> value.serializers
> >>> or value.deserializers are specified. But, it seems that we could keep
> >>> ComposableSerializer as an internal implementation. For example,
> >>> ProducerInterceptors is automatically loaded when multiple interceptors
> >> are
> >>> specified and is an internal class.
> >>>
> >>> JR17. We could estimate the size after compression, but the estimator
> is
> >>> not 100% accurate. It seems that it's simpler to just use the original
> >>> message size.
> >>>
> >>> Jun
> >>>
> >>> On Mon, Aug 4, 2025 at 10:33 AM Omnia Ibrahim <o.g.h.ibrahim@gmail.com
> >> <ma...@gmail.com>>
> >>> wrote:
> >>>
> >>>>
> >>>> Hi Jun
> >>>>> JR11. value.serializers and value.deserializers: Should they be of
> type
> >>>>> List? Also, where are key.serializers and key.deserializers?
> >>>>>
> >>>> Updated now
> >>>>> JR12. Do we still need ComposableSerializer and
> ComposableDeserializer?
> >>>> The initial thinking here was if `value.serializers or
> >> value.deserializers
> >>>> ` exists the client will load  ComposableSerializer or
> >>>> ComposableDeserializer  automatically and use them. Unfortunately this
> >>>> would need us to define these serializers.
> >>>>
> >>>> The other option is to update `Plugin<Serializer<V>>
> >>>> valueSerializerPlugin` and KafkaProducer constructor to accept
> >>>> List<Serializer<V>> and move the logic of the ComposableSerializer
> into
> >>>> KafkaProducer::doSend which when we do serialization (same with
> >>>> KafkaConsumer). This option hide the logic and reduce exposure for
> >> client
> >>>> to these.
> >>>> WDYT?
> >>>>
> >>>>>
> >>>>> JR13. large.message.payload.store.class : should it be of type class?
> >>>> Updated
> >>>>>
> >>>>> JR14.
> >>>>>
> >>>>
> >>
> org.apache.kafka.common.serialization.largemessage.LargeMessageSerializer :
> >>>>> The name seems redundant since largemessage appears twice.
> >>>> Updated
> >>>>>
> >>>>> JR15. PayloadResponse: It still mentions response code. It mentions
> >>>>> "isRetryable flag", which no longer exists in PayloadStoreException.
> >>>> There
> >>>>> are typos in "then it will serialiser will”.
> >>>> The KIP is updated now
> >>>>
> >>>>> JR16. Regarding returning new byte[0] if
> >>>> large.message.skip.not.found.error
> >>>>> is true, this will likely fail the next deserializer and the
> >> application
> >>>>> won't have the right context of the error. It's probably better to
> just
> >>>>> propagate the specific exception and let the caller handle it.
> >>>> You right this will cause issue if there is another deserializer
> waiting
> >>>> for the data. I have updated the KIP with this.
> >>>>
> >>>>>
> >>>>> JR17. LargeMessageSerializer:  "Check if the estimated size of the
> data
> >>>>> (bytes) after applying provided compression (if there is one)"
> >>>> Compression
> >>>>> actually happens after serialization and is done on a batch of
> records.
> >>>> Yes the compression itself happened after but we also have
> >>>> `estimateSizeInBytesUpperBound` which my understanding is this take
> the
> >>>> compression type into the account as well
> >>>>>
> >>>>>
> >>>>> JR18. Could you define the type T for LargeMessageSerializer?
> >>>> Update the KIP this T would be byte[]
> >>>>
> >>>> Thanks
> >>>> Omnia
> >>>>>
> >>>>> Jun
> >>>>>
> >>>>> On Fri, Jul 25, 2025 at 6:28 AM Omnia Ibrahim <
> o.g.h.ibrahim@gmail.com
> >>>> <ma...@gmail.com>>
> >>>>> wrote:
> >>>>>
> >>>>>> Hi Jun, thanks for having the time to review this
> >>>>>>
> >>>>>>> JR1. While the KIP is potentially useful, I am wondering who is
> >>>>>> responsible
> >>>>>>> for retention for the objects in the payload store. Once a message
> >>>> with a
> >>>>>>> reference is deleted, the key of the external object is lost and
> the
> >>>>>> object
> >>>>>>> may never be deleted.
> >>>>>>
> >>>>>> The `ttl` in the object store is the responsibility of the owner of
> >> this
> >>>>>> store it should be configured in away that is reasonable with the
> >>>> retention
> >>>>>> config in Kafka.
> >>>>>> I have updated the KIP with `Consideration` section.
> >>>>>>>
> >>>>>>> JR2. Configs: For all new configs, it would be useful to list their
> >>>>>> types.
> >>>>>> Updated the KIP now
> >>>>>>>
> >>>>>>> JR3. value.serializers: Why is this required? If a user doesn't set
> >> it,
> >>>>>> we
> >>>>>>> should just use value.serializer, right? Ditto for key.serializers.
> >>>>>> No you right this was copy/past mistake
> >>>>>>
> >>>>>>>
> >>>>>>> JR4. For all new public interfaces such as LargeMessageSerializer,
> >>>>>>> PayloadStore and PayloadResponse, it would be useful to include the
> >>>> full
> >>>>>>> package name.
> >>>>>> Updated the KIP now
> >>>>>>>
> >>>>>>> JR5. large.message.payload.store.retry.max.backoff.ms and
> >>>>>>> large.message.payload.store.retry.delay.backoff.ms: Is the
> intention
> >>>> to
> >>>>>>> implement exponential backoff on retries? If so, it's more
> consistent
> >>>> if
> >>>>>> we
> >>>>>>> can follow the existing naming convention like
> retry.backoff.max.ms
> >> <
> >>>>>> http://retry.backoff.max.ms/> and
> >>>>>>> retry.backoff.ms <http://retry.backoff.ms/> <
> >> http://retry.backoff.ms/> <http://retry.backoff.ms/
> >>>>> .
> >>>>>> I have removed these to simplify the config more (as Luke suggested
> >>>>>> initially) and added these to the consideration section.
> >>>>>>
> >>>>>>>
> >>>>>>> JR6. large.message.skip.not.found.error : If the reference can't be
> >>>>>> found,
> >>>>>>> what value does the deserializer return? Note that null has a
> special
> >>>>>>> meaning for tombstone in compacted topics.
> >>>>>> The deserialiser will return `new byte[0]` not null.
> >>>>>>>
> >>>>>>> JR7. PayloadResponse: Why do we have both responseCode and
> >>>>>>> PayloadStoreException?
> >>>>>> We can do without responseCode, the initial though was to report
> >>>> response
> >>>>>> code form payload store.
> >>>>>> Update the KIP.
> >>>>>>> JR8. Why do we need PayloadStore.metrics? Note that we could
> monitor
> >>>> the
> >>>>>>> metrics in a plugin through the Monitorable interface.
> >>>>>> Oh nice, I didn’t know about this interface before. Updated the KIP
> >> with
> >>>>>> this now.
> >>>>>>>
> >>>>>>> JR9. Why do we need the protected field
> >>>>>> PayloadStoreException.isRetryable?
> >>>>>> Initial thought here was the serializer can retry the upload. But I
> >> have
> >>>>>> removed all the retry logic from serializer and it will be up to the
> >>>>>> PayloadStore provider to implement this if they need it.
> >>>>>>>
> >>>>>>> JR10. As Luke mentioned earlier, we could turn PayloadStore to an
> >>>>>> interface.
> >>>>>> It is updated now to interface.
> >>>>>>
> >>>>>> Hope the last version of the KIP is more simpler now
> >>>>>>
> >>>>>> Thanks
> >>>>>> Omnia
> >>>>>>
> >>>>>>> On 23 Jul 2025, at 00:43, Jun Rao <jun@confluent.io.INVALID
> <ma...@confluent.io.INVALID> <mailto:
> >> jun@confluent.io.INVALID <ma...@confluent.io.INVALID>> <mailto:
> >>>> jun@confluent.io.INVALID <ma...@confluent.io.INVALID> <mailto:
> jun@confluent.io.INVALID>>> wrote:
> >>>>>>>
> >>>>>>> Thanks for the KIP. A few comments.
> >>>>>>>
> >>>>>>> JR1. While the KIP is potentially useful, I am wondering who is
> >>>>>> responsible
> >>>>>>> for retention for the objects in the payload store. Once a message
> >>>> with a
> >>>>>>> reference is deleted, the key of the external object is lost and
> the
> >>>>>> object
> >>>>>>> may never be deleted.
> >>>>>>>
> >>>>>>> JR2. Configs: For all new configs, it would be useful to list their
> >>>>>> types.
> >>>>>>>
> >>>>>>> JR3. value.serializers: Why is this required? If a user doesn't set
> >> it,
> >>>>>> we
> >>>>>>> should just use value.serializer, right? Ditto for key.serializers.
> >>>>>>>
> >>>>>>> JR4. For all new public interfaces such as LargeMessageSerializer,
> >>>>>>> PayloadStore and PayloadResponse, it would be useful to include the
> >>>> full
> >>>>>>> package name.
> >>>>>>>
> >>>>>>> JR5. large.message.payload.store.retry.max.backoff.ms and
> >>>>>>> large.message.payload.store.retry.delay.backoff.ms: Is the
> intention
> >>>> to
> >>>>>>> implement exponential backoff on retries? If so, it's more
> consistent
> >>>> if
> >>>>>> we
> >>>>>>> can follow the existing naming convention like
> retry.backoff.max.ms <http://retry.backoff.max.ms/>
> >> <http://retry.backoff.max.ms/> <
> >>>> http://retry.backoff.max.ms/> <
> >>>>>> http://retry.backoff.max.ms/> and
> >>>>>>> retry.backoff.ms <http://retry.backoff.ms/> <
> http://retry.backoff.ms/> <
> >> http://retry.backoff.ms/> <http://retry.backoff.ms/
> >>>>> .
> >>>>>>>
> >>>>>>> JR6. large.message.skip.not.found.error : If the reference can't be
> >>>>>> found,
> >>>>>>> what value does the deserializer return? Note that null has a
> special
> >>>>>>> meaning for tombstone in compacted topics.
> >>>>>>>
> >>>>>>> JR7. PayloadResponse: Why do we have both responseCode and
> >>>>>>> PayloadStoreException?
> >>>>>>>
> >>>>>>> JR8. Why do we need PayloadStore.metrics? Note that we could
> monitor
> >>>> the
> >>>>>>> metrics in a plugin through the Monitorable interface.
> >>>>>>>
> >>>>>>> JR9. Why do we need the protected field
> >>>>>> PayloadStoreException.isRetryable?
> >>>>>>>
> >>>>>>> JR10. As Luke mentioned earlier, we could turn PayloadStore to an
> >>>>>> interface.
> >>>>>>>
> >>>>>>> Thanks,
>
>

Re: [DISCUSS] KIP-1159: Large message reference based Serializer

[Omnia Ibrahim <o....@gmail.com>]

Hi Jun, 

> JR20. Could you fix the inconsistency?
> org.apache.kafka.common.serialization.largemessage.PayloadStore vs
> org.apache.kafka.common.serialization.largemessage.store.PayloadStore
Updated
> 
> JR21. "Encapsulate that ID into a simple Kafka event using a structured
> format.". Could you define the structured format explicitly?

This was from the initial design we had Id format but after discussing this with Luke I ended up simplifying this by removing the format and I guess I forgot this line. I have updated the kip now. 

> JR22. TTL for object stores: Could the new serializer be used on a
> compacted topic? If so, how should a user configure the TTL since the Kafka
> retention is infinite.

Great question about compacted topics! There are few usage of compacted topic to consider: 
1. Typical Compacted Topic Usage: The compacted topics are commonly used as cache/state layers where the latest value for each key represents the current state. These use cases typically involve smaller records since they need to be loaded efficiently for state reconstruction. Large message serialization would be unusual for this pattern.

2. User-Defined Compacted Topics with Large Payloads: for applications that do use compacted topics with large payloads, the PayloadStore implementation should handle this by:
	a. Consistent ID Generation: Use deterministic IDs based on the Kafka key (rather than random UUIDs) so that when a payload id is updated, it overwrites the same payload store object instead of creating new ones.  

	b. TTL Strategy: Since compacted topics can retain data indefinitely, users have two options:    
		- Set a business-appropriate TTL (basically "we know our cache data becomes stale after 30 days”)   
		- Configure no TTL and accept indefinite storage costs as a trade-off for the architectural benefits  

3. Compact + Delete Policy: Topics with `cleanup.policy=compact,delete` will eventually remove old data, so standard TTL approaches work normally.  The key insight is that TTL configuration depends on the business requirements and usage patterns rather than just the Kafka retention policy. The PayloadStore implementation should provide flexibility for users to make this trade-off consciously.

I have update the consideration section with this. 

Thanks 
Omnia

> On 11 Aug 2025, at 23:32, Jun Rao <ju...@confluent.io.INVALID> wrote:
> 
> Hi, Omnia,
> 
> Thanks for the reply. A few more comments.
> 
> JR20. Could you fix the inconsistency?
> org.apache.kafka.common.serialization.largemessage.PayloadStore vs
> org.apache.kafka.common.serialization.largemessage.store.PayloadStore
> 
> JR21. "Encapsulate that ID into a simple Kafka event using a structured
> format.". Could you define the structured format explicitly?
> 
> JR22. TTL for object stores: Could the new serializer be used on a
> compacted topic? If so, how should a user configure the TTL since the Kafka
> retention is infinite.
> 
> Jun
> 
> On Mon, Aug 11, 2025 at 10:08 AM Omnia Ibrahim <o.g.h.ibrahim@gmail.com <ma...@gmail.com>>
> wrote:
> 
>> Hi Jun
>>> JR11. Do you think that we need to add key.serializers and
>>> key.deserializers or do you think covering large messages in value is
>>> enough?
>> The large message issue usually is a value issue, I never saw a key that
>> is bigger than broker message size. If I would add key.serializers and
>> key.deserializers it would be for consistency and maybe there are use-cases
>> where developers want to apply multiple serializations in order on the key
>> as well outside the context of the large message support.
>> I updated the KIP to add these two configs now.
>> 
>>> JR12. We can load ComposableSerializer automatically if value.serializers
>>> or value.deserializers are specified. But, it seems that we could keep
>>> ComposableSerializer as an internal implementation. For example,
>>> ProducerInterceptors is automatically loaded when multiple interceptors
>> are
>>> specified and is an internal class.
>> I updated the kip to highlight that the changes is updating ProducerConfig
>> and ConsumerConfig to have new config for serializers/deserializers and not
>> the actual class this is implementation details and not public interfaces.
>> 
>>> JR17. We could estimate the size after compression, but the estimator is
>>> not 100% accurate. It seems that it's simpler to just use the original
>>> message size.
>> We can keep it as original message size, I was thinking if it is good
>> enough for max.request.size it might be good enough for this. I updated the
>> KIP anyway to simplify it and keep it to the original size.
>> 
>> Hope the final version addressed all feedbacks and we can resume with the
>> voting
>> 
>> Thanks
>> Omnia
>> 
>>> On 8 Aug 2025, at 22:10, Jun Rao <ju...@confluent.io.INVALID> wrote:
>>> 
>>> Hi, Omnia,
>>> 
>>> Thanks for the reply.
>>> 
>>> JR11. Do you think that we need to add key.serializers and
>>> key.deserializers or do you think covering large messages in value is
>>> enough?
>>> 
>>> JR12. We can load ComposableSerializer automatically if value.serializers
>>> or value.deserializers are specified. But, it seems that we could keep
>>> ComposableSerializer as an internal implementation. For example,
>>> ProducerInterceptors is automatically loaded when multiple interceptors
>> are
>>> specified and is an internal class.
>>> 
>>> JR17. We could estimate the size after compression, but the estimator is
>>> not 100% accurate. It seems that it's simpler to just use the original
>>> message size.
>>> 
>>> Jun
>>> 
>>> On Mon, Aug 4, 2025 at 10:33 AM Omnia Ibrahim <o.g.h.ibrahim@gmail.com
>> <ma...@gmail.com>>
>>> wrote:
>>> 
>>>> 
>>>> Hi Jun
>>>>> JR11. value.serializers and value.deserializers: Should they be of type
>>>>> List? Also, where are key.serializers and key.deserializers?
>>>>> 
>>>> Updated now
>>>>> JR12. Do we still need ComposableSerializer and ComposableDeserializer?
>>>> The initial thinking here was if `value.serializers or
>> value.deserializers
>>>> ` exists the client will load  ComposableSerializer or
>>>> ComposableDeserializer  automatically and use them. Unfortunately this
>>>> would need us to define these serializers.
>>>> 
>>>> The other option is to update `Plugin<Serializer<V>>
>>>> valueSerializerPlugin` and KafkaProducer constructor to accept
>>>> List<Serializer<V>> and move the logic of the ComposableSerializer into
>>>> KafkaProducer::doSend which when we do serialization (same with
>>>> KafkaConsumer). This option hide the logic and reduce exposure for
>> client
>>>> to these.
>>>> WDYT?
>>>> 
>>>>> 
>>>>> JR13. large.message.payload.store.class : should it be of type class?
>>>> Updated
>>>>> 
>>>>> JR14.
>>>>> 
>>>> 
>> org.apache.kafka.common.serialization.largemessage.LargeMessageSerializer :
>>>>> The name seems redundant since largemessage appears twice.
>>>> Updated
>>>>> 
>>>>> JR15. PayloadResponse: It still mentions response code. It mentions
>>>>> "isRetryable flag", which no longer exists in PayloadStoreException.
>>>> There
>>>>> are typos in "then it will serialiser will”.
>>>> The KIP is updated now
>>>> 
>>>>> JR16. Regarding returning new byte[0] if
>>>> large.message.skip.not.found.error
>>>>> is true, this will likely fail the next deserializer and the
>> application
>>>>> won't have the right context of the error. It's probably better to just
>>>>> propagate the specific exception and let the caller handle it.
>>>> You right this will cause issue if there is another deserializer waiting
>>>> for the data. I have updated the KIP with this.
>>>> 
>>>>> 
>>>>> JR17. LargeMessageSerializer:  "Check if the estimated size of the data
>>>>> (bytes) after applying provided compression (if there is one)"
>>>> Compression
>>>>> actually happens after serialization and is done on a batch of records.
>>>> Yes the compression itself happened after but we also have
>>>> `estimateSizeInBytesUpperBound` which my understanding is this take the
>>>> compression type into the account as well
>>>>> 
>>>>> 
>>>>> JR18. Could you define the type T for LargeMessageSerializer?
>>>> Update the KIP this T would be byte[]
>>>> 
>>>> Thanks
>>>> Omnia
>>>>> 
>>>>> Jun
>>>>> 
>>>>> On Fri, Jul 25, 2025 at 6:28 AM Omnia Ibrahim <o.g.h.ibrahim@gmail.com
>>>> <ma...@gmail.com>>
>>>>> wrote:
>>>>> 
>>>>>> Hi Jun, thanks for having the time to review this
>>>>>> 
>>>>>>> JR1. While the KIP is potentially useful, I am wondering who is
>>>>>> responsible
>>>>>>> for retention for the objects in the payload store. Once a message
>>>> with a
>>>>>>> reference is deleted, the key of the external object is lost and the
>>>>>> object
>>>>>>> may never be deleted.
>>>>>> 
>>>>>> The `ttl` in the object store is the responsibility of the owner of
>> this
>>>>>> store it should be configured in away that is reasonable with the
>>>> retention
>>>>>> config in Kafka.
>>>>>> I have updated the KIP with `Consideration` section.
>>>>>>> 
>>>>>>> JR2. Configs: For all new configs, it would be useful to list their
>>>>>> types.
>>>>>> Updated the KIP now
>>>>>>> 
>>>>>>> JR3. value.serializers: Why is this required? If a user doesn't set
>> it,
>>>>>> we
>>>>>>> should just use value.serializer, right? Ditto for key.serializers.
>>>>>> No you right this was copy/past mistake
>>>>>> 
>>>>>>> 
>>>>>>> JR4. For all new public interfaces such as LargeMessageSerializer,
>>>>>>> PayloadStore and PayloadResponse, it would be useful to include the
>>>> full
>>>>>>> package name.
>>>>>> Updated the KIP now
>>>>>>> 
>>>>>>> JR5. large.message.payload.store.retry.max.backoff.ms and
>>>>>>> large.message.payload.store.retry.delay.backoff.ms: Is the intention
>>>> to
>>>>>>> implement exponential backoff on retries? If so, it's more consistent
>>>> if
>>>>>> we
>>>>>>> can follow the existing naming convention like retry.backoff.max.ms
>> <
>>>>>> http://retry.backoff.max.ms/> and
>>>>>>> retry.backoff.ms <http://retry.backoff.ms/> <
>> http://retry.backoff.ms/> <http://retry.backoff.ms/
>>>>> .
>>>>>> I have removed these to simplify the config more (as Luke suggested
>>>>>> initially) and added these to the consideration section.
>>>>>> 
>>>>>>> 
>>>>>>> JR6. large.message.skip.not.found.error : If the reference can't be
>>>>>> found,
>>>>>>> what value does the deserializer return? Note that null has a special
>>>>>>> meaning for tombstone in compacted topics.
>>>>>> The deserialiser will return `new byte[0]` not null.
>>>>>>> 
>>>>>>> JR7. PayloadResponse: Why do we have both responseCode and
>>>>>>> PayloadStoreException?
>>>>>> We can do without responseCode, the initial though was to report
>>>> response
>>>>>> code form payload store.
>>>>>> Update the KIP.
>>>>>>> JR8. Why do we need PayloadStore.metrics? Note that we could monitor
>>>> the
>>>>>>> metrics in a plugin through the Monitorable interface.
>>>>>> Oh nice, I didn’t know about this interface before. Updated the KIP
>> with
>>>>>> this now.
>>>>>>> 
>>>>>>> JR9. Why do we need the protected field
>>>>>> PayloadStoreException.isRetryable?
>>>>>> Initial thought here was the serializer can retry the upload. But I
>> have
>>>>>> removed all the retry logic from serializer and it will be up to the
>>>>>> PayloadStore provider to implement this if they need it.
>>>>>>> 
>>>>>>> JR10. As Luke mentioned earlier, we could turn PayloadStore to an
>>>>>> interface.
>>>>>> It is updated now to interface.
>>>>>> 
>>>>>> Hope the last version of the KIP is more simpler now
>>>>>> 
>>>>>> Thanks
>>>>>> Omnia
>>>>>> 
>>>>>>> On 23 Jul 2025, at 00:43, Jun Rao <jun@confluent.io.INVALID <ma...@confluent.io.INVALID> <mailto:
>> jun@confluent.io.INVALID <ma...@confluent.io.INVALID>> <mailto:
>>>> jun@confluent.io.INVALID <ma...@confluent.io.INVALID> <ma...@confluent.io.INVALID>>> wrote:
>>>>>>> 
>>>>>>> Thanks for the KIP. A few comments.
>>>>>>> 
>>>>>>> JR1. While the KIP is potentially useful, I am wondering who is
>>>>>> responsible
>>>>>>> for retention for the objects in the payload store. Once a message
>>>> with a
>>>>>>> reference is deleted, the key of the external object is lost and the
>>>>>> object
>>>>>>> may never be deleted.
>>>>>>> 
>>>>>>> JR2. Configs: For all new configs, it would be useful to list their
>>>>>> types.
>>>>>>> 
>>>>>>> JR3. value.serializers: Why is this required? If a user doesn't set
>> it,
>>>>>> we
>>>>>>> should just use value.serializer, right? Ditto for key.serializers.
>>>>>>> 
>>>>>>> JR4. For all new public interfaces such as LargeMessageSerializer,
>>>>>>> PayloadStore and PayloadResponse, it would be useful to include the
>>>> full
>>>>>>> package name.
>>>>>>> 
>>>>>>> JR5. large.message.payload.store.retry.max.backoff.ms and
>>>>>>> large.message.payload.store.retry.delay.backoff.ms: Is the intention
>>>> to
>>>>>>> implement exponential backoff on retries? If so, it's more consistent
>>>> if
>>>>>> we
>>>>>>> can follow the existing naming convention like retry.backoff.max.ms <http://retry.backoff.max.ms/>
>> <http://retry.backoff.max.ms/> <
>>>> http://retry.backoff.max.ms/> <
>>>>>> http://retry.backoff.max.ms/> and
>>>>>>> retry.backoff.ms <http://retry.backoff.ms/> <http://retry.backoff.ms/> <
>> http://retry.backoff.ms/> <http://retry.backoff.ms/
>>>>> .
>>>>>>> 
>>>>>>> JR6. large.message.skip.not.found.error : If the reference can't be
>>>>>> found,
>>>>>>> what value does the deserializer return? Note that null has a special
>>>>>>> meaning for tombstone in compacted topics.
>>>>>>> 
>>>>>>> JR7. PayloadResponse: Why do we have both responseCode and
>>>>>>> PayloadStoreException?
>>>>>>> 
>>>>>>> JR8. Why do we need PayloadStore.metrics? Note that we could monitor
>>>> the
>>>>>>> metrics in a plugin through the Monitorable interface.
>>>>>>> 
>>>>>>> JR9. Why do we need the protected field
>>>>>> PayloadStoreException.isRetryable?
>>>>>>> 
>>>>>>> JR10. As Luke mentioned earlier, we could turn PayloadStore to an
>>>>>> interface.
>>>>>>> 
>>>>>>> Thanks,

Re: [DISCUSS] KIP-1159: Large message reference based Serializer

[Jun Rao <ju...@confluent.io.INVALID>]

Hi, Omnia,

Thanks for the reply. A few more comments.

JR20. Could you fix the inconsistency?
org.apache.kafka.common.serialization.largemessage.PayloadStore vs
org.apache.kafka.common.serialization.largemessage.store.PayloadStore

JR21. "Encapsulate that ID into a simple Kafka event using a structured
format.". Could you define the structured format explicitly?

JR22. TTL for object stores: Could the new serializer be used on a
compacted topic? If so, how should a user configure the TTL since the Kafka
retention is infinite.

Jun

On Mon, Aug 11, 2025 at 10:08 AM Omnia Ibrahim <o....@gmail.com>
wrote:

> Hi Jun
> > JR11. Do you think that we need to add key.serializers and
> > key.deserializers or do you think covering large messages in value is
> > enough?
> The large message issue usually is a value issue, I never saw a key that
> is bigger than broker message size. If I would add key.serializers and
> key.deserializers it would be for consistency and maybe there are use-cases
> where developers want to apply multiple serializations in order on the key
> as well outside the context of the large message support.
> I updated the KIP to add these two configs now.
>
> > JR12. We can load ComposableSerializer automatically if value.serializers
> > or value.deserializers are specified. But, it seems that we could keep
> > ComposableSerializer as an internal implementation. For example,
> > ProducerInterceptors is automatically loaded when multiple interceptors
> are
> > specified and is an internal class.
> I updated the kip to highlight that the changes is updating ProducerConfig
> and ConsumerConfig to have new config for serializers/deserializers and not
> the actual class this is implementation details and not public interfaces.
>
> > JR17. We could estimate the size after compression, but the estimator is
> > not 100% accurate. It seems that it's simpler to just use the original
> > message size.
> We can keep it as original message size, I was thinking if it is good
> enough for max.request.size it might be good enough for this. I updated the
> KIP anyway to simplify it and keep it to the original size.
>
> Hope the final version addressed all feedbacks and we can resume with the
> voting
>
> Thanks
> Omnia
>
> > On 8 Aug 2025, at 22:10, Jun Rao <ju...@confluent.io.INVALID> wrote:
> >
> > Hi, Omnia,
> >
> > Thanks for the reply.
> >
> > JR11. Do you think that we need to add key.serializers and
> > key.deserializers or do you think covering large messages in value is
> > enough?
> >
> > JR12. We can load ComposableSerializer automatically if value.serializers
> > or value.deserializers are specified. But, it seems that we could keep
> > ComposableSerializer as an internal implementation. For example,
> > ProducerInterceptors is automatically loaded when multiple interceptors
> are
> > specified and is an internal class.
> >
> > JR17. We could estimate the size after compression, but the estimator is
> > not 100% accurate. It seems that it's simpler to just use the original
> > message size.
> >
> > Jun
> >
> > On Mon, Aug 4, 2025 at 10:33 AM Omnia Ibrahim <o.g.h.ibrahim@gmail.com
> <ma...@gmail.com>>
> > wrote:
> >
> >>
> >> Hi Jun
> >>> JR11. value.serializers and value.deserializers: Should they be of type
> >>> List? Also, where are key.serializers and key.deserializers?
> >>>
> >> Updated now
> >>> JR12. Do we still need ComposableSerializer and ComposableDeserializer?
> >> The initial thinking here was if `value.serializers or
> value.deserializers
> >> ` exists the client will load  ComposableSerializer or
> >> ComposableDeserializer  automatically and use them. Unfortunately this
> >> would need us to define these serializers.
> >>
> >> The other option is to update `Plugin<Serializer<V>>
> >> valueSerializerPlugin` and KafkaProducer constructor to accept
> >> List<Serializer<V>> and move the logic of the ComposableSerializer into
> >> KafkaProducer::doSend which when we do serialization (same with
> >> KafkaConsumer). This option hide the logic and reduce exposure for
> client
> >> to these.
> >> WDYT?
> >>
> >>>
> >>> JR13. large.message.payload.store.class : should it be of type class?
> >> Updated
> >>>
> >>> JR14.
> >>>
> >>
> org.apache.kafka.common.serialization.largemessage.LargeMessageSerializer :
> >>> The name seems redundant since largemessage appears twice.
> >> Updated
> >>>
> >>> JR15. PayloadResponse: It still mentions response code. It mentions
> >>> "isRetryable flag", which no longer exists in PayloadStoreException.
> >> There
> >>> are typos in "then it will serialiser will”.
> >> The KIP is updated now
> >>
> >>> JR16. Regarding returning new byte[0] if
> >> large.message.skip.not.found.error
> >>> is true, this will likely fail the next deserializer and the
> application
> >>> won't have the right context of the error. It's probably better to just
> >>> propagate the specific exception and let the caller handle it.
> >> You right this will cause issue if there is another deserializer waiting
> >> for the data. I have updated the KIP with this.
> >>
> >>>
> >>> JR17. LargeMessageSerializer:  "Check if the estimated size of the data
> >>> (bytes) after applying provided compression (if there is one)"
> >> Compression
> >>> actually happens after serialization and is done on a batch of records.
> >> Yes the compression itself happened after but we also have
> >> `estimateSizeInBytesUpperBound` which my understanding is this take the
> >> compression type into the account as well
> >>>
> >>>
> >>> JR18. Could you define the type T for LargeMessageSerializer?
> >> Update the KIP this T would be byte[]
> >>
> >> Thanks
> >> Omnia
> >>>
> >>> Jun
> >>>
> >>> On Fri, Jul 25, 2025 at 6:28 AM Omnia Ibrahim <o.g.h.ibrahim@gmail.com
> >> <ma...@gmail.com>>
> >>> wrote:
> >>>
> >>>> Hi Jun, thanks for having the time to review this
> >>>>
> >>>>> JR1. While the KIP is potentially useful, I am wondering who is
> >>>> responsible
> >>>>> for retention for the objects in the payload store. Once a message
> >> with a
> >>>>> reference is deleted, the key of the external object is lost and the
> >>>> object
> >>>>> may never be deleted.
> >>>>
> >>>> The `ttl` in the object store is the responsibility of the owner of
> this
> >>>> store it should be configured in away that is reasonable with the
> >> retention
> >>>> config in Kafka.
> >>>> I have updated the KIP with `Consideration` section.
> >>>>>
> >>>>> JR2. Configs: For all new configs, it would be useful to list their
> >>>> types.
> >>>> Updated the KIP now
> >>>>>
> >>>>> JR3. value.serializers: Why is this required? If a user doesn't set
> it,
> >>>> we
> >>>>> should just use value.serializer, right? Ditto for key.serializers.
> >>>> No you right this was copy/past mistake
> >>>>
> >>>>>
> >>>>> JR4. For all new public interfaces such as LargeMessageSerializer,
> >>>>> PayloadStore and PayloadResponse, it would be useful to include the
> >> full
> >>>>> package name.
> >>>> Updated the KIP now
> >>>>>
> >>>>> JR5. large.message.payload.store.retry.max.backoff.ms and
> >>>>> large.message.payload.store.retry.delay.backoff.ms: Is the intention
> >> to
> >>>>> implement exponential backoff on retries? If so, it's more consistent
> >> if
> >>>> we
> >>>>> can follow the existing naming convention like retry.backoff.max.ms
> <
> >>>> http://retry.backoff.max.ms/> and
> >>>>> retry.backoff.ms <http://retry.backoff.ms/> <
> http://retry.backoff.ms/> <http://retry.backoff.ms/
> >>> .
> >>>> I have removed these to simplify the config more (as Luke suggested
> >>>> initially) and added these to the consideration section.
> >>>>
> >>>>>
> >>>>> JR6. large.message.skip.not.found.error : If the reference can't be
> >>>> found,
> >>>>> what value does the deserializer return? Note that null has a special
> >>>>> meaning for tombstone in compacted topics.
> >>>> The deserialiser will return `new byte[0]` not null.
> >>>>>
> >>>>> JR7. PayloadResponse: Why do we have both responseCode and
> >>>>> PayloadStoreException?
> >>>> We can do without responseCode, the initial though was to report
> >> response
> >>>> code form payload store.
> >>>> Update the KIP.
> >>>>> JR8. Why do we need PayloadStore.metrics? Note that we could monitor
> >> the
> >>>>> metrics in a plugin through the Monitorable interface.
> >>>> Oh nice, I didn’t know about this interface before. Updated the KIP
> with
> >>>> this now.
> >>>>>
> >>>>> JR9. Why do we need the protected field
> >>>> PayloadStoreException.isRetryable?
> >>>> Initial thought here was the serializer can retry the upload. But I
> have
> >>>> removed all the retry logic from serializer and it will be up to the
> >>>> PayloadStore provider to implement this if they need it.
> >>>>>
> >>>>> JR10. As Luke mentioned earlier, we could turn PayloadStore to an
> >>>> interface.
> >>>> It is updated now to interface.
> >>>>
> >>>> Hope the last version of the KIP is more simpler now
> >>>>
> >>>> Thanks
> >>>> Omnia
> >>>>
> >>>>> On 23 Jul 2025, at 00:43, Jun Rao <jun@confluent.io.INVALID <mailto:
> jun@confluent.io.INVALID> <mailto:
> >> jun@confluent.io.INVALID <ma...@confluent.io.INVALID>>> wrote:
> >>>>>
> >>>>> Thanks for the KIP. A few comments.
> >>>>>
> >>>>> JR1. While the KIP is potentially useful, I am wondering who is
> >>>> responsible
> >>>>> for retention for the objects in the payload store. Once a message
> >> with a
> >>>>> reference is deleted, the key of the external object is lost and the
> >>>> object
> >>>>> may never be deleted.
> >>>>>
> >>>>> JR2. Configs: For all new configs, it would be useful to list their
> >>>> types.
> >>>>>
> >>>>> JR3. value.serializers: Why is this required? If a user doesn't set
> it,
> >>>> we
> >>>>> should just use value.serializer, right? Ditto for key.serializers.
> >>>>>
> >>>>> JR4. For all new public interfaces such as LargeMessageSerializer,
> >>>>> PayloadStore and PayloadResponse, it would be useful to include the
> >> full
> >>>>> package name.
> >>>>>
> >>>>> JR5. large.message.payload.store.retry.max.backoff.ms and
> >>>>> large.message.payload.store.retry.delay.backoff.ms: Is the intention
> >> to
> >>>>> implement exponential backoff on retries? If so, it's more consistent
> >> if
> >>>> we
> >>>>> can follow the existing naming convention like retry.backoff.max.ms
> <http://retry.backoff.max.ms/> <
> >> http://retry.backoff.max.ms/> <
> >>>> http://retry.backoff.max.ms/> and
> >>>>> retry.backoff.ms <http://retry.backoff.ms/> <
> http://retry.backoff.ms/> <http://retry.backoff.ms/
> >>> .
> >>>>>
> >>>>> JR6. large.message.skip.not.found.error : If the reference can't be
> >>>> found,
> >>>>> what value does the deserializer return? Note that null has a special
> >>>>> meaning for tombstone in compacted topics.
> >>>>>
> >>>>> JR7. PayloadResponse: Why do we have both responseCode and
> >>>>> PayloadStoreException?
> >>>>>
> >>>>> JR8. Why do we need PayloadStore.metrics? Note that we could monitor
> >> the
> >>>>> metrics in a plugin through the Monitorable interface.
> >>>>>
> >>>>> JR9. Why do we need the protected field
> >>>> PayloadStoreException.isRetryable?
> >>>>>
> >>>>> JR10. As Luke mentioned earlier, we could turn PayloadStore to an
> >>>> interface.
> >>>>>
> >>>>> Thanks,
>
>

Re: [DISCUSS] KIP-1159: Large message reference based Serializer

[Omnia Ibrahim <o....@gmail.com>]

Hi Jun
> JR11. Do you think that we need to add key.serializers and
> key.deserializers or do you think covering large messages in value is
> enough?
The large message issue usually is a value issue, I never saw a key that is bigger than broker message size. If I would add key.serializers and key.deserializers it would be for consistency and maybe there are use-cases where developers want to apply multiple serializations in order on the key as well outside the context of the large message support. 
I updated the KIP to add these two configs now.

> JR12. We can load ComposableSerializer automatically if value.serializers
> or value.deserializers are specified. But, it seems that we could keep
> ComposableSerializer as an internal implementation. For example,
> ProducerInterceptors is automatically loaded when multiple interceptors are
> specified and is an internal class.
I updated the kip to highlight that the changes is updating ProducerConfig and ConsumerConfig to have new config for serializers/deserializers and not the actual class this is implementation details and not public interfaces. 

> JR17. We could estimate the size after compression, but the estimator is
> not 100% accurate. It seems that it's simpler to just use the original
> message size.
We can keep it as original message size, I was thinking if it is good enough for max.request.size it might be good enough for this. I updated the KIP anyway to simplify it and keep it to the original size. 

Hope the final version addressed all feedbacks and we can resume with the voting 

Thanks 
Omnia

> On 8 Aug 2025, at 22:10, Jun Rao <ju...@confluent.io.INVALID> wrote:
> 
> Hi, Omnia,
> 
> Thanks for the reply.
> 
> JR11. Do you think that we need to add key.serializers and
> key.deserializers or do you think covering large messages in value is
> enough?
> 
> JR12. We can load ComposableSerializer automatically if value.serializers
> or value.deserializers are specified. But, it seems that we could keep
> ComposableSerializer as an internal implementation. For example,
> ProducerInterceptors is automatically loaded when multiple interceptors are
> specified and is an internal class.
> 
> JR17. We could estimate the size after compression, but the estimator is
> not 100% accurate. It seems that it's simpler to just use the original
> message size.
> 
> Jun
> 
> On Mon, Aug 4, 2025 at 10:33 AM Omnia Ibrahim <o.g.h.ibrahim@gmail.com <ma...@gmail.com>>
> wrote:
> 
>> 
>> Hi Jun
>>> JR11. value.serializers and value.deserializers: Should they be of type
>>> List? Also, where are key.serializers and key.deserializers?
>>> 
>> Updated now
>>> JR12. Do we still need ComposableSerializer and ComposableDeserializer?
>> The initial thinking here was if `value.serializers or value.deserializers
>> ` exists the client will load  ComposableSerializer or
>> ComposableDeserializer  automatically and use them. Unfortunately this
>> would need us to define these serializers.
>> 
>> The other option is to update `Plugin<Serializer<V>>
>> valueSerializerPlugin` and KafkaProducer constructor to accept
>> List<Serializer<V>> and move the logic of the ComposableSerializer into
>> KafkaProducer::doSend which when we do serialization (same with
>> KafkaConsumer). This option hide the logic and reduce exposure for client
>> to these.
>> WDYT?
>> 
>>> 
>>> JR13. large.message.payload.store.class : should it be of type class?
>> Updated
>>> 
>>> JR14.
>>> 
>> org.apache.kafka.common.serialization.largemessage.LargeMessageSerializer :
>>> The name seems redundant since largemessage appears twice.
>> Updated
>>> 
>>> JR15. PayloadResponse: It still mentions response code. It mentions
>>> "isRetryable flag", which no longer exists in PayloadStoreException.
>> There
>>> are typos in "then it will serialiser will”.
>> The KIP is updated now
>> 
>>> JR16. Regarding returning new byte[0] if
>> large.message.skip.not.found.error
>>> is true, this will likely fail the next deserializer and the application
>>> won't have the right context of the error. It's probably better to just
>>> propagate the specific exception and let the caller handle it.
>> You right this will cause issue if there is another deserializer waiting
>> for the data. I have updated the KIP with this.
>> 
>>> 
>>> JR17. LargeMessageSerializer:  "Check if the estimated size of the data
>>> (bytes) after applying provided compression (if there is one)"
>> Compression
>>> actually happens after serialization and is done on a batch of records.
>> Yes the compression itself happened after but we also have
>> `estimateSizeInBytesUpperBound` which my understanding is this take the
>> compression type into the account as well
>>> 
>>> 
>>> JR18. Could you define the type T for LargeMessageSerializer?
>> Update the KIP this T would be byte[]
>> 
>> Thanks
>> Omnia
>>> 
>>> Jun
>>> 
>>> On Fri, Jul 25, 2025 at 6:28 AM Omnia Ibrahim <o.g.h.ibrahim@gmail.com
>> <ma...@gmail.com>>
>>> wrote:
>>> 
>>>> Hi Jun, thanks for having the time to review this
>>>> 
>>>>> JR1. While the KIP is potentially useful, I am wondering who is
>>>> responsible
>>>>> for retention for the objects in the payload store. Once a message
>> with a
>>>>> reference is deleted, the key of the external object is lost and the
>>>> object
>>>>> may never be deleted.
>>>> 
>>>> The `ttl` in the object store is the responsibility of the owner of this
>>>> store it should be configured in away that is reasonable with the
>> retention
>>>> config in Kafka.
>>>> I have updated the KIP with `Consideration` section.
>>>>> 
>>>>> JR2. Configs: For all new configs, it would be useful to list their
>>>> types.
>>>> Updated the KIP now
>>>>> 
>>>>> JR3. value.serializers: Why is this required? If a user doesn't set it,
>>>> we
>>>>> should just use value.serializer, right? Ditto for key.serializers.
>>>> No you right this was copy/past mistake
>>>> 
>>>>> 
>>>>> JR4. For all new public interfaces such as LargeMessageSerializer,
>>>>> PayloadStore and PayloadResponse, it would be useful to include the
>> full
>>>>> package name.
>>>> Updated the KIP now
>>>>> 
>>>>> JR5. large.message.payload.store.retry.max.backoff.ms and
>>>>> large.message.payload.store.retry.delay.backoff.ms: Is the intention
>> to
>>>>> implement exponential backoff on retries? If so, it's more consistent
>> if
>>>> we
>>>>> can follow the existing naming convention like retry.backoff.max.ms <
>>>> http://retry.backoff.max.ms/> and
>>>>> retry.backoff.ms <http://retry.backoff.ms/> <http://retry.backoff.ms/> <http://retry.backoff.ms/
>>> .
>>>> I have removed these to simplify the config more (as Luke suggested
>>>> initially) and added these to the consideration section.
>>>> 
>>>>> 
>>>>> JR6. large.message.skip.not.found.error : If the reference can't be
>>>> found,
>>>>> what value does the deserializer return? Note that null has a special
>>>>> meaning for tombstone in compacted topics.
>>>> The deserialiser will return `new byte[0]` not null.
>>>>> 
>>>>> JR7. PayloadResponse: Why do we have both responseCode and
>>>>> PayloadStoreException?
>>>> We can do without responseCode, the initial though was to report
>> response
>>>> code form payload store.
>>>> Update the KIP.
>>>>> JR8. Why do we need PayloadStore.metrics? Note that we could monitor
>> the
>>>>> metrics in a plugin through the Monitorable interface.
>>>> Oh nice, I didn’t know about this interface before. Updated the KIP with
>>>> this now.
>>>>> 
>>>>> JR9. Why do we need the protected field
>>>> PayloadStoreException.isRetryable?
>>>> Initial thought here was the serializer can retry the upload. But I have
>>>> removed all the retry logic from serializer and it will be up to the
>>>> PayloadStore provider to implement this if they need it.
>>>>> 
>>>>> JR10. As Luke mentioned earlier, we could turn PayloadStore to an
>>>> interface.
>>>> It is updated now to interface.
>>>> 
>>>> Hope the last version of the KIP is more simpler now
>>>> 
>>>> Thanks
>>>> Omnia
>>>> 
>>>>> On 23 Jul 2025, at 00:43, Jun Rao <jun@confluent.io.INVALID <ma...@confluent.io.INVALID> <mailto:
>> jun@confluent.io.INVALID <ma...@confluent.io.INVALID>>> wrote:
>>>>> 
>>>>> Thanks for the KIP. A few comments.
>>>>> 
>>>>> JR1. While the KIP is potentially useful, I am wondering who is
>>>> responsible
>>>>> for retention for the objects in the payload store. Once a message
>> with a
>>>>> reference is deleted, the key of the external object is lost and the
>>>> object
>>>>> may never be deleted.
>>>>> 
>>>>> JR2. Configs: For all new configs, it would be useful to list their
>>>> types.
>>>>> 
>>>>> JR3. value.serializers: Why is this required? If a user doesn't set it,
>>>> we
>>>>> should just use value.serializer, right? Ditto for key.serializers.
>>>>> 
>>>>> JR4. For all new public interfaces such as LargeMessageSerializer,
>>>>> PayloadStore and PayloadResponse, it would be useful to include the
>> full
>>>>> package name.
>>>>> 
>>>>> JR5. large.message.payload.store.retry.max.backoff.ms and
>>>>> large.message.payload.store.retry.delay.backoff.ms: Is the intention
>> to
>>>>> implement exponential backoff on retries? If so, it's more consistent
>> if
>>>> we
>>>>> can follow the existing naming convention like retry.backoff.max.ms <http://retry.backoff.max.ms/> <
>> http://retry.backoff.max.ms/> <
>>>> http://retry.backoff.max.ms/> and
>>>>> retry.backoff.ms <http://retry.backoff.ms/> <http://retry.backoff.ms/> <http://retry.backoff.ms/
>>> .
>>>>> 
>>>>> JR6. large.message.skip.not.found.error : If the reference can't be
>>>> found,
>>>>> what value does the deserializer return? Note that null has a special
>>>>> meaning for tombstone in compacted topics.
>>>>> 
>>>>> JR7. PayloadResponse: Why do we have both responseCode and
>>>>> PayloadStoreException?
>>>>> 
>>>>> JR8. Why do we need PayloadStore.metrics? Note that we could monitor
>> the
>>>>> metrics in a plugin through the Monitorable interface.
>>>>> 
>>>>> JR9. Why do we need the protected field
>>>> PayloadStoreException.isRetryable?
>>>>> 
>>>>> JR10. As Luke mentioned earlier, we could turn PayloadStore to an
>>>> interface.
>>>>> 
>>>>> Thanks,

Re: [DISCUSS] KIP-1159: Large message reference based Serializer

[Jun Rao <ju...@confluent.io.INVALID>]

Hi, Omnia,

Thanks for the reply.

JR11. Do you think that we need to add key.serializers and
key.deserializers or do you think covering large messages in value is
enough?

JR12. We can load ComposableSerializer automatically if value.serializers
or value.deserializers are specified. But, it seems that we could keep
ComposableSerializer as an internal implementation. For example,
ProducerInterceptors is automatically loaded when multiple interceptors are
specified and is an internal class.

JR17. We could estimate the size after compression, but the estimator is
not 100% accurate. It seems that it's simpler to just use the original
message size.

Jun

On Mon, Aug 4, 2025 at 10:33 AM Omnia Ibrahim <o....@gmail.com>
wrote:

>
> Hi Jun
> > JR11. value.serializers and value.deserializers: Should they be of type
> > List? Also, where are key.serializers and key.deserializers?
> >
> Updated now
> > JR12. Do we still need ComposableSerializer and ComposableDeserializer?
> The initial thinking here was if `value.serializers or value.deserializers
> ` exists the client will load  ComposableSerializer or
> ComposableDeserializer  automatically and use them. Unfortunately this
> would need us to define these serializers.
>
> The other option is to update `Plugin<Serializer<V>>
> valueSerializerPlugin` and KafkaProducer constructor to accept
> List<Serializer<V>> and move the logic of the ComposableSerializer into
> KafkaProducer::doSend which when we do serialization (same with
> KafkaConsumer). This option hide the logic and reduce exposure for client
> to these.
> WDYT?
>
> >
> > JR13. large.message.payload.store.class : should it be of type class?
> Updated
> >
> > JR14.
> >
> org.apache.kafka.common.serialization.largemessage.LargeMessageSerializer :
> > The name seems redundant since largemessage appears twice.
> Updated
> >
> > JR15. PayloadResponse: It still mentions response code. It mentions
> > "isRetryable flag", which no longer exists in PayloadStoreException.
> There
> > are typos in "then it will serialiser will”.
> The KIP is updated now
>
> > JR16. Regarding returning new byte[0] if
> large.message.skip.not.found.error
> > is true, this will likely fail the next deserializer and the application
> > won't have the right context of the error. It's probably better to just
> > propagate the specific exception and let the caller handle it.
> You right this will cause issue if there is another deserializer waiting
> for the data. I have updated the KIP with this.
>
> >
> > JR17. LargeMessageSerializer:  "Check if the estimated size of the data
> > (bytes) after applying provided compression (if there is one)"
> Compression
> > actually happens after serialization and is done on a batch of records.
> Yes the compression itself happened after but we also have
> `estimateSizeInBytesUpperBound` which my understanding is this take the
> compression type into the account as well
> >
> >
> > JR18. Could you define the type T for LargeMessageSerializer?
> Update the KIP this T would be byte[]
>
> Thanks
> Omnia
> >
> > Jun
> >
> > On Fri, Jul 25, 2025 at 6:28 AM Omnia Ibrahim <o.g.h.ibrahim@gmail.com
> <ma...@gmail.com>>
> > wrote:
> >
> >> Hi Jun, thanks for having the time to review this
> >>
> >>> JR1. While the KIP is potentially useful, I am wondering who is
> >> responsible
> >>> for retention for the objects in the payload store. Once a message
> with a
> >>> reference is deleted, the key of the external object is lost and the
> >> object
> >>> may never be deleted.
> >>
> >> The `ttl` in the object store is the responsibility of the owner of this
> >> store it should be configured in away that is reasonable with the
> retention
> >> config in Kafka.
> >> I have updated the KIP with `Consideration` section.
> >>>
> >>> JR2. Configs: For all new configs, it would be useful to list their
> >> types.
> >> Updated the KIP now
> >>>
> >>> JR3. value.serializers: Why is this required? If a user doesn't set it,
> >> we
> >>> should just use value.serializer, right? Ditto for key.serializers.
> >> No you right this was copy/past mistake
> >>
> >>>
> >>> JR4. For all new public interfaces such as LargeMessageSerializer,
> >>> PayloadStore and PayloadResponse, it would be useful to include the
> full
> >>> package name.
> >> Updated the KIP now
> >>>
> >>> JR5. large.message.payload.store.retry.max.backoff.ms and
> >>> large.message.payload.store.retry.delay.backoff.ms: Is the intention
> to
> >>> implement exponential backoff on retries? If so, it's more consistent
> if
> >> we
> >>> can follow the existing naming convention like retry.backoff.max.ms <
> >> http://retry.backoff.max.ms/> and
> >>> retry.backoff.ms <http://retry.backoff.ms/> <http://retry.backoff.ms/
> >.
> >> I have removed these to simplify the config more (as Luke suggested
> >> initially) and added these to the consideration section.
> >>
> >>>
> >>> JR6. large.message.skip.not.found.error : If the reference can't be
> >> found,
> >>> what value does the deserializer return? Note that null has a special
> >>> meaning for tombstone in compacted topics.
> >> The deserialiser will return `new byte[0]` not null.
> >>>
> >>> JR7. PayloadResponse: Why do we have both responseCode and
> >>> PayloadStoreException?
> >> We can do without responseCode, the initial though was to report
> response
> >> code form payload store.
> >> Update the KIP.
> >>> JR8. Why do we need PayloadStore.metrics? Note that we could monitor
> the
> >>> metrics in a plugin through the Monitorable interface.
> >> Oh nice, I didn’t know about this interface before. Updated the KIP with
> >> this now.
> >>>
> >>> JR9. Why do we need the protected field
> >> PayloadStoreException.isRetryable?
> >> Initial thought here was the serializer can retry the upload. But I have
> >> removed all the retry logic from serializer and it will be up to the
> >> PayloadStore provider to implement this if they need it.
> >>>
> >>> JR10. As Luke mentioned earlier, we could turn PayloadStore to an
> >> interface.
> >> It is updated now to interface.
> >>
> >> Hope the last version of the KIP is more simpler now
> >>
> >> Thanks
> >> Omnia
> >>
> >>> On 23 Jul 2025, at 00:43, Jun Rao <jun@confluent.io.INVALID <mailto:
> jun@confluent.io.INVALID>> wrote:
> >>>
> >>> Thanks for the KIP. A few comments.
> >>>
> >>> JR1. While the KIP is potentially useful, I am wondering who is
> >> responsible
> >>> for retention for the objects in the payload store. Once a message
> with a
> >>> reference is deleted, the key of the external object is lost and the
> >> object
> >>> may never be deleted.
> >>>
> >>> JR2. Configs: For all new configs, it would be useful to list their
> >> types.
> >>>
> >>> JR3. value.serializers: Why is this required? If a user doesn't set it,
> >> we
> >>> should just use value.serializer, right? Ditto for key.serializers.
> >>>
> >>> JR4. For all new public interfaces such as LargeMessageSerializer,
> >>> PayloadStore and PayloadResponse, it would be useful to include the
> full
> >>> package name.
> >>>
> >>> JR5. large.message.payload.store.retry.max.backoff.ms and
> >>> large.message.payload.store.retry.delay.backoff.ms: Is the intention
> to
> >>> implement exponential backoff on retries? If so, it's more consistent
> if
> >> we
> >>> can follow the existing naming convention like retry.backoff.max.ms <
> http://retry.backoff.max.ms/> <
> >> http://retry.backoff.max.ms/> and
> >>> retry.backoff.ms <http://retry.backoff.ms/> <http://retry.backoff.ms/
> >.
> >>>
> >>> JR6. large.message.skip.not.found.error : If the reference can't be
> >> found,
> >>> what value does the deserializer return? Note that null has a special
> >>> meaning for tombstone in compacted topics.
> >>>
> >>> JR7. PayloadResponse: Why do we have both responseCode and
> >>> PayloadStoreException?
> >>>
> >>> JR8. Why do we need PayloadStore.metrics? Note that we could monitor
> the
> >>> metrics in a plugin through the Monitorable interface.
> >>>
> >>> JR9. Why do we need the protected field
> >> PayloadStoreException.isRetryable?
> >>>
> >>> JR10. As Luke mentioned earlier, we could turn PayloadStore to an
> >> interface.
> >>>
> >>> Thanks,
>
>

Re: [DISCUSS] KIP-1159: Large message reference based Serializer

[Omnia Ibrahim <o....@gmail.com>]

Hi Jun
> JR11. value.serializers and value.deserializers: Should they be of type
> List? Also, where are key.serializers and key.deserializers?
> 
Updated now
> JR12. Do we still need ComposableSerializer and ComposableDeserializer?
The initial thinking here was if `value.serializers or value.deserializers ` exists the client will load  ComposableSerializer or ComposableDeserializer  automatically and use them. Unfortunately this would need us to define these serializers. 

The other option is to update `Plugin<Serializer<V>> valueSerializerPlugin` and KafkaProducer constructor to accept List<Serializer<V>> and move the logic of the ComposableSerializer into KafkaProducer::doSend which when we do serialization (same with KafkaConsumer). This option hide the logic and reduce exposure for client to these. 
WDYT?

> 
> JR13. large.message.payload.store.class : should it be of type class?
Updated
> 
> JR14.
> org.apache.kafka.common.serialization.largemessage.LargeMessageSerializer :
> The name seems redundant since largemessage appears twice.
Updated 
> 
> JR15. PayloadResponse: It still mentions response code. It mentions
> "isRetryable flag", which no longer exists in PayloadStoreException. There
> are typos in "then it will serialiser will”.
The KIP is updated now 

> JR16. Regarding returning new byte[0] if large.message.skip.not.found.error
> is true, this will likely fail the next deserializer and the application
> won't have the right context of the error. It's probably better to just
> propagate the specific exception and let the caller handle it.
You right this will cause issue if there is another deserializer waiting for the data. I have updated the KIP with this.

> 
> JR17. LargeMessageSerializer:  "Check if the estimated size of the data
> (bytes) after applying provided compression (if there is one)"  Compression
> actually happens after serialization and is done on a batch of records.
Yes the compression itself happened after but we also have `estimateSizeInBytesUpperBound` which my understanding is this take the compression type into the account as well 
> 
> 
> JR18. Could you define the type T for LargeMessageSerializer?
Update the KIP this T would be byte[]

Thanks 
Omnia
> 
> Jun
> 
> On Fri, Jul 25, 2025 at 6:28 AM Omnia Ibrahim <o.g.h.ibrahim@gmail.com <ma...@gmail.com>>
> wrote:
> 
>> Hi Jun, thanks for having the time to review this
>> 
>>> JR1. While the KIP is potentially useful, I am wondering who is
>> responsible
>>> for retention for the objects in the payload store. Once a message with a
>>> reference is deleted, the key of the external object is lost and the
>> object
>>> may never be deleted.
>> 
>> The `ttl` in the object store is the responsibility of the owner of this
>> store it should be configured in away that is reasonable with the retention
>> config in Kafka.
>> I have updated the KIP with `Consideration` section.
>>> 
>>> JR2. Configs: For all new configs, it would be useful to list their
>> types.
>> Updated the KIP now
>>> 
>>> JR3. value.serializers: Why is this required? If a user doesn't set it,
>> we
>>> should just use value.serializer, right? Ditto for key.serializers.
>> No you right this was copy/past mistake
>> 
>>> 
>>> JR4. For all new public interfaces such as LargeMessageSerializer,
>>> PayloadStore and PayloadResponse, it would be useful to include the full
>>> package name.
>> Updated the KIP now
>>> 
>>> JR5. large.message.payload.store.retry.max.backoff.ms and
>>> large.message.payload.store.retry.delay.backoff.ms: Is the intention to
>>> implement exponential backoff on retries? If so, it's more consistent if
>> we
>>> can follow the existing naming convention like retry.backoff.max.ms <
>> http://retry.backoff.max.ms/> and
>>> retry.backoff.ms <http://retry.backoff.ms/> <http://retry.backoff.ms/>.
>> I have removed these to simplify the config more (as Luke suggested
>> initially) and added these to the consideration section.
>> 
>>> 
>>> JR6. large.message.skip.not.found.error : If the reference can't be
>> found,
>>> what value does the deserializer return? Note that null has a special
>>> meaning for tombstone in compacted topics.
>> The deserialiser will return `new byte[0]` not null.
>>> 
>>> JR7. PayloadResponse: Why do we have both responseCode and
>>> PayloadStoreException?
>> We can do without responseCode, the initial though was to report response
>> code form payload store.
>> Update the KIP.
>>> JR8. Why do we need PayloadStore.metrics? Note that we could monitor the
>>> metrics in a plugin through the Monitorable interface.
>> Oh nice, I didn’t know about this interface before. Updated the KIP with
>> this now.
>>> 
>>> JR9. Why do we need the protected field
>> PayloadStoreException.isRetryable?
>> Initial thought here was the serializer can retry the upload. But I have
>> removed all the retry logic from serializer and it will be up to the
>> PayloadStore provider to implement this if they need it.
>>> 
>>> JR10. As Luke mentioned earlier, we could turn PayloadStore to an
>> interface.
>> It is updated now to interface.
>> 
>> Hope the last version of the KIP is more simpler now
>> 
>> Thanks
>> Omnia
>> 
>>> On 23 Jul 2025, at 00:43, Jun Rao <jun@confluent.io.INVALID <ma...@confluent.io.INVALID>> wrote:
>>> 
>>> Thanks for the KIP. A few comments.
>>> 
>>> JR1. While the KIP is potentially useful, I am wondering who is
>> responsible
>>> for retention for the objects in the payload store. Once a message with a
>>> reference is deleted, the key of the external object is lost and the
>> object
>>> may never be deleted.
>>> 
>>> JR2. Configs: For all new configs, it would be useful to list their
>> types.
>>> 
>>> JR3. value.serializers: Why is this required? If a user doesn't set it,
>> we
>>> should just use value.serializer, right? Ditto for key.serializers.
>>> 
>>> JR4. For all new public interfaces such as LargeMessageSerializer,
>>> PayloadStore and PayloadResponse, it would be useful to include the full
>>> package name.
>>> 
>>> JR5. large.message.payload.store.retry.max.backoff.ms and
>>> large.message.payload.store.retry.delay.backoff.ms: Is the intention to
>>> implement exponential backoff on retries? If so, it's more consistent if
>> we
>>> can follow the existing naming convention like retry.backoff.max.ms <http://retry.backoff.max.ms/> <
>> http://retry.backoff.max.ms/> and
>>> retry.backoff.ms <http://retry.backoff.ms/> <http://retry.backoff.ms/>.
>>> 
>>> JR6. large.message.skip.not.found.error : If the reference can't be
>> found,
>>> what value does the deserializer return? Note that null has a special
>>> meaning for tombstone in compacted topics.
>>> 
>>> JR7. PayloadResponse: Why do we have both responseCode and
>>> PayloadStoreException?
>>> 
>>> JR8. Why do we need PayloadStore.metrics? Note that we could monitor the
>>> metrics in a plugin through the Monitorable interface.
>>> 
>>> JR9. Why do we need the protected field
>> PayloadStoreException.isRetryable?
>>> 
>>> JR10. As Luke mentioned earlier, we could turn PayloadStore to an
>> interface.
>>> 
>>> Thanks,

Re: [DISCUSS] KIP-1159: Large message reference based Serializer

[Jun Rao <ju...@confluent.io.INVALID>]

Hi, Omnia,

Thanks for the updated KIP. A few more comments.

JR11. value.serializers and value.deserializers: Should they be of type
List? Also, where are key.serializers and key.deserializers?

JR12. Do we still need ComposableSerializer and ComposableDeserializer?

JR13. large.message.payload.store.class : should it be of type class?

JR14.
org.apache.kafka.common.serialization.largemessage.LargeMessageSerializer :
The name seems redundant since largemessage appears twice.

JR15. PayloadResponse: It still mentions response code. It mentions
"isRetryable flag", which no longer exists in PayloadStoreException. There
are typos in "then it will serialiser will".

JR16. Regarding returning new byte[0] if large.message.skip.not.found.error
is true, this will likely fail the next deserializer and the application
won't have the right context of the error. It's probably better to just
propagate the specific exception and let the caller handle it.

JR17. LargeMessageSerializer:  "Check if the estimated size of the data
(bytes) after applying provided compression (if there is one)"  Compression
actually happens after serialization and is done on a batch of records.

JR18. Could you define the type T for LargeMessageSerializer?

Jun

On Fri, Jul 25, 2025 at 6:28 AM Omnia Ibrahim <o....@gmail.com>
wrote:

> Hi Jun, thanks for having the time to review this
>
> > JR1. While the KIP is potentially useful, I am wondering who is
> responsible
> > for retention for the objects in the payload store. Once a message with a
> > reference is deleted, the key of the external object is lost and the
> object
> > may never be deleted.
>
> The `ttl` in the object store is the responsibility of the owner of this
> store it should be configured in away that is reasonable with the retention
> config in Kafka.
> I have updated the KIP with `Consideration` section.
> >
> > JR2. Configs: For all new configs, it would be useful to list their
> types.
> Updated the KIP now
> >
> > JR3. value.serializers: Why is this required? If a user doesn't set it,
> we
> > should just use value.serializer, right? Ditto for key.serializers.
> No you right this was copy/past mistake
>
> >
> > JR4. For all new public interfaces such as LargeMessageSerializer,
> > PayloadStore and PayloadResponse, it would be useful to include the full
> > package name.
> Updated the KIP now
> >
> > JR5. large.message.payload.store.retry.max.backoff.ms and
> > large.message.payload.store.retry.delay.backoff.ms: Is the intention to
> > implement exponential backoff on retries? If so, it's more consistent if
> we
> > can follow the existing naming convention like retry.backoff.max.ms <
> http://retry.backoff.max.ms/> and
> > retry.backoff.ms <http://retry.backoff.ms/>.
> I have removed these to simplify the config more (as Luke suggested
> initially) and added these to the consideration section.
>
> >
> > JR6. large.message.skip.not.found.error : If the reference can't be
> found,
> > what value does the deserializer return? Note that null has a special
> > meaning for tombstone in compacted topics.
> The deserialiser will return `new byte[0]` not null.
> >
> > JR7. PayloadResponse: Why do we have both responseCode and
> > PayloadStoreException?
> We can do without responseCode, the initial though was to report response
> code form payload store.
> Update the KIP.
> > JR8. Why do we need PayloadStore.metrics? Note that we could monitor the
> > metrics in a plugin through the Monitorable interface.
> Oh nice, I didn’t know about this interface before. Updated the KIP with
> this now.
> >
> > JR9. Why do we need the protected field
> PayloadStoreException.isRetryable?
> Initial thought here was the serializer can retry the upload. But I have
> removed all the retry logic from serializer and it will be up to the
> PayloadStore provider to implement this if they need it.
> >
> > JR10. As Luke mentioned earlier, we could turn PayloadStore to an
> interface.
> It is updated now to interface.
>
> Hope the last version of the KIP is more simpler now
>
> Thanks
> Omnia
>
> > On 23 Jul 2025, at 00:43, Jun Rao <ju...@confluent.io.INVALID> wrote:
> >
> > Thanks for the KIP. A few comments.
> >
> > JR1. While the KIP is potentially useful, I am wondering who is
> responsible
> > for retention for the objects in the payload store. Once a message with a
> > reference is deleted, the key of the external object is lost and the
> object
> > may never be deleted.
> >
> > JR2. Configs: For all new configs, it would be useful to list their
> types.
> >
> > JR3. value.serializers: Why is this required? If a user doesn't set it,
> we
> > should just use value.serializer, right? Ditto for key.serializers.
> >
> > JR4. For all new public interfaces such as LargeMessageSerializer,
> > PayloadStore and PayloadResponse, it would be useful to include the full
> > package name.
> >
> > JR5. large.message.payload.store.retry.max.backoff.ms and
> > large.message.payload.store.retry.delay.backoff.ms: Is the intention to
> > implement exponential backoff on retries? If so, it's more consistent if
> we
> > can follow the existing naming convention like retry.backoff.max.ms <
> http://retry.backoff.max.ms/> and
> > retry.backoff.ms <http://retry.backoff.ms/>.
> >
> > JR6. large.message.skip.not.found.error : If the reference can't be
> found,
> > what value does the deserializer return? Note that null has a special
> > meaning for tombstone in compacted topics.
> >
> > JR7. PayloadResponse: Why do we have both responseCode and
> > PayloadStoreException?
> >
> > JR8. Why do we need PayloadStore.metrics? Note that we could monitor the
> > metrics in a plugin through the Monitorable interface.
> >
> > JR9. Why do we need the protected field
> PayloadStoreException.isRetryable?
> >
> > JR10. As Luke mentioned earlier, we could turn PayloadStore to an
> interface.
> >
> > Thanks,
>
>

Re: [DISCUSS] KIP-1159: Large message reference based Serializer

[Omnia Ibrahim <o....@gmail.com>]

Hi Jun, thanks for having the time to review this 

> JR1. While the KIP is potentially useful, I am wondering who is responsible
> for retention for the objects in the payload store. Once a message with a
> reference is deleted, the key of the external object is lost and the object
> may never be deleted.

The `ttl` in the object store is the responsibility of the owner of this store it should be configured in away that is reasonable with the retention config in Kafka. 
I have updated the KIP with `Consideration` section.
> 
> JR2. Configs: For all new configs, it would be useful to list their types.
Updated the KIP now
> 
> JR3. value.serializers: Why is this required? If a user doesn't set it, we
> should just use value.serializer, right? Ditto for key.serializers.
No you right this was copy/past mistake

> 
> JR4. For all new public interfaces such as LargeMessageSerializer,
> PayloadStore and PayloadResponse, it would be useful to include the full
> package name.
Updated the KIP now 
> 
> JR5. large.message.payload.store.retry.max.backoff.ms and
> large.message.payload.store.retry.delay.backoff.ms: Is the intention to
> implement exponential backoff on retries? If so, it's more consistent if we
> can follow the existing naming convention like retry.backoff.max.ms <http://retry.backoff.max.ms/> and
> retry.backoff.ms <http://retry.backoff.ms/>.
I have removed these to simplify the config more (as Luke suggested initially) and added these to the consideration section. 

> 
> JR6. large.message.skip.not.found.error : If the reference can't be found,
> what value does the deserializer return? Note that null has a special
> meaning for tombstone in compacted topics.
The deserialiser will return `new byte[0]` not null. 
> 
> JR7. PayloadResponse: Why do we have both responseCode and
> PayloadStoreException?
We can do without responseCode, the initial though was to report response code form payload store.
Update the KIP.
> JR8. Why do we need PayloadStore.metrics? Note that we could monitor the
> metrics in a plugin through the Monitorable interface.
Oh nice, I didn’t know about this interface before. Updated the KIP with this now.
> 
> JR9. Why do we need the protected field PayloadStoreException.isRetryable?
Initial thought here was the serializer can retry the upload. But I have removed all the retry logic from serializer and it will be up to the PayloadStore provider to implement this if they need it. 
> 
> JR10. As Luke mentioned earlier, we could turn PayloadStore to an interface.
It is updated now to interface. 

Hope the last version of the KIP is more simpler now 

Thanks 
Omnia

> On 23 Jul 2025, at 00:43, Jun Rao <ju...@confluent.io.INVALID> wrote:
> 
> Thanks for the KIP. A few comments.
> 
> JR1. While the KIP is potentially useful, I am wondering who is responsible
> for retention for the objects in the payload store. Once a message with a
> reference is deleted, the key of the external object is lost and the object
> may never be deleted.
> 
> JR2. Configs: For all new configs, it would be useful to list their types.
> 
> JR3. value.serializers: Why is this required? If a user doesn't set it, we
> should just use value.serializer, right? Ditto for key.serializers.
> 
> JR4. For all new public interfaces such as LargeMessageSerializer,
> PayloadStore and PayloadResponse, it would be useful to include the full
> package name.
> 
> JR5. large.message.payload.store.retry.max.backoff.ms and
> large.message.payload.store.retry.delay.backoff.ms: Is the intention to
> implement exponential backoff on retries? If so, it's more consistent if we
> can follow the existing naming convention like retry.backoff.max.ms <http://retry.backoff.max.ms/> and
> retry.backoff.ms <http://retry.backoff.ms/>.
> 
> JR6. large.message.skip.not.found.error : If the reference can't be found,
> what value does the deserializer return? Note that null has a special
> meaning for tombstone in compacted topics.
> 
> JR7. PayloadResponse: Why do we have both responseCode and
> PayloadStoreException?
> 
> JR8. Why do we need PayloadStore.metrics? Note that we could monitor the
> metrics in a plugin through the Monitorable interface.
> 
> JR9. Why do we need the protected field PayloadStoreException.isRetryable?
> 
> JR10. As Luke mentioned earlier, we could turn PayloadStore to an interface.
> 
> Thanks,

Re: [DISCUSS] KIP-1159: Large message reference based Serializer

[Jun Rao <ju...@confluent.io.INVALID>]

Hi, Omnia,

Thanks for the KIP. A few comments.

JR1. While the KIP is potentially useful, I am wondering who is responsible
for retention for the objects in the payload store. Once a message with a
reference is deleted, the key of the external object is lost and the object
may never be deleted.

JR2. Configs: For all new configs, it would be useful to list their types.

JR3. value.serializers: Why is this required? If a user doesn't set it, we
should just use value.serializer, right? Ditto for key.serializers.

JR4. For all new public interfaces such as LargeMessageSerializer,
PayloadStore and PayloadResponse, it would be useful to include the full
package name.

JR5. large.message.payload.store.retry.max.backoff.ms and
large.message.payload.store.retry.delay.backoff.ms: Is the intention to
implement exponential backoff on retries? If so, it's more consistent if we
can follow the existing naming convention like retry.backoff.max.ms and
retry.backoff.ms.

JR6. large.message.skip.not.found.error : If the reference can't be found,
what value does the deserializer return? Note that null has a special
meaning for tombstone in compacted topics.

JR7. PayloadResponse: Why do we have both responseCode and
PayloadStoreException?

JR8. Why do we need PayloadStore.metrics? Note that we could monitor the
metrics in a plugin through the Monitorable interface.

JR9. Why do we need the protected field PayloadStoreException.isRetryable?

JR10. As Luke mentioned earlier, we could turn PayloadStore to an interface.

Thanks,

Jun


On Fri, Jul 18, 2025 at 11:03 AM Omnia Ibrahim <o....@gmail.com>
wrote:

> Hi everyone,
> If there is no any more feedback need to be addressed then I will open
> voting on this KIP early next week.
>
> Regards
> Omnia
>
> > On 28 May 2025, at 11:13, Omnia Ibrahim <o....@gmail.com> wrote:
> >
> > Hi Luke, I had another round of updating the KIP.
> > I now made `PayloadStore::publish` returns a simplified version of
> “PayloadResponse” which will now will simply have the "fullPayloadPath",
> and “PayloadException”.
> > If PayloadResponse
> > -  contains PayloadStoreException with `isRetriable` flag set to true
> then this will trigger the retry logic
> > - If response code is 404 and large.message.skip.not.found.error then we
> will skip failing on the download in the deserialiser side.
> >
> > WDYT?
> >
> > Regards
> > Omnia
> >
> >
> >> On 22 May 2025, at 15:16, Omnia Ibrahim <o....@gmail.com>
> wrote:
> >>
> >> Hi Luke, sorry for late response.
> >>> IMO, the retry should be the logic inside the
> >>> "large.message.payload.store.class"
> >>> implementation. If we really want it, I think we need to make it clear
> in
> >>> which circumstance we will retry. For example, if it's an unknown
> exception
> >>> thrown from S3 API, what will we do to it?
> >> There was one class definition missing which is PayloadStoreException
> that is used in PayloadResponse. The PayloadStoreException has flag if it
> is true then we retry. And then serialised retry on any exception with this
> flag. We can make publish and download methods throw  PayloadStoreException
> directly to simplify things WDYT?
> >>
> >>> Moving it into the store class makes it much clearer. And it's good to
> have
> >>> a default implementation.
> >>> But to me, it's still an implementation-level detail that we don't
> need to
> >>> expose to users to implement it.
> >>> Could I know more about when the Id generator will be invoked?
> >>> My thought is :
> >>> Users can implement a `publish` method to publish like this:
> >>> public PayloadResponse publish(String topic, byte[] data) {
> >>>  String id = genId();
> >>>  // put the id and data to the remote storage
> >>>  s3Client.put(id, data, ...);
> >>> }
> >>>
> >>> So, with the id method in the interface, who will invoke it? Suppose it
> >>> will be the serializer/deserializer, but no one passes the generated
> id to
> >>> the publish method, how do we use it?
> >>
> >> It will be invoked by the payload class when publish the payload. So it
> is an implementation details of this class yes.
> >>
> >>> 7. Why do we need "PayloadResponse"? Why can't we return a String or a
> URL
> >>> object?
> >>
> >> Originally because I needed something bit of wrapper for re-try and
> skip logic. But as I mentioned in first point we can make publish and
> download methods return the ref or payload otherwise they throw
> PayloadStoreException directly to simplify things WDYT? Other option have
> publish return only ref as string and download would be the one that could
> throw PayloadNotFoundException.
> >>
> >>> 8. Could we change the abstract class to interface?
> >> We can once we agree on what config we will drop and move to the
> details of payload provided by the users.
> >>
> >> regards
> >> Omnia
> >>
> >>> On 19 May 2025, at 09:54, Luke Chen <sh...@gmail.com> wrote:
> >>>
> >>> Hi Omnia,
> >>>
> >>> Thanks for the explanation and update.
> >>> It's better now.
> >>>
> >>> Questions:
> >>>> 2. It's not clear how the "retry count" comes into play in the KIP. It
> >>>> needs more explanation.
> >>> My initial thinking is the retry configs are a must for all blob
> stores, so
> >>> we can provide them, and validate them for free for all blob stores so
> not
> >>> every implementation will go through verifying them.
> >>>
> >>> IMO, the retry should be the logic inside the
> >>> "large.message.payload.store.class"
> >>> implementation. If we really want it, I think we need to make it clear
> in
> >>> which circumstance we will retry. For example, if it's an unknown
> exception
> >>> thrown from S3 API, what will we do to it?
> >>>
> >>>> 6. About the BlobIdGenerator, why do we need it, could you explain
> more?
> >>>> Again, I thought we only need to replace value to a path, and add the
> >>>> "large-message" header, so that when consumer reads this message,
> it'll
> >>>> read the path from value and get the original data via BlobStore. Why
> do
> >>> we
> >>>> need this ID generator? I think users should do the object naming when
> >>>> putting the object by themselves, not via another interface. WDYT?
> >>> In some cases generating the ID might need some smart work for example
> to
> >>> avoid s3 throttling the recommended way on their doc it to create sub
> paths
> >>> under the original bucket, to decide this we might hash the data to
> find a
> >>> suitable sub-path.
> >>> Here is an example of how I would generate an path for s3 file
> >>> ```
> >>> public String id(byte[] data) {
> >>>   String subFolder = topic + "-" +
> Utils.toPositive(Utils.murmur2(data)) %
> >>> distributionFactor // distributionFactor is a config for the Id
> generator
> >>> and it represent the max number of sub-folders under the bucket
> >>>   return subFolder + “/“ + UUID.randomUUID().toString()
> >>> }
> >>> ```
> >>> Hope this example clarify a bit. However I do agree here it might not
> need
> >>> a class. I have move it to be part of the store class.
> >>>
> >>> Moving it into the store class makes it much clearer. And it's good to
> have
> >>> a default implementation.
> >>> But to me, it's still an implementation-level detail that we don't
> need to
> >>> expose to users to implement it.
> >>> Could I know more about when the Id generator will be invoked?
> >>> My thought is :
> >>> Users can implement a `publish` method to publish like this:
> >>> public PayloadResponse publish(String topic, byte[] data) {
> >>>  String id = genId();
> >>>  // put the id and data to the remote storage
> >>>  s3Client.put(id, data, ...);
> >>> }
> >>>
> >>> So, with the id method in the interface, who will invoke it? Suppose it
> >>> will be the serializer/deserializer, but no one passes the generated
> id to
> >>> the publish method, how do we use it?
> >>>
> >>> 7. Why do we need "PayloadResponse"? Why can't we return a String or a
> URL
> >>> object?
> >>>
> >>> 8. Could we change the abstract class to interface?
> >>>
> >>> Thanks.
> >>> Luke
> >>>
> >>>
> >>>
> >>>
> >>> On Wed, Apr 30, 2025 at 9:21 PM Omnia Ibrahim <o.g.h.ibrahim@gmail.com
> <ma...@gmail.com>>
> >>> wrote:
> >>>
> >>>> Hi Luke,
> >>>>
> >>>>>> 3. What does "LargeMessageFormatter" do in the process?
> >>>>>> I thought all we want to do is to replace the "large value data"
> into a
> >>>>>> path, and consumers will read the path via blob store class.
> >>>>>> All these should be done in serializer/deserializer, so why do we
> need
> >>>> the
> >>>>>> formatter?
> >>>>>
> >>>>> I wanted to bit of more info than just the path to download, for
> example
> >>>> I want to add stuff like the class path for the original blob store
> for
> >>>> example if consumer is setup with the unmatched blob store to the one
> used
> >>>> during publishing.
> >>>>> I have updated the KIP to simplify this by having this always as a
> >>>> simple json of path and publisher class which can be represented as
> >>>> PayloadReferenceValue. WDYT?
> >>>>
> >>>> I thought of another case where having the freedom to form the
> reference
> >>>> might be nice feature, which is DR. Let’s imagine this case where
> someone
> >>>> publish large messages to S3 and reference to Kafka topic then they
> want to
> >>>> have DR. This can be achievable if they have mirrored Kafka topic
> which
> >>>> contains the references but if S3 is unreachable form the DR backup
> >>>> location then the reference they have is bit useless. However if the
> >>>> message formatter is customisable then dev can implement a complicated
> >>>> store that publish to two store locations and the publish both
> references
> >>>> to Kafka as one message and the consumer store can download from
> either
> >>>> buckets that are available. I think keeping the door open to such
> use-case
> >>>> might be good feature but also having such use case might be
> questionable a
> >>>> bit with the latency it will add as we will be publishing to N number
> of
> >>>> stores.
> >>>>
> >>>> Regards
> >>>> Omnia
> >>>>
> >>>>> On 24 Apr 2025, at 17:40, Omnia Ibrahim <o....@gmail.com>
> wrote:
> >>>>>
> >>>>> Hi Luke, thanks for having the time to look into the KIP
> >>>>>> 2. It's not clear how the "retry count" comes into play in the KIP.
> It
> >>>>>> needs more explanation.
> >>>>> My initial thinking is the retry configs are a must for all blob
> stores,
> >>>> so we can provide them, and validate them for free for all blob
> stores so
> >>>> not every implementation will go through verifying them.
> >>>>>
> >>>>>> 3. What does "LargeMessageFormatter" do in the process?
> >>>>>> I thought all we want to do is to replace the "large value data"
> into a
> >>>>>> path, and consumers will read the path via blob store class.
> >>>>>> All these should be done in serializer/deserializer, so why do we
> need
> >>>> the
> >>>>>> formatter?
> >>>>>
> >>>>> I wanted to bit of more info than just the path to download, for
> example
> >>>> I want to add stuff like the class path for the original blob store
> for
> >>>> example if consumer is setup with the unmatched blob store to the one
> used
> >>>> during publishing.
> >>>>> I have updated the KIP to simplify this by having this always as a
> >>>> simple json of path and publisher class which can be represented as
> >>>> PayloadReferenceValue. WDYT?
> >>>>>
> >>>>>> 4. In the BlobStore, it looks like we presume users will use object
> >>>> stores,
> >>>>>> which is not good.
> >>>>>> Could we make it more generic? Javadoc, method names, …
> >>>>> This is a good point, I have updated the method names and Javadoc. I
> >>>> also thinking of renaming the class name to PayloadStoreinstead of
> >>>> BlobStore as blob store still tide to object store as well. To set
> some
> >>>> context here, I am proposing this after working with some community
> form
> >>>> Apache Cassandra who are working on Cassandra CEP <
> >>>>
> https://cwiki.apache.org/confluence/display/CASSANDRA/CEP-44%3A+Kafka+integration+for+Cassandra+CDC+using+Sidecar#CEP44:KafkaintegrationforCassandraCDCusingSidecar-LargeBlob
> >-44
> >>>> to handle large CDC and the initial thinking was let’s publish any
> large
> >>>> cdc to an object store instead of Kafka this why the naming was
> suggesting
> >>>> “blob store” only.
> >>>>>
> >>>>>> 5. It would be good to have some explanation for the purpose of
> each new
> >>>>>> interface/class, and clear javadoc for each method.
> >>>>>
> >>>>> Updated the KIP with javadocs
> >>>>>
> >>>>>> 6. About the BlobIdGenerator, why do we need it, could you explain
> more?
> >>>>>> Again, I thought we only need to replace value to a path, and add
> the
> >>>>>> "large-message" header, so that when consumer reads this message,
> it'll
> >>>>>> read the path from value and get the original data via BlobStore.
> Why
> >>>> do we
> >>>>>> need this ID generator? I think users should do the object naming
> when
> >>>>>> putting the object by themselves, not via another interface. WDYT?
> >>>>> In some cases generating the ID might need some smart work for
> example
> >>>> to avoid s3 throttling the recommended way on their doc it to create
> sub
> >>>> paths under the original bucket, to decide this we might hash the
> data to
> >>>> find a suitable sub-path.
> >>>>> Here is an example of how I would generate an path for s3 file
> >>>>> ```
> >>>>> public String id(byte[] data) {
> >>>>>   String subFolder = topic + "-" +
> >>>> Utils.toPositive(Utils.murmur2(data)) % distributionFactor //
> >>>> distributionFactor is a config for the Id generator and it represent
> the
> >>>> max number of sub-folders under the bucket
> >>>>>   return subFolder + “/“ + UUID.randomUUID().toString()
> >>>>> }
> >>>>> ```
> >>>>> Hope this example clarify a bit. However I do agree here it might not
> >>>> need a class. I have move it to be part of the store class.
> >>>>>
> >>>>> Please let me know WDYT of the final shape of the KIP now
> >>>>>
> >>>>> Thanks
> >>>>> Omnia
> >>>>>
> >>>>>> On 24 Apr 2025, at 13:31, Luke Chen <sh...@gmail.com> wrote:
> >>>>>>
> >>>>>> Hi Omnia,
> >>>>>>
> >>>>>> Thanks for proposing this feature that many users expected.
> >>>>>>
> >>>>>> Some comments:
> >>>>>> 1. It's quite interesting to see the idea of chained
> >>>>>> serializer/deserializer used here. I like it.
> >>>>>>
> >>>>>> 2. It's not clear how the "retry count" comes into play in the KIP.
> It
> >>>>>> needs more explanation.
> >>>>>>
> >>>>>> 3. What does "LargeMessageFormatter" do in the process?
> >>>>>> I thought all we want to do is to replace the "large value data"
> into a
> >>>>>> path, and consumers will read the path via blob store class.
> >>>>>> All these should be done in serializer/deserializer, so why do we
> need
> >>>> the
> >>>>>> formatter?
> >>>>>>
> >>>>>> 4. In the BlobStore, it looks like we presume users will use object
> >>>> stores,
> >>>>>> which is not good.
> >>>>>> Could we make it more generic? Javadoc, method names, ...
> >>>>>>
> >>>>>> 5. It would be good to have some explanation for the purpose of
> each new
> >>>>>> interface/class, and clear javadoc for each method.
> >>>>>>
> >>>>>> 6. About the BlobIdGenerator, why do we need it, could you explain
> more?
> >>>>>> Again, I thought we only need to replace value to a path, and add
> the
> >>>>>> "large-message" header, so that when consumer reads this message,
> it'll
> >>>>>> read the path from value and get the original data via BlobStore.
> Why
> >>>> do we
> >>>>>> need this ID generator? I think users should do the object naming
> when
> >>>>>> putting the object by themselves, not via another interface. WDYT?
> >>>>>>
> >>>>>> Thanks.
> >>>>>> Luke
> >>>>>>
> >>>>>>
> >>>>>>
> >>>>>>
> >>>>>>
> >>>>>> On Thu, Apr 10, 2025 at 9:31 PM Omnia Ibrahim <
> o.g.h.ibrahim@gmail.com>
> >>>>>> wrote:
> >>>>>>
> >>>>>>> Hi there I would like to start discussions on
> >>>>>>>
> >>>>
> https://cwiki.apache.org/confluence/display/KAFKA/KIP-1159%3A+Large+message+reference+based+Serializer
> >>>>>>>
> >>>>>>> Thanks
> >>>>>>> Omnia
> >>
> >
>
>