You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "UladzislauBlok (via GitHub)" <gi...@apache.org> on 2026/03/14 19:49:59 UTC
[PR] KAFKA-20173: Propagate headers into serde 5/N [kafka]
UladzislauBlok opened a new pull request, #21756:
URL: https://github.com/apache/kafka/pull/21756
Follow-up for:
- https://github.com/apache/kafka/pull/21490
- https://github.com/apache/kafka/pull/21536
- https://github.com/apache/kafka/pull/21681
- https://github.com/apache/kafka/pull/21755
This one brings update for `Change` and related classes. **WIP**
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20173: Propagate headers into serde 5/N [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #21756:
URL: https://github.com/apache/kafka/pull/21756#discussion_r2937212270
##########
streams/src/main/java/org/apache/kafka/streams/state/internals/InMemoryTimeOrderedKeyValueChangeBuffer.java:
##########
Review Comment:
Not sure TBH -- looking briefly into git history, it seems to be a historic artifact -- when it was added, it was needed, and when we later got `InternalProcessorContext` it was just never cleaned up.
If you want to explore if it's possible to remove the parameter, feel free to open a PR -- I should be internal changes only.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20173: Propagate headers into serde 5/N [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on PR #21756:
URL: https://github.com/apache/kafka/pull/21756#issuecomment-4064079596
Thanks for the PR. Merged to trunk.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20173: Propagate headers into serde 5/N [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax merged PR #21756:
URL: https://github.com/apache/kafka/pull/21756
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20173: Propagate headers into serde 5/N [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #21756:
URL: https://github.com/apache/kafka/pull/21756#discussion_r2937187577
##########
streams/src/main/java/org/apache/kafka/streams/kstream/internals/FullChangeSerde.java:
##########
Review Comment:
Yeah. We have some asymmetry for the header case...
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [WIP] KAFKA-20173: Propagate headers into serde 5/N [kafka]
Posted by "UladzislauBlok (via GitHub)" <gi...@apache.org>.
UladzislauBlok commented on code in PR #21756:
URL: https://github.com/apache/kafka/pull/21756#discussion_r2937053572
##########
streams/src/main/java/org/apache/kafka/streams/state/internals/InMemoryTimeOrderedKeyValueChangeBuffer.java:
##########
Review Comment:
Out of curiosity: why do we propagate `ProcessorRecordContext` if we already have this
`private InternalProcessorContext<?, ?> context;`. I assume if processor was initialized we should have the same task / record.
This is important because this is used to propagate headers to underlying serdes
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [WIP] KAFKA-20173: Propagate headers into serde 5/N [kafka]
Posted by "UladzislauBlok (via GitHub)" <gi...@apache.org>.
UladzislauBlok commented on code in PR #21756:
URL: https://github.com/apache/kafka/pull/21756#discussion_r2936488032
##########
streams/src/main/java/org/apache/kafka/streams/kstream/internals/FullChangeSerde.java:
##########
Review Comment:
I was thinking about that. If the byte array already contains the headers, we should be good. I did it this way because I wanted to be consistent with the serialization path
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [WIP] KAFKA-20173: Propagate headers into serde 5/N [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #21756:
URL: https://github.com/apache/kafka/pull/21756#discussion_r2936122432
##########
streams/src/main/java/org/apache/kafka/streams/kstream/internals/FullChangeSerde.java:
##########
Review Comment:
I think we need to pass in a Headers parameter and use for both old and new value.
The caller should make a decision what to pass. If <T> == HeaderValueTimestamp whatever header get's passed, would be ignored by innerSerializer which would be HeaderValueTimestampSserializer which take the header from data.oldValue / data.newValue which are both of type `HeaderValueTimestamp`.
##########
streams/src/main/java/org/apache/kafka/streams/kstream/internals/Change.java:
##########
@@ -16,21 +16,32 @@
*/
package org.apache.kafka.streams.kstream.internals;
+import org.apache.kafka.common.header.Headers;
+import org.apache.kafka.common.header.internals.RecordHeaders;
+
import java.util.Objects;
public class Change<T> {
public final T newValue;
+ public final Headers newHeaders;
public final T oldValue;
+ public final Headers oldHeaders;
Review Comment:
I don't think we need to add `newHeader` or `oldHeader`. Cf discussion on https://issues.apache.org/jira/browse/KAFKA-20311
##########
streams/src/main/java/org/apache/kafka/streams/kstream/internals/FullChangeSerde.java:
##########
Review Comment:
I think we need to pass in a `Headers` parameter and use for both old and new value.
The caller should make a decision what to pass. For the header-case (ie, if the serialized `byte[]` arrays hold a ValueTimestampeHeader object) whatever header get's passed, would be ignored by `innerDeserializer` which would be `HeaderValueTimestampDeserializer` which take the header from the `byte[]` array.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [WIP] KAFKA-20173: Propagate headers into serde 5/N [kafka]
Posted by "UladzislauBlok (via GitHub)" <gi...@apache.org>.
UladzislauBlok commented on code in PR #21756:
URL: https://github.com/apache/kafka/pull/21756#discussion_r2936491737
##########
streams/src/main/java/org/apache/kafka/streams/kstream/internals/FullChangeSerde.java:
##########
Review Comment:
I see, in case of headers propagation effectively we'll have
'Change<ValueTimestampHeaders<T>>'
And inner serializer will take headers from there. Thanks for clarifying
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [WIP] KAFKA-20173: Propagate headers into serde 5/N [kafka]
Posted by "UladzislauBlok (via GitHub)" <gi...@apache.org>.
UladzislauBlok commented on code in PR #21756:
URL: https://github.com/apache/kafka/pull/21756#discussion_r2936491737
##########
streams/src/main/java/org/apache/kafka/streams/kstream/internals/FullChangeSerde.java:
##########
Review Comment:
I see, in case of headers propagation effectively we'll have
`Change<ValueTimestampHeaders<T>>`
And inner serializer will take headers from there. Thanks for clarifying
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org