Re: [PR] KAFKA-20462: Preserve source headers in error handlers [kafka]

["aliehsaeedii (via GitHub)" <gi...@apache.org>]

aliehsaeedii commented on code in PR #22155:
URL: https://github.com/apache/kafka/pull/22155#discussion_r3987561039


##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorRecordContext.java:
##########
@@ -107,6 +114,10 @@ public byte[] sourceRawValue() {
         return sourceRawValue;
     }
 
+    public Headers sourceRawHeaders() {
+        return sourceRawHeaders == null ? headers : sourceRawHeaders;

Review Comment:
   Unlike `sourceRawKey()`/`sourceRawValue()`, which return null when unset, this getter falls back to the live `headers`. That's needed so `ProcessorNode` can call it unconditionally, but it's non-obvious — maybe a short javadoc here noting the null-to-live-headers fallback?



##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorNode.java:
##########
@@ -220,7 +220,7 @@ public void process(final Record<KIn, VIn> record) {
                 internalProcessorContext.recordContext().topic(),
                 internalProcessorContext.recordContext().partition(),
                 internalProcessorContext.recordContext().offset(),
-                internalProcessorContext.recordContext().headers(),
+                internalProcessorContext.recordContext().sourceRawHeaders(),

Review Comment:
   The processing handler now gets the source snapshot, but `RecordCollectorImpl#errorHandlerContext` (~line 418) still passes `recordContext.headers()` while copying `sourceRawKey()`/`sourceRawValue()` — so a mutating deserializer still corrupts the DLQ record built on a produce/serialize failure. Should that site use `sourceRawHeaders()` too, or is the production path out of scope for this PR?



##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorContextImpl.java:
##########
@@ -262,7 +262,8 @@ public <K, V> void forward(final Record<K, V> record, final String childName) {
                     recordContext.topic(),
                     record.headers(),
                     recordContext.sourceRawKey(),
-                    recordContext.sourceRawValue()
+                    recordContext.sourceRawValue(),
+                    recordContext.sourceRawHeaders()

Review Comment:
   This rebuild only runs when a forwarded record's headers or timestamp differ, and no test enters it with a snapshot present — the new integration test forwards without changing headers. Good to add a test that forwards with changed headers then fails downstream, to pin that `sourceRawHeaders` survives the rebuild.



##########
streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/ProcessingExceptionHandlerIntegrationTest.java:
##########
@@ -398,6 +407,39 @@ public void shouldStopProcessingWhenFatalUserExceptionProcessingExceptionHandler
         }
     }
 
+    @Test
+    public void shouldExposeOriginalHeadersAfterCachedProcessingError() {

Review Comment:
   This drives the same cached KV-store path as the existing cached-aggregate case in `sourceRawRecordTopologyTestCases()`; the only new thing it checks is the header assertion. Could it fold into that parameterized case (a header-removing serde plus the header assert) instead of a separate test with its own handler and three helpers?



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