Tracking issue: #18479
Follows: #38957 (WatermarkManager part 1 in-memory core, merged)
Summary
Wire the part-1 WatermarkManager into the data path so each fused stage computes and forwards its input watermark through it, replacing the provisional "flush the bundle on every received watermark" behavior in ExecutableStageProcessor.
This is the in-JVM wiring. The stage still forwards its output watermark downstream via ctx.forward (now gated by the WatermarkManager so it only forwards when min() advances), so watermarks keep propagating and can be observed in tests. Deferred is only the durable/distributed form of producing that report — flushing it atomically with the EOS offset commit, fanning it out to all downstream partitions, and a serde for it to cross real topic boundaries — which needs the topic-based shuffle (GroupByKey / redistribute-by-key) infra that isn't built yet. See "Out of scope" below.
Scope (this PR)
- Extend the KStreamsPayload watermark variant to carry the in-band report
fields from the agreed design: (sourcePartition, totalSourcePartitions)
alongside the watermark millis.
- ExecutableStageProcessor: hold a WatermarkManager; on a watermark payload,
observe(sourcePartition, watermark, totalSourcePartitions), flush the open
bundle, and forward the stage's output watermark downstream only when
WatermarkManager.advance() moves it forward (monotonic min across the source
partitions), stamped with this stage's own (partition, total). Removes the
provisional flush-on-every-watermark.
- ImpulseProcessor (a source): stamp its terminal TIMESTAMP_MAX_VALUE
watermark with (sourcePartition=0, totalSourcePartitions=1).
Out of scope (later parts, depend on topic-based shuffle)
- Producing the watermark report atomically with the EOS offset commit.
- Fanning the report out to all downstream partitions (vs the single in-JVM
ctx.forward) and a serde for it to cross topic boundaries.
- Real-Kafka-cluster integration tests over the 5 scenarios (steady,
scale-out, clean scale-in, SIGKILL, partition reassignment).
- Watermark holds / persistence and downstream timer firing (needs state +
timers).
Testing
- Processor-level test (TopologyTestDriver / MockProcessorContext) injecting
reports from multiple source partitions: holds until every partition
reports, forwards min(), stays monotonic, re-holds on a partition-count
change.
- End-to-end TopologyTestDriver test: a watermark propagates
Impulse -> ExecutableStage -> a recording sink processor, and the forwarded
terminal TIMESTAMP_MAX_VALUE is captured downstream. (The stage forwards via
ctx.forward as it does today; part 2 just routes it through the
WatermarkManager.)
- ./gradlew :runners:kafka-streams:check green.
cc: @je-ik
Tracking issue: #18479
Follows: #38957 (WatermarkManager part 1 in-memory core, merged)
Summary
Wire the part-1 WatermarkManager into the data path so each fused stage computes and forwards its input watermark through it, replacing the provisional "flush the bundle on every received watermark" behavior in ExecutableStageProcessor.
This is the in-JVM wiring. The stage still forwards its output watermark downstream via ctx.forward (now gated by the WatermarkManager so it only forwards when min() advances), so watermarks keep propagating and can be observed in tests. Deferred is only the durable/distributed form of producing that report — flushing it atomically with the EOS offset commit, fanning it out to all downstream partitions, and a serde for it to cross real topic boundaries — which needs the topic-based shuffle (GroupByKey / redistribute-by-key) infra that isn't built yet. See "Out of scope" below.
Scope (this PR)
fields from the agreed design: (sourcePartition, totalSourcePartitions)
alongside the watermark millis.
observe(sourcePartition, watermark, totalSourcePartitions), flush the open
bundle, and forward the stage's output watermark downstream only when
WatermarkManager.advance() moves it forward (monotonic min across the source
partitions), stamped with this stage's own (partition, total). Removes the
provisional flush-on-every-watermark.
watermark with (sourcePartition=0, totalSourcePartitions=1).
Out of scope (later parts, depend on topic-based shuffle)
ctx.forward) and a serde for it to cross topic boundaries.
scale-out, clean scale-in, SIGKILL, partition reassignment).
timers).
Testing
reports from multiple source partitions: holds until every partition
reports, forwards min(), stays monotonic, re-holds on a partition-count
change.
Impulse -> ExecutableStage -> a recording sink processor, and the forwarded
terminal TIMESTAMP_MAX_VALUE is captured downstream. (The stage forwards via
ctx.forward as it does today; part 2 just routes it through the
WatermarkManager.)
cc: @je-ik