[jira] [Commented] (KAFKA-20179) Avoiding headers deserialization while changelogging

["Shekhar Prasad Rajak (Jira)" <ji...@apache.org>]

    [ https://issues.apache.org/jira/browse/KAFKA-20179?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18071936#comment-18071936 ] 

Shekhar Prasad Rajak commented on KAFKA-20179:
----------------------------------------------

I believe the core concern of this ticket -- eager header deserialization during changelogging -- is addressed by KAFKA-20155 https://issues.apache.org/jira/browse/KAFKA-20155

In KAFKA-20155, we introduce LazyHeaders, a lazy Headers implementation that wraps the raw serialized header bytes and defers calling HeadersDeserializer.deserialize() until first read access. The two key wiring points are:

ValueTimestampHeadersDeserializer.deserialize() now returns LazyHeaders instead of eagerly parsed RecordHeaders
Utils.readHeaders() (used by AggregationWithHeadersDeserializer) now returns LazyHeaders as well
This directly affects the changelogging path. Today, ChangeLoggingTimestampedKeyValueBytesStoreWithHeaders, ChangeLoggingTimestampedWindowBytesStoreWithHeaders, and ChangeLoggingSessionBytesStoreWithHeaders all call Utils.headers(valueTimestampHeaders) to extract headers for logChange(). After KAFKA-20155, this call returns a LazyHeaders that:

Does not parse on construction (just stores the raw byte reference)
Does not parse when add() is called (relevant for addVectorClockToHeaders in consistency-enabled setups -- adds are accumulated in a side list)
Only parses when the Kafka producer serializes the changelog record and calls toArray() or iterator() on the headers
So the eager deserialization during changelogging is eliminated. The only remaining cost is the deferred parse when the producer actually needs the individual Header objects, but that is inherent to the Headers API contract -- avoiding it would require changes to the producer internals (accepting raw header bytes), which is outside the scope of KIP-1271.

Given Matthias's earlier comment about ROI concerns, I think KAFKA-20155 provides a good balance: it removes the eagerness without introducing the complexity of a changelog format change or producer API modification.

Suggest we either close this as "Resolved by KAFKA-20155" or keep it open only if we want to pursue the more invasive raw-byte-passthrough approach in the future.

> Avoiding headers deserialization while changelogging
> ----------------------------------------------------
>
>                 Key: KAFKA-20179
>                 URL: https://issues.apache.org/jira/browse/KAFKA-20179
>             Project: Kafka
>          Issue Type: Sub-task
>          Components: streams
>            Reporter: Alieh Saeedi
>            Priority: Major
>
> When writing to the changelog topic, we currently need to deserialize the record headers, which is not ideal from a performance perspective. We should look for a way to avoid this overhead.
> One option, as suggested by [~mjsax] , would be to reuse the header bytes from the Kafka message format directly, without deserialization. However, this would couple Kafka Streams tightly to the internal serialization format of headers in the Kafka record. Although we already use the same format today, we are not yet _formally_ coupled to it. If the Kafka message format for headers ever changes, this approach would directly impact Kafka Streams and raise compatibility concerns (ref: [https://github.com/apache/kafka/pull/21345/changes#r2761685550)|https://github.com/apache/kafka/pull/21345/changes#r2761685550].
> Another complication is that the processor context can still mutate headers—for example, by adding new entries—before the record is sent to the producer (see {{{}[ProcessorContextImpl#log(): https://github.com/apache/kafka/blob/6d9ba767c544d600acbcc1eec8dd38d94c739b01/streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorContextImpl.java#L136{}}}). This leaves us with two options:
>  # Deserialize the headers anyway, which means we lose the optimization; or
>  # Manually merge: serialize any newly added header entries (e.g., the vector clock), append them to the existing serialized headers, and then send the combined header bytes to the producer.
> Neither option is entirely straightforward, so we need to weigh the performance gains of avoiding deserialization against the tighter coupling and added complexity this would introduce.



--
This message was sent by Atlassian Jira
(v8.20.10#820010)