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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
