Re: [PR] [WIP] KAFKA-20173: Propagate headers into serde 5/N [kafka]

["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