[WIP] Merge Kafka Streams Runner skeleton - #39785
Open
je-ik wants to merge 45 commits into
Open
Conversation
- cancel(): wrap delegate.cancel() in try/finally so the embedded job server is always stopped, even if cancellation throws IOException. - waitUntilFinish(Duration): only stop the job server when the returned state is terminal, so a timed-out wait does not prematurely kill the job server while the pipeline is still running. - waitUntilFinish(): wrap in try/finally for the same defensive cleanup reason as cancel(). - KafkaStreamsPipelineOptions.StateDirDefaultFactory: include the job name in the default Kafka Streams state directory so that multiple pipelines on the same host (e.g. parallel tests) do not collide and hit a LockException.
…module + portable entry points
…anslator (#38689) * Add Impulse translator and URN-dispatch framework - KafkaStreamsPipelineTranslator now walks the pipeline in topological order via QueryablePipeline and dispatches each transform to a PTransformTranslator keyed by URN. Unknown URNs still fail fast with a clear "No translator registered for URN ..." message. - ImpulseTranslator implements beam:transform:impulse:v1 per design doc §4.1: a per-application bootstrap topic source (__beam_impulse_<app>) satisfies Kafka Streams' real-source requirement, ImpulseProcessor emits exactly one WindowedValue<byte[]> in the GlobalWindow via a one-shot wall-clock punctuator scheduled on init, and a persistent state store records a "fired" flag so task restarts do not duplicate. Bootstrap-topic auto-creation is deferred to a follow-up sub-issue (design doc §12.1); the topic is expected to pre-exist in production. - KafkaStreamsTranslationContext now holds the Topology being built and a PCollection-id -> processor-name map so downstream translators can wire to their parent nodes. - KafkaStreamsPipelineRunner.run now translates and starts the KafkaStreams application, returning a KafkaStreamsPortablePipelineResult that maps KafkaStreams.State to Beam's PipelineResult.State. Forces processing.guarantee=exactly_once_v2. - Tests: * KafkaStreamsPipelineTranslatorTest also covers the Impulse success path; the unsupported-URN check now uses GroupByKey. * ImpulseTranslatorTest exercises the topology via TopologyTestDriver: exactly one empty byte[] in GlobalWindow is emitted, and a second wall-clock advance does not re-emit. * Address review feedback on translation framework + Impulse PR - KafkaStreamsPortablePipelineResult: close the race where KafkaStreams could transition to a terminal state before the state listener was registered, leaving waitUntilFinish() to block forever. Also add a volatile cancelled flag so that getState() returns State.CANCELLED after a user cancel(), instead of mapping NOT_RUNNING to State.DONE. - ImpulseProcessor: capture the Cancellable returned by context.schedule and cancel the wall-clock punctuator once the impulse has fired (or if the state store already records a prior emission), so the processor stops doing periodic state-store lookups for the lifetime of the task. - KafkaStreamsPipelineRunner.run: invoke PipelineOptionsValidator.validate on the pipeline options at the start of run() so a missing required option (e.g. applicationId) fails with a clear IllegalArgumentException rather than a raw NullPointerException on Properties.put further down. - ImpulseTranslatorTest: wrap CapturingProcessor.received in Collections.synchronizedList for best-practice thread-safety even though TopologyTestDriver runs single-threaded. * Address review feedback on Impulse translator
Beam's standard PreCommit workflows gate on self-hosted runners and a branches: [master, release-*] filter, so they never trigger for PRs targeting the feat/18479-* integration branch. This lightweight github-hosted workflow bypasses that: checkout -> JDK 11 -> ./gradlew :runners:kafka-streams:build. To be removed when the runner work merges to master and the standard PreCommit takes over. Refs #18479
…translator (#38764) * Add ExecutableStage (stateless ParDo) translator with SDK-harness bridge
…leStage type-agnostic edge (#38843) * Add Redistribute (arbitrarily) translator + type-agnostic ExecutableStage edge * Drain pendingOutputs via poll() loop instead of forEach + clear * Assert ChainedExecutableStageTest pipeline actually has two ExecutableStages
…xecutableStageProcessor
…r crossing topic boundaries
…, fire at watermark)
…er (PAssert works) (#39362)
…atten suites green (#39380)
…ner (#39494) * [GSoC 2026] Kafka Streams runner: windowed GroupByKey via ReduceFnRunner Replaces the global-window-only GroupByKey with a windowed one that drives Beam's ReduceFnRunner on the runner side, the way the Flink and Spark portable runners do, backed by Kafka Streams state and timers. WindowedGroupByKeyProcessor builds a ReduceFnRunner per key (like GroupAlsoByWindowViaWindowSetNewDoFn) over two new backends: KafkaStreamsStateInternals, which stores each Beam state cell as one entry in a KeyValueStore under a composite key of key + namespace + tag (modelled on SparkStateInternals), and KafkaStreamsTimerInternals, which persists timers keyed by identity and is fired by the processor scanning for due event-time timers on each input-watermark advance. GroupByKeyTranslator hydrates the input windowing strategy from the pipeline proto and wires the state and timer stores. Windowing, the default trigger, panes, allowed lateness and timestamp combiners all come from ReduceFnRunner.
…oss partitions (#39546) * [GSoC 2026] Kafka Streams runner: run against a real broker Adds what the runner needs to execute on an actual Kafka cluster, and an integration test that runs a pipeline through the production KafkaStreamsPipelineRunner against a Kafka container. Everything until now ran through TopologyTestDriver, which fakes the topics and never builds a KafkaStreams application, so the production path had not been executed.
…9578) * [GSoC 2026] Kafka Streams runner: bound a bundle by element count A bundle stayed open until the next watermark, so on a stream that produces steadily it grew without limit and nothing it had already processed was emitted until a watermark happened to arrive. maxBundleSize was declared as a pipeline option but nothing read it.
… follow-ups (#39610) * [GSoC 2026] Kafka Streams runner: CombineTest coverage and two review follow-ups Enables CombineTest in the ValidatesRunner suite, taking it from 49 to 59 tests. Combine was expected to work without a translator of its own, since the fuser expands Combine.perKey into a GroupByKey with the combining logic running as ordinary ParDos in the SDK harness, but nothing exercised that. BasicTests passes in full, including hot-key fanout and the accumulation-mode variant, and WindowingTests contributes the fixed-window and empty-window cases. The remainder falls out on category excludes the task already declares. testSessionsCombine is sickbayed alongside the existing merging windows entry, and it is the only Combine failure. Corrects the Flatten partition-count comment, which asserted that the inputs are co-partitioned and so implied the Math.max over them was redundant. Neither half held. The max is not a no-op in principle: Kafka Streams merges the subtopologies of every parent a processor is wired to and gives the result as many tasks as its largest source topic has partitions. But the mismatched shape does not reach this translator, because the fuser folds such a Flatten into the harness stage, and the runner Flattens that do arrive come from the fuser deduplicating partial outputs of one PCollection. FlattenParallelismTest records that, so a change letting the mismatched shape through starts failing there rather than producing a pipeline that stalls waiting for a watermark report that never comes. Guards the null record key in GroupByKeyBroadcastPartitioner.partitions(). partition() already guarded it, but partitions() is the method Kafka Streams calls and it hashed the key unguarded.
[GSoC 2026] Kafka Streams runner: read unbounded sources
…ntal (#39627) * [GSoC 2026] Kafka Streams runner: user documentation, marked experimental Adds the runner's documentation page, linked from the runners menu: what the runner is and why someone would choose it, how to start the job server and submit a pipeline, every pipeline option with its default, the internal topics it creates, and what is and is not supported. The unsupported list is specific rather than a general disclaimer, since these are core parts of the Beam model rather than nice-to-haves: side inputs, stateful ParDo and user timers, merging windows, custom WindowFns, splittable DoFn, TestStream, reading a source in parallel, the bundle time bound, finalizeCheckpoint, and committed metrics. Each says what it means for a user.
…job server (#39680) * [GSoC 2026] Kafka Streams runner: Python wrapper that starts its own job server The runner starting its own job server only helped Java, so a Python user still had to run one by hand. This adds the wrapper Flink and Spark provide, so a Python pipeline can select the runner and nothing else.
…t is drained (#39700) * [GSoC 2026] Kafka Streams runner: terminate a bounded pipeline when it is drained Kafka Streams runs a topology until something closes the client, so a bounded pipeline produced its output and then ran for ever. Every processor already emits TIMESTAMP_MAX_VALUE once its input is exhausted, so each one now schedules its own termination when it emits that watermark, and the client is closed once they have all reported. Termination is scheduled rather than reported inline so that the work which follows the final watermark still runs. The callback waits for every processor rather than the first, because one instance can own both sides of a repartition topic, and it waits until the topology has finished starting, because processors register as their tasks are initialized.
…Python (#39736) * [GSoC 2026] Kafka Streams runner: portable ValidatesRunner suite for Python Runs Beam's portable ValidatesRunner suite against the runner, which is what shows it works for a pipeline that was not written in Java. 29 tests pass and 46 are skipped, each skip naming the issue for the feature it needs. * [GSoC 2026] Kafka Streams runner: integration test for two instances over three partitions
…m the bundle size, and expose the session timeout (#39748)
…ly in elements (#39761) * [GSoC 2026] Kafka Streams runner: bound a source poll in time, not only in elements --readMaxPollTimeMs bounds the turn in time as well; whichever bound comes first ends it.
… wrapper (#39766) A Read expands into a splittable DoFn by default and the runner cannot translate one, so a pipeline that merely reads failed to translate unless it knew to convert the reads itself. The wrapper now sets use_deprecated_read and converts the pipeline before handing it on, so a pipeline does not have to know, and a pipeline that asks for splittable reads still gets primitive ones rather than something that cannot run.
…ces coming and going (#39752) * [GSoC 2026] Kafka Streams runner: an application for measuring instances coming and going * [GSoC 2026] Kafka Streams runner: count groups in the pipeline rather than beside it SpotBugs is turned off for this module. It runs the pipeline in process, so the SDK harness and its dependencies are on the classpath and SpotBugs reports on those instead of on the four classes here. The it/ modules do the same for the same reason.
…g for master CI The measurement docker-compose file had no Apache license header, which RAT rejects, and one line in the Python wrapper was not as yapf formats it. Neither ran on the feature branch, whose CI only built the runner.
Several class comments explained more than they needed to, which makes them less likely to be read rather than more. Shortened the longest, keeping the reasoning and dropping the retelling. Also corrects four that had gone stale: two translators claiming topics are not created automatically, the payload claiming its serde does not exist yet, and the read translator pointing at the test runner for a conversion the runner now does itself.
…comments, and fix the license header and Python formatting
The entry still described the first skeleton PR — a module whose translation failed fast on every unsupported URN — which has not been true for months. Describes the runner as it is, and says it is opt-in.
…he CHANGES.md entry
…79-kafka-streams-runner-skeleton
je-ik
force-pushed
the
feat/18479-kafka-streams-runner-skeleton
branch
from
August 18, 2026 07:49
5ce92d4 to
89617f5
Compare
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
This is final merge of GSoC Kafka Streams project to
apache:master. All code has been reviewed when pulled into this branch.Thank you for your contribution! Follow this checklist to help us incorporate your contribution quickly and easily:
addresses #123), if applicable. This will automatically add a link to the pull request in the issue. If you would like the issue to automatically close on merging the pull request, commentfixes #<ISSUE NUMBER>instead.CHANGES.mdwith noteworthy changes.See the Contributor Guide for more tips on how to make review process smoother.
To check the build health, please visit https://github.com/apache/beam/blob/master/.test-infra/BUILD_STATUS.md
GitHub Actions Tests Status (on master branch)
See CI.md for more information about GitHub Actions CI or the workflows README to see a list of phrases to trigger workflows.