[WIP][GSoC 2026] Merge the Kafka Streams runner into master - #39770
Closed
junaiddshaukat wants to merge 39 commits into
Closed
[WIP][GSoC 2026] Merge the Kafka Streams runner into master#39770junaiddshaukat wants to merge 39 commits into
junaiddshaukat wants to merge 39 commits into
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.
…leton module + portable entry points
…anslator (apache#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 apache#18479
…translator (apache#38764) * Add ExecutableStage (stateless ParDo) translator with SDK-harness bridge
…leStage type-agnostic edge (apache#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
…into ExecutableStageProcessor
…rde for crossing topic boundaries
…Window, fire at watermark)
…nner test harness
…Flatten support
…er (PAssert works) (apache#39362)
…atten suites green (apache#39380)
…ner (apache#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 (apache#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.
…ache#39578) * [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 (apache#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 (apache#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 (apache#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 (apache#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 (apache#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 (apache#39748)
…ly in elements (apache#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 (apache#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 (apache#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.
Merges feat/18479-kafka-streams-runner-skeleton. The runner is not part of the standard build: its subprojects are only included with -Pwith-kafka-streams-runner, so nothing reaches anyone who has not asked for it and no release artifact contains it. Removes the feature branch's own CI workflow, which only triggered on feat/18479-* and has no purpose on master. The precommit workflow, which passes the flag, is what remains.
…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.
je-ik
self-requested a review
August 17, 2026 06:52
Contributor
|
R: @je-ik |
Contributor
|
Stopping reviewer notifications for this pull request: review requested by someone other than the bot, ceding control. If you'd like to restart, comment |
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.
junaiddshaukat
force-pushed
the
feat/ks-runner-to-master
branch
from
August 17, 2026 08:20
d54c56e to
c5d8006
Compare
Contributor
|
Closed in favor of github.com//pull/39785. |
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.
Summary
Merges the Kafka Streams runner from its feature branch (
feat/18479-kafka-streams-runner-skeleton) intomaster. Tracking issue: #18479. Developed over this summer as a Google Summer of Code project, mentored by @je-ik.Opened as [WIP] while the lazy consensus thread on dev@ runs. It is not to be merged before that thread closes.
What this is
A portable runner that translates a Beam pipeline into a Kafka Streams topology and runs user code over the Fn API.
What makes it different from the other runners is that Kafka Streams is a library rather than a cluster. There is no job manager and no resource manager to operate: a pipeline is an ordinary JVM process reading from and writing to Kafka, scaled by starting more copies of that process. State, fault tolerance and exactly-once come from Kafka itself, through consumer groups, changelog topics and transactions.
It is not in the standard build
The runner's subprojects are only included when
-Pwith-kafka-streams-runneris passed:Without the flag they are not in the build at all, so nothing reaches anyone who has not asked for it and no release artifact contains it.
beam_PreCommit_Java_Kafka_Streams_Runnerpasses the flag, so the runner is still built and tested on every PR that touches it and cannot rot unnoticed.The point is to give the runner somewhere it can be built, used and worked on, rather than have it quietly die on a branch. If it becomes stable enough the flag comes off; if it does not, it can be dropped without affecting anyone, because no release ever shipped it.
State
Supported, and covered by tests: bounded and unbounded reads, stateless
ParDoincluding multiple outputs,GroupByKeyandCombine, global, fixed and sliding windows with the default trigger and allowed lateness,Flatten,Redistribute, metrics, and exactly-once via Kafka transactions.Known gaps, each tracked: side inputs (#39628), stateful
ParDoand user timers (#39629), merging windows and customWindowFns (#39630), splittableDoFn(#39631),TestStream(#39632), reading a source in parallel (#39626).Known bugs rather than missing features: bundles are not closed after a bounded time (#39633) —
maxBundleTimeMsis accepted and has no effect, because closing a bundle from a wall-clock punctuator duplicated output against a real broker and the cause is not yet understood. There may be others not yet found.Testing
Both pass against current
master. Beam's Python portable suite also runs against the runner, exercising it from a non-Java SDK; it needs a broker, so it is not part of any aggregate build.Checked that the default build is untouched:
./gradlew projectslists no Kafka Streams projects, andjavaPreCommitconfigures with none of its tasks.Changes to the branch as merged
The feature branch's own CI workflow (
beam_KafkaStreamsRunner_FeatureBranch.yml) is removed here, since it only triggered onfeat/18479-*and has no purpose onmaster.beam_PreCommit_Java_Kafka_Streams_Runneris the workflow that remains.