aliehsaeedii commented on code in PR #22155:
URL: https://github.com/apache/kafka/pull/22155#discussion_r4195156815
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/RecordCollectorTest.java:
##########
@@ -43,6 +43,8 @@
import org.apache.kafka.common.serialization.ByteArraySerializer;
import org.apache.kafka.common.serialization.LongSerializer;
import org.apache.kafka.common.serialization.Serializer;
+
+import java.nio.charset.StandardCharsets;
Review Comment:
This import sits in the middle of the `org.apache` block, so `spotlessCheck`
will fail. Please run `./gradlew spotlessApply`.
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorRecordContext.java:
##########
@@ -107,6 +114,15 @@ public byte[] sourceRawValue() {
return sourceRawValue;
}
+ /**
+ * Returns the snapshot of the original source-record headers if one was
captured,
+ * or falls back to {@link #headers()} if no snapshot was set (e.g., when
the context
+ * was initialized from punctuation or synthetic events).
Review Comment:
The fallback also kicks in after `freeRawRecord()`, which runs once the
record is first sent to a sink. After that, a later handler gets the live
headers and null raw key/value. Could the javadoc mention this case too?
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/GlobalStateUpdateTask.java:
##########
@@ -109,7 +111,8 @@ public Map<TopicPartition, Long> initialize() {
@Override
public void update(final ConsumerRecord<byte[], byte[]> record) {
final RecordDeserializer sourceNodeAndDeserializer =
deserializers.get(record.topic());
- final ConsumerRecord<Object, Object> deserialized =
sourceNodeAndDeserializer.deserialize(processorContext, record);
+ final Headers sourceRawHeaders = new RecordHeaders(record.headers());
Review Comment:
No test covers the global snapshot paths (here and in
`GlobalStateManagerImpl#reprocessState`). Please add one where the deserializer
removes a header and the handler still sees it.
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorNode.java:
##########
@@ -310,4 +310,5 @@ public String toString(final String indent) {
}
return sb.toString();
}
+
Review Comment:
nit: stray blank line, please drop it.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
