[GSoC 2026] Kafka Streams runner: Python wrapper that starts its own job server

[junaiddshaukat]

Summary

Part of #18479. Follows the review point on #39627: the runner starting its own job server only helped Java, and 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.

python my_pipeline.py \
    --runner=KafkaStreamsRunner \
    --bootstrap_servers=localhost:9092 \
    --application_id=my-beam-pipeline

The job server jar

A portable runner needs its job server to be launchable without a Beam source tree, so
runners/kafka-streams/job-server packages the runner and its dependencies into one shaded jar, in
the same shape as the Flink and Spark job server modules. KafkaStreamsRunner on the Python side
builds it if it has to, starts it, and stops it with the pipeline.

The module carries an SLF4J binding. Without one the jar starts and then logs nothing at all, which
is unhelpful for a process a user runs in the foreground to watch their pipeline — the Spark module
excludes the binding because Spark supplies its own, and copying that left us silent.

The Python runner

KafkaStreamsRunner extends PortableRunner, resolves the jar through JavaJarJobServer, and
defaults the environment to LOOPBACK so a local run needs no Docker. KafkaStreamsRunnerOptions
adds --bootstrap_servers, --application_id and --kafka_streams_job_server_jar, the last of
which takes a prebuilt jar and skips the build.

Testing

Beam's Python is not built in my environment, so rather than assert this works I checked each link
of the chain against something real:

• The shaded jar starts and brings up all three services — job service, artifact staging and
  expansion — on the ports it is given.
• The arguments the Python runner passes (--artifacts-dir, --job-port, --artifact-port,
  --expansion-port) are all accepted by JobServerDriver.ServerConfiguration, and our driver
  overrides none of them. The jar was run with exactly that argument set.
• Against an installed Beam with these two files grafted in, --runner=KafkaStreamsRunner resolves,
  the options parse, default_environment sets LOOPBACK, and java_arguments produces the flags
  above.
• :runners:kafka-streams:job-server:shadowJar, the target the runner asks Beam to build, exists.

What that leaves untested is a Python pipeline actually running through to completion, which needs a
built Beam Python SDK. If CI does not cover that path I would rather it were checked before this is
relied on.

The runner documentation gains a Python section, which I will add to #39627 since the page is still
open for review there.

./gradlew :runners:kafka-streams:validatesRunner        # 59 tests, 0 failures
./gradlew :runners:kafka-streams:build                  # unit tests, spotless + checker + errorprone
./gradlew :runners:kafka-streams:job-server:shadowJar   # the jar

[github-actions]

Checks are failing. Will not request review until checks are succeeding. If you'd like to override that behavior, comment assign set of reviewers

[je-ik]

I think we should also add a test (similar to flink and spark), otherwise this LGMT!

[junaiddshaukat]

I think we should also add a test (similar to flink and spark), otherwise this LGMT!

Added kafka_streams_java_job_server_test.py, following spark_java_job_server_test.py. It covers the job server cache (and that it's keyed only on the two options groups that change the server's configuration), the jar resolution, and the arguments the driver is launched with.

I checked the tests actually fail when the code is wrong rather than just passing — dropping the runner options from the cache key breaks the cache test, and pointing the gradle target elsewhere breaks the jar test.

One thing I want to check with you: flink and spark also have the runner-level suite (spark_runner_test.py / flink_runner_test.py) that extends PortableRunnerTest. That needs a gradle task, a workflow, and a real broker in CI, so I didn't put it in this PR. Should I file it as a follow-up issue, or do you want it here before merging?