KAFKA-20413: Make Streams suppress headers-aware#22165
Conversation
aliehsaeedii
changed the title
aa6f424 to
d02cd7f
Compare
mjsax
left a comment
mjsax
left a comment
There was a problem hiding this comment.
The fix is surprisingly simple... Should we add some TTD test (or even integration test) for some e2e verification?
Add shouldPropagateHeadersThroughSuppression to SuppressScenarioTest: it drives a table/groupBy/count/suppress topology through TopologyTestDriver with a header-bearing input record and asserts the header survives on the suppressed output, giving end-to-end coverage of the headers-aware buffer put/evict path. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
ac1f21d to
e1a0a65
Compare
Good point. Added. |
|
Build failed because of unrelated flaky tests! |
| .count(); | ||
|
|
||
| valueCounts | ||
| .suppress(untilTimeLimit(ZERO, unbounded())) |
There was a problem hiding this comment.
I don't think this work correctly. If we set ZERO we flush the suppress buffer right away, and even in the old code, headers would get carried forward via the context... (I just ran this this unmodified on trunk and it also passed).
Instead, we need to set an actual suppress time and pipe two input record. The second record (with larger ts) would flush the suppress buffer emitting the first. W/o the fix, the headers from the second record would carry forward into the result, but with the fix, the correct headers from the suppressed record should be there.
There was a problem hiding this comment.
I verified the reworked scenario (2-record) and it still passes on trunk. The question is whether TTD is propagating/mutating the specific ProcessorRecordContext instance that gets stored inside k1's BufferValue when k1 (1st input record) is buffered? I assume not. I think I must remove the e2e test since it does not prove anything.
| valueCounts | ||
| .suppress(untilTimeLimit(ZERO, unbounded())) | ||
| .toStream() | ||
| .to("output-suppressed", Produced.with(STRING_SERDE, Serdes.Long())); |
There was a problem hiding this comment.
I am also wondering, if we should add another process() step before we write to the output topic, to access context.headers() (and maybe add information about it to the record's value, so we can capture it on the output topic).
When the suppress() flush happens while processing the second record, the context should not have been set to the flushed/first record's header I believe, but should be the headers from the second record triggering the flush? Not 100% sure.
| time, | ||
| serializedKey, | ||
| new BufferValue(serializedPriorValue, serialChange.oldValue, serialChange.newValue, recordContext) | ||
| new BufferValue(serializedPriorValue, serialChange.oldValue, serialChange.newValue, effectiveContext) |
There was a problem hiding this comment.
I am confuse the this change -- when we do a put() we need to put the headers for the new record, not the old record, right?
I did more digging and it seems we need a different fix... What the buffer stores is an "oldValue" and a "newValue" but only one record-context for both. Your change is changing what we put into the record context, but I believe that is not what we should change -- the context should be from the currently processes record.
Thus, we actually need to change the serialization format of oldValue and newValue -- atm, we serializer plain value types, but we would need to start using ValueTimestampHeader type (this would also fix another issue that we atm also lose the correct record timestamp via suppress). This allows us to store both header for old and new and get the correct old headers into the output on eviction.
So we need to update keySerde and valueSerde and create a corresponding ValueTimestampHeaders object here that we put into the suppress buffer.
For backward compatibility, we should also do this only if header stores are enable I believe? Otherwise, we would increase users' in-memory consumption significantly.
4 participants
This PR fixes the suppress-buffer part of
KAFKA-20413:
InMemoryTimeOrderedKeyValueChangeBuffernow uses the buffered record'sown headers in serde calls and on eviction.
Testing: Added
shouldPropagateHeadersThroughEviction, andshouldUseRecordHeadersNotProcessorContextHeadersOnPut.Reviewers: Matthias J. Sax matthias@confluent.io