mjsax commented on code in PR #22165:
URL: https://github.com/apache/kafka/pull/22165#discussion_r3591106823
##########
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(
Review Comment:
Not sure why we read the input as table? Seem unnecessary complex if we
chain a `count()`? Or we just omit the `count()`?
##########
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 first, 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]
