mjsax commented on code in PR #22165:
URL: https://github.com/apache/kafka/pull/22165#discussion_r3591119485
##########
streams/src/test/java/org/apache/kafka/streams/kstream/internals/SuppressScenarioTest.java:
##########
@@ -161,6 +164,45 @@ public void shouldImmediatelyEmitEventsWithZeroEmitAfter()
{
}
}
+ @Test
+ public void shouldPropagateHeadersThroughSuppression() {
+ final StreamsBuilder builder = new StreamsBuilder();
+
+ final KTable<String, Long> valueCounts = builder
+ .table(
+ "input",
+ Consumed.with(STRING_SERDE, STRING_SERDE),
+ Materialized.<String, String, KeyValueStore<Bytes,
byte[]>>with(STRING_SERDE, STRING_SERDE)
+ .withCachingDisabled()
+ .withLoggingDisabled()
+ )
+ .groupBy((k, v) -> new KeyValue<>(v, k),
Grouped.with(STRING_SERDE, STRING_SERDE))
+ .count();
+
+ valueCounts
+ .suppress(untilTimeLimit(ZERO, unbounded()))
Review 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.
--
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]
