Topology in question:
builder.<String, String>stream(someTopic)
.filter((k, v) -> !k.equals("heartbeat"))
.filter((k, v) -> v != null)
.filter(this::isRedactedInstance)
// update our current cache of "known" redacted records
.peek(this::updateRedactedCache)
// change key so we can properly do the join further down
.selectKey(this::getKeyFromRedacted)
// now we slice the stream into 10 sec long slices and aggregate by key
// Our redacted comes in every 10 seconds, so we should never get more than one set of redacted per cluster
.groupByKey(Grouped.with(Serdes.String(), Serdes.String()))
.windowedBy(TimeWindows.of(Duration.ofSeconds(10)).advanceBy(Duration.ofSeconds(10)).grace(Duration.ofSeconds(3)))
.aggregate(() -> null, this::aggregateRedactedByKey)
// the suppress allows us to ignore intermediate records and only get the final 10 second worth aggregation
.suppress(Suppressed.untilWindowCloses(Suppressed.BufferConfig.unbounded()))
.toStream()
.peek(this::deleteme)
The unit test being used to test is:
TopologyTestDriver driver = new TopologyTestDriver(redacted.getMainTopology(), streamingProperties);
driver.pipeInput(
redactedRecordFactory.create(REDACTED_TOPIC,getRedactedKey(inst1),
getEntityEventString(inst1, instSeed1), 0L));
driver.pipeInput(
redactedRecordFactory.create(REDACTED_TOPIC,getRedactedKey(inst2),
getEntityEventString(inst1, instSeed1), 1L));
driver.pipeInput(
redactedRecordFactory.create(REDACTED_TOPIC,getRedactedKey(inst3),
getEntityEventString(inst3, instSeed3), Duration.ofDays(5).toMillis()));
1) When I run the test in debug mode, with a breakpoint at the "deleteMe" method, it is never hit.
2) When I add breakpoints in the "aggregate" method, they are hit as expected (thrice)
3) If I step through the test slowly enough, I do hit the breakpoint in the "deleteMe" method.
I have tried advancing the wallclock, but I understand that's not relevant to window suppression (and it didn't work besides).
I'm not sure what else to try - I would have expected the third event, with a super long timestamp, to have triggered the suppression.