[GSoC 2026] Kafka Streams runner: multi-output executable stages#39410
Conversation
A fused executable stage can have more than one output (a DoFn with side outputs, or a Read whose SDF wrapper produces several). The runner previously rejected these; it now routes each output to its own downstream. Each output PCollection of a multi-output stage gets a relay node (StageOutputProcessor) that stands in as that output's producer, so downstream transforms have a producer node to wire to at translation time. The stage tags each harness output with its PCollection id and forwards it to the matching relay by name (in-process routing, no topic); the relay forwards data through and re-stamps the stage's watermark with its own id so downstream watermark aggregation stays consistent. A single-output stage is unchanged -- it forwards to its one downstream directly. Enables GroupByKeyTest (its PAssert path produces multi-output stages) and un-sickbays the multi-consumer Flatten test. Adds MultiOutputStageTest, a DoFn with a side output whose two outputs each feed a separate GroupByKey. The remaining GroupByKey failures are non-global windowing (sickbayed).
Summary of ChangesHello, I'm Gemini Code Assist1! I'm currently reviewing this pull request and will post my feedback shortly. In the meantime, here's a summary to help you and other reviewers quickly get up to speed! This pull request enables the Kafka Streams runner to handle fused executable stages that produce multiple outputs. Previously, the runner rejected such stages, limiting its ability to support complex pipeline patterns like side outputs. The implementation introduces a relay-based routing mechanism where each output PCollection is assigned a dedicated relay node, ensuring correct downstream wiring and consistent watermark management. Highlights
New Features🧠 You can now enable Memory (public preview) to help Gemini Code Assist learn from your team's feedback. This makes future code reviews more consistent and personalized to your project's style. Click here to enable Memory in your admin console. Using Gemini Code AssistThe full guide for Gemini Code Assist can be found on our documentation page, here are some quick tips. Invoking Gemini You can request assistance from Gemini at any point by creating a comment using either
Customization To customize the Gemini Code Assist for GitHub experience, repository maintainers can create a configuration file and/or provide a custom code review style guide (such as PEP-8 for Python) by creating and adding files to a Limitations & Feedback Gemini Code Assist may make mistakes. Please leave feedback on any instances where its feedback is incorrect or counterproductive. You can react with 👍 and 👎 on @gemini-code-assist comments. If you're interested in giving your feedback about your experience with Gemini Code Assist for GitHub and other Google products, sign up here. Footnotes
|
![gemini-code-assist[bot]](https://avatars.githubusercontent.com/in/956858?s=60&v=4){kind=small}
There was a problem hiding this comment.
Code Review
This pull request implements support for multi-output executable stages in the Kafka Streams runner by introducing a StageOutputProcessor relay node and updating ExecutableStageProcessor and ExecutableStageTranslator to route outputs to their respective downstream nodes. Feedback on the changes suggests using record.withValue(...) instead of manually reconstructing Record objects in both ExecutableStageProcessor and StageOutputProcessor to ensure metadata headers are preserved and to make the code more concise.
Important
The consumer version of Gemini Code Assist on GitHub is being sunset. Starting June 18, 2026, new organization installations will be blocked, and all code review activity will officially cease on July 17, 2026.
For more details on the timeline and next steps, please review the Help Documentation.
| Record<byte[], KStreamsPayload<?>> outputRecord = | ||
| new Record<byte[], KStreamsPayload<?>>( | ||
| record.key(), KStreamsPayload.data(output), record.timestamp())); | ||
| record.key(), KStreamsPayload.data(output.value), record.timestamp()); |
There was a problem hiding this comment.
Using record.withValue(...) is more concise and idiomatic than manually reconstructing the Record object. More importantly, it automatically preserves any metadata headers present on the original input record, which is crucial for features like distributed tracing (e.g., OpenTelemetry/Jaeger) or custom interceptors.
Record<byte[], KStreamsPayload<?>> outputRecord =
record.withValue(KStreamsPayload.data(output.value));| ctx.forward( | ||
| new Record<byte[], KStreamsPayload<?>>( | ||
| record.key(), | ||
| KStreamsPayload.watermark(report.getWatermarkMillis(), transformId, 0, 1), | ||
| record.timestamp())); |
There was a problem hiding this comment.
Using record.withValue(...) is more concise and idiomatic than manually reconstructing the Record object. It also ensures that any metadata headers present on the original record are preserved when forwarding the watermark payload.
ctx.forward(
record.withValue(
KStreamsPayload.watermark(report.getWatermarkMillis(), transformId, 0, 1)));
|
Assigning reviewers: R: @jrmccluskey added as fallback since no labels match configuration Note: If you would like to opt out of this review, comment Available commands:
The PR bot will only process comments in the main thread (not review comments). |
je-ik
left a comment
je-ik
left a comment
There was a problem hiding this comment.
I'm not sure if we correctly propagate the watermark. Maybe we could add a test that will verify it? We would need multiple instances of the multi-output transform running in parallel.
… relay StageOutputProcessor re-stamped the watermark as partition 0 of 1, which drops the reporting stage instance's partition and count. That is fine while the stage runs as a single instance, but once it runs in parallel a downstream aggregator could no longer infer how many instances produced the output. The relay now relabels only the transform id and keeps the incoming source_partition and total_partitions. Adds StageOutputProcessorTest, which feeds watermark reports from several stage partitions through a MockProcessorContext (a single-instance TopologyTestDriver cannot) and asserts they stay distinct downstream. Also simplifies MultiOutputStageTest to assert one branch directly and route only the other through a GroupByKey.
fca1435
into
apache:feat/18479-kafka-streams-runner-skeleton
2 participants
Summary
A fused executable stage can have more than one output — a DoFn with side outputs, or a Read whose SDF wrapper produces several. The runner rejected multi-output stages; this routes each output to its own downstream.
How it works:
(
StageOutputProcessor) that stands in as that output's producer. Downstreamtransforms are wired to a producer node by PCollection id, and that node must
exist when the stage is translated (before its consumers are), so the relay
gives each output a stable producer to wire to.
the matching relay with
ctx.forward(record, childName)— in-process routingto a named processor node, no extra topic. The relay forwards data unchanged
and re-stamps the stage's watermark with its own id so a downstream watermark
aggregator sees a consistent source.
and registers itself as that output's producer.
An output that feeds a GroupByKey is stored in that GroupByKey's own repartition topic (the shuffle creates it for the key change); an output feeding a stateless ParDo flows in-process. Discussed the storage with @je-ik on Slack.
Result on Beam's suite:
GroupByKeyTestbasic tests now pass (their PAssert path produces multi-output stages), and the multi-consumerFlattenTestis un-sickbayed (14/14).CreateTeststays 5/5. AddsMultiOutputStageTest: a DoFn with a side output whose two outputs each feed a separate GroupByKey.The remaining
GroupByKeyTestfailures are non-global windowing (aGlobalWindowcast) — sickbayed until windowing lands.