junaiddshaukat opened a new pull request, #39700: URL: https://github.com/apache/beam/pull/39700
## Summary Part of #18479. A bounded pipeline never finished against a real broker: Kafka Streams runs a topology until something closes the client, and nothing did. The pipeline produced the right answer and then sat there, and `waitUntilFinish()` blocked for ever. This was invisible to every existing suite. The `@ValidatesRunner` tests run on `TopologyTestDriver`, which is synchronous and reports `DONE` unconditionally; the broker integration tests poll a counter and then `cancel()`, so none of them ever waits for a pipeline to finish on its own. ## How termination is decided Every processor already emits `TIMESTAMP_MAX_VALUE` once its input is exhausted, so the runner knows when it is drained — it just did nothing with it. Each processor now schedules its own termination when it emits that watermark, and the client is closed once they have all reported. Scheduling rather than reporting inline is what makes it safe: a punctuator runs after the current record has been handled, so flushing a bundle, forwarding downstream and committing all still happen first. It has to be `WALL_CLOCK_TIME`, since no records arrive after the final watermark and stream time would never advance. The interval is 1ms because Kafka Streams rejects anything smaller. Two things that turned out to matter: **It waits for every processor, not the first.** One instance can own tasks from both sides of a repartition topic. The upstream side goes terminal as soon as it has written to the topic while the downstream side still has to consume it, so stopping at the first would discard that work and report a successful run. **It waits until the topology has finished starting.** Processors register as their task is initialized, so mid-startup the registered set is only part of the pipeline. On a short pipeline the source can drain before the stage downstream of the repartition topic exists, and stopping there reports success having produced nothing. No coordination between instances is needed. Watermarks crossing a repartition topic are broadcast to every partition, so every task observes the terminal watermark wherever it runs, and each instance reaches the same conclusion on its own. ## `run()` now blocks `JobInvocation` reads the pipeline result's state once, when `run()` returns. Ours returned right after `start()`, so a job stayed `RUNNING` for ever even after the client had stopped cleanly. It now blocks until the pipeline finishes, which is what `FlinkPipelineRunner` does by blocking in `executor.execute()`. This is a deliberate contract change: an unbounded pipeline blocks the calling thread until the job is cancelled. That is the same behaviour Flink has, and the job service already runs `run()` on its own executor and interrupts it to cancel — which the runner now handles by closing the client when it sees the interrupt. ## An unrelated bug this turned up Repartition topics were named `__beam_gbk_<transformId>`, with no application id, while the Impulse and Read bootstrap topics already include one. Transform ids come from the pipeline's structure, so two jobs running the same pipeline shuffled through the same topic and read each other's data. Because the topic is only created when it does not already exist, the second job also silently inherited the first one's partition count instead of the one it asked for. It surfaced here because the new integration test runs the same pipeline at a different parallelism to an existing one, which no two tests had done before. It is a small fix and independent of the rest of this PR — happy to split it out if you would rather review it separately. Note that this renames the topics, so a job restarted after this change shuffles through new ones and leaves its old repartition topics behind. That seems fine for a runner that is on a feature branch and marked experimental, but it is a behaviour change rather than a pure fix. ## Testing ``` ./gradlew :runners:kafka-streams:build # 95 unit tests, spotless + checker + errorprone ./gradlew :runners:kafka-streams:validatesRunner # 59 tests ./gradlew :runners:kafka-streams:brokerIntegrationTest # 4 tests ``` `TerminationTrackerTest` covers the decision logic, including the two cases above. I checked each fails without the code that makes it pass, rather than trusting that they were green. `KafkaStreamsRunnerBrokerIT.aBoundedPipelineTerminatesOnItsOwn` is the end-to-end one: two chained GroupByKeys across four partitions, and nothing cancels it, so returning from `run()` at all is the assertion. It also asserts the output is produced exactly once, because termination rides a wall-clock punctuator and that is the same mechanism that duplicates output when it is used to close bundles on time (#39633). The broker suite also got faster — 2m56s to 31s — since the topic fix stopped the tests inheriting each other's partition counts. Beyond the suites, a Beam **Python** pipeline now runs and finishes on its own against a real broker: `Create -> Map`, terminating in about ten seconds, reporting `DONE`, with the expected output. Before this it hung indefinitely. -- 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]
