This is an automated email from the ASF dual-hosted git repository.
je-ik pushed a commit to branch feat/18479-kafka-streams-runner-skeleton
in repository https://gitbox.apache.org/repos/asf/beam.git
The following commit(s) were added to
refs/heads/feat/18479-kafka-streams-runner-skeleton by this push:
new e051e06bf83 [GSoC 2026] Kafka Streams runner: Python wrapper that
starts its own job server (#39680)
e051e06bf83 is described below
commit e051e06bf83f34520f1c39cb26ddce9a68467bcd
Author: M Junaid Shaukat <[email protected]>
AuthorDate: Mon Aug 10 12:18:37 2026 +0500
[GSoC 2026] Kafka Streams runner: Python wrapper that starts its own 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.
---
runners/kafka-streams/build.gradle | 6 +-
runners/kafka-streams/job-server/build.gradle | 88 +++++++++++++++
.../python/apache_beam/options/pipeline_options.py | 20 ++++
.../kafka_streams_java_job_server_test.py | 123 +++++++++++++++++++++
.../runners/portability/kafka_streams_runner.py | 106 ++++++++++++++++++
settings.gradle.kts | 1 +
6 files changed, 343 insertions(+), 1 deletion(-)
diff --git a/runners/kafka-streams/build.gradle
b/runners/kafka-streams/build.gradle
index 203168d7ec4..39955053409 100644
--- a/runners/kafka-streams/build.gradle
+++ b/runners/kafka-streams/build.gradle
@@ -21,7 +21,11 @@ import java.time.Duration
plugins { id 'org.apache.beam.module' }
-def kafka_version = '3.9.0'
+// An extension property rather than a local, so the job server module can pin
the same version
+// instead of repeating it. Both have to pin: applyJavaNature forces every
version in library.java,
+// which includes an older kafka-clients, and a module that does not override
it links against that
+// one at runtime.
+ext.kafka_version = '3.9.0'
applyJavaNature(
automaticModuleName: 'org.apache.beam.runners.kafka.streams',
diff --git a/runners/kafka-streams/job-server/build.gradle
b/runners/kafka-streams/job-server/build.gradle
new file mode 100644
index 00000000000..aef2f271b7b
--- /dev/null
+++ b/runners/kafka-streams/job-server/build.gradle
@@ -0,0 +1,88 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * License); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an AS IS BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+/**
+ * Kafka Streams Runner JobServer build file.
+ *
+ * Packages the runner and everything it needs into one jar, so a pipeline
from an SDK other than
+ * Java can start a job server without a Beam source tree. The Python
KafkaStreamsRunner builds and
+ * launches this jar for the user.
+ */
+
+apply plugin: 'org.apache.beam.module'
+apply plugin: 'application'
+// Must be set before the shadow plugin is applied.
+mainClassName =
"org.apache.beam.runners.kafka.streams.KafkaStreamsJobServerDriver"
+
+applyJavaNature(
+ automaticModuleName: 'org.apache.beam.runners.kafka.streams.jobserver',
+ validateShadowJar: false,
+ exportJavadoc: false,
+ shadowClosure: {
+ // Kafka's clients and Streams libraries ship reference.conf-style
resources that have to be
+ // concatenated rather than overwritten when everything lands in one jar.
+ append "reference.conf"
+ },
+)
+
+def kafkaStreamsRunnerProject = ":runners:kafka-streams"
+
+description = "Apache Beam :: Runners :: Kafka Streams :: Job Server"
+
+evaluationDependsOn(kafkaStreamsRunnerProject)
+
+// The runner is compiled against this version, so the jar has to carry it.
Without this the
+// versions forced by applyJavaNature win, the shaded jar ships an older
kafka-clients, and every
+// pipeline fails at translation with a NoSuchMethodError rather than at build
time.
+def kafka_version = project(kafkaStreamsRunnerProject).kafka_version
+
+configurations.configureEach {
+ resolutionStrategy.eachDependency { details ->
+ if (details.requested.group == "org.apache.kafka") {
+ details.useVersion(kafka_version)
+ details.because("Kafka Streams runner is developed against Kafka
${kafka_version}.")
+ }
+ }
+}
+
+dependencies {
+ implementation project(kafkaStreamsRunnerProject)
+ permitUnusedDeclared project(kafkaStreamsRunnerProject)
+ // A binding, or the job server starts but logs nothing at all, which is
unhelpful for something
+ // a user runs in the foreground and reads to see what their pipeline is
doing.
+ runtimeOnly library.java.slf4j_simple
+ runtimeOnly project(":sdks:java:extensions:google-cloud-platform-core")
+}
+
+// The runner's classes only exist in the shadow jar, so the job server has to
be started through
+// runShadow rather than the plain run task.
+runShadow {
+ args = []
+ if (project.hasProperty('jobHost'))
+ args += ["--job-host=${project.property('jobHost')}"]
+ if (project.hasProperty('jobPort'))
+ args += ["--job-port=${project.property('jobPort')}"]
+ if (project.hasProperty('artifactPort'))
+ args += ["--artifact-port=${project.property('artifactPort')}"]
+ if (project.hasProperty('expansionPort'))
+ args += ["--expansion-port=${project.property('expansionPort')}"]
+ if (project.hasProperty('artifactsDir'))
+ args += ["--artifacts-dir=${project.property('artifactsDir')}"]
+ if (project.hasProperty('cleanArtifactsPerJob'))
+ args +=
["--clean-artifacts-per-job=${project.property('cleanArtifactsPerJob')}"]
+}
diff --git a/sdks/python/apache_beam/options/pipeline_options.py
b/sdks/python/apache_beam/options/pipeline_options.py
index 2533083f7e7..b0d0bdfbaa6 100644
--- a/sdks/python/apache_beam/options/pipeline_options.py
+++ b/sdks/python/apache_beam/options/pipeline_options.py
@@ -732,6 +732,7 @@ class StandardOptions(PipelineOptions):
'apache_beam.runners.interactive.interactive_runner.InteractiveRunner',
'apache_beam.runners.portability.flink_runner.FlinkRunner',
'apache_beam.runners.portability.fn_api_runner.FnApiRunner',
+
'apache_beam.runners.portability.kafka_streams_runner.KafkaStreamsRunner',
'apache_beam.runners.portability.portable_runner.PortableRunner',
'apache_beam.runners.portability.prism_runner.PrismRunner',
'apache_beam.runners.portability.spark_runner.SparkRunner',
@@ -2064,6 +2065,25 @@ class FlinkRunnerOptions(PipelineOptions):
' and the number of key groups used for partitioned state.')
+class KafkaStreamsRunnerOptions(PipelineOptions):
+ @classmethod
+ def _add_argparse_args(cls, parser):
+ parser.add_argument(
+ '--bootstrap_servers',
+ default='localhost:9092',
+ help='Comma-separated list of host:port Kafka brokers the pipeline '
+ 'connects to.')
+ parser.add_argument(
+ '--application_id',
+ help='Kafka Streams application.id for the pipeline. Must be unique '
+ 'per pipeline, since it identifies the consumer group and the '
+ 'runner\'s internal topics.')
+ parser.add_argument(
+ '--kafka_streams_job_server_jar',
+ help='Path or URL to a Beam Kafka Streams job server jar. If unset, '
+ 'the jar is built from the Beam source tree.')
+
+
class SparkRunnerOptions(PipelineOptions):
@classmethod
def _add_argparse_args(cls, parser):
diff --git
a/sdks/python/apache_beam/runners/portability/kafka_streams_java_job_server_test.py
b/sdks/python/apache_beam/runners/portability/kafka_streams_java_job_server_test.py
new file mode 100644
index 00000000000..fa785781ad4
--- /dev/null
+++
b/sdks/python/apache_beam/runners/portability/kafka_streams_java_job_server_test.py
@@ -0,0 +1,123 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+# pytype: skip-file
+
+import logging
+import tempfile
+import unittest
+
+import mock
+
+from apache_beam.options import pipeline_options
+from apache_beam.runners.portability.kafka_streams_runner import
KafkaStreamsJarJobServer
+from apache_beam.runners.portability.kafka_streams_runner import
KafkaStreamsRunner
+
+
+class KafkaStreamsTestPipelineOptions(pipeline_options.PipelineOptions):
+ def view_as(self, cls):
+ # Ensure only KafkaStreamsRunnerOptions and JobServerOptions are used when
+ # calling default_job_server. If other options classes are needed, the
+ # cache key must include them to prevent incorrect hits.
+ assert (
+ cls is pipeline_options.KafkaStreamsRunnerOptions or
+ cls is pipeline_options.JobServerOptions)
+ return super().view_as(cls)
+
+
+class KafkaStreamsJavaJobServerTest(unittest.TestCase):
+ def test_job_server_cache(self):
+ # Multiple KafkaStreamsRunner instances may be created, so job servers have
+ # to be cached across runner instances: each one is an external Java
+ # process, and starting a second for the same configuration would fail to
+ # bind the same ports.
+
+ # Options that do not affect job server configuration, such as
+ # sdk_worker_parallelism, should still hit the same cache entry.
+ job_server1 = KafkaStreamsRunner().default_job_server(
+ KafkaStreamsTestPipelineOptions(['--sdk_worker_parallelism=1']))
+ job_server2 = KafkaStreamsRunner().default_job_server(
+ KafkaStreamsTestPipelineOptions(['--sdk_worker_parallelism=2']))
+ self.assertIs(job_server2, job_server1)
+
+ # JobServerOptions do affect it, so a different port is a different server.
+ job_server3 = KafkaStreamsRunner().default_job_server(
+ KafkaStreamsTestPipelineOptions(['--job_port=1234']))
+ self.assertIsNot(job_server3, job_server1)
+
+ # So do the runner's own options.
+ job_server4 = KafkaStreamsRunner().default_job_server(
+ KafkaStreamsTestPipelineOptions(['--bootstrap_servers=other:9092']))
+ self.assertIsNot(job_server4, job_server1)
+ self.assertIsNot(job_server4, job_server3)
+
+ job_server5 = KafkaStreamsRunner().default_job_server(
+ KafkaStreamsTestPipelineOptions(['--application_id=other-pipeline']))
+ self.assertIsNot(job_server5, job_server1)
+ self.assertIsNot(job_server5, job_server4)
+
+ def test_java_arguments(self):
+ # These are what the job server driver is launched with, so they have to be
+ # options it accepts.
+ job_server = KafkaStreamsJarJobServer(
+ pipeline_options.PipelineOptions(['--application_id=test-pipeline']))
+ self.assertEqual([
+ '--artifacts-dir',
+ '/tmp/artifacts',
+ '--job-port',
+ 8099,
+ '--artifact-port',
+ 8098,
+ '--expansion-port',
+ 8097
+ ],
+ job_server.java_arguments(
+ 8099, 8098, 8097, '/tmp/artifacts'))
+
+ def test_path_to_jar_defaults_to_the_job_server_module(self):
+ job_server = KafkaStreamsJarJobServer(pipeline_options.PipelineOptions([]))
+ # Without an explicit jar the runner resolves the one built by the job
+ # server module, which is what lets a user run a pipeline without having
+ # built or started anything first. Resolving it for real would either
+ # download or demand a built jar, so only the target is checked here.
+ with mock.patch.object(job_server, 'path_to_beam_jar') as path_to_beam_jar:
+ job_server.path_to_jar()
+ path_to_beam_jar.assert_called_once_with(
+ ':runners:kafka-streams:job-server:shadowJar')
+
+ def test_path_to_jar_uses_an_explicit_jar(self):
+ with tempfile.NamedTemporaryFile(suffix='.jar') as jar:
+ job_server = KafkaStreamsJarJobServer(
+ pipeline_options.PipelineOptions(
+ ['--kafka_streams_job_server_jar=%s' % jar.name]))
+ self.assertEqual(jar.name, job_server.path_to_jar())
+
+ def test_path_to_jar_rejects_an_unusable_path(self):
+ job_server = KafkaStreamsJarJobServer(
+ pipeline_options.PipelineOptions(
+ ['--kafka_streams_job_server_jar=/no/such/jar.jar']))
+ # A path that is neither an existing file nor a URL cannot be recovered
+ # from, so it fails with the command that would produce a jar rather than
+ # letting the job server fail to start later.
+ with self.assertRaises(ValueError) as context:
+ job_server.path_to_jar()
+ self.assertIn('job-server:shadowJar', str(context.exception))
+
+
+if __name__ == '__main__':
+ logging.getLogger().setLevel(logging.INFO)
+ unittest.main()
diff --git
a/sdks/python/apache_beam/runners/portability/kafka_streams_runner.py
b/sdks/python/apache_beam/runners/portability/kafka_streams_runner.py
new file mode 100644
index 00000000000..a2d043ca174
--- /dev/null
+++ b/sdks/python/apache_beam/runners/portability/kafka_streams_runner.py
@@ -0,0 +1,106 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+"""A runner for executing portable pipelines on Kafka Streams."""
+
+# pytype: skip-file
+
+import os
+import urllib
+
+from apache_beam.options import pipeline_options
+from apache_beam.runners.portability import job_server
+from apache_beam.runners.portability import portable_runner
+
+# A Java job server is a heavyweight external process, so reuse one across
+# pipelines configured the same way.
+JOB_SERVER_CACHE = {}
+
+
+class KafkaStreamsRunner(portable_runner.PortableRunner):
+ """A runner for executing pipelines on Kafka Streams.
+
+ Starts a job server automatically, so a pipeline can be submitted without
+ running one by hand:
+
+ python my_pipeline.py \\
+ --runner=KafkaStreamsRunner \\
+ --bootstrap_servers=localhost:9092 \\
+ --application_id=my-pipeline
+
+ Pass --job_endpoint instead to submit to a job server that is already
+ running.
+ """
+
+ # Inherits run_portable_pipeline from PortableRunner.
+
+ def default_environment(self, options):
+ portable_options = options.view_as(pipeline_options.PortableOptions)
+ if (not portable_options.environment_type and
+ not portable_options.output_executable_path):
+ # The job server runs on this machine, so the SDK harness can too, which
+ # saves the user from needing Docker for a local run.
+ portable_options.environment_type = 'LOOPBACK'
+ return super().default_environment(options)
+
+ def default_job_server(self, options):
+ # Only these two option groups affect how the job server is configured, so
+ # they are what the cache is keyed on.
+ kafka_streams_options = options.view_as(
+ pipeline_options.KafkaStreamsRunnerOptions)
+ job_server_options = options.view_as(pipeline_options.JobServerOptions)
+ options_str = str(kafka_streams_options) + str(job_server_options)
+ if options_str not in JOB_SERVER_CACHE:
+ JOB_SERVER_CACHE[options_str] = job_server.StopOnExitJobServer(
+ KafkaStreamsJarJobServer(options))
+ return JOB_SERVER_CACHE[options_str]
+
+
+class KafkaStreamsJarJobServer(job_server.JavaJarJobServer):
+ def __init__(self, options):
+ super().__init__(options)
+ kafka_streams_options = options.view_as(
+ pipeline_options.KafkaStreamsRunnerOptions)
+ self._jar = kafka_streams_options.kafka_streams_job_server_jar
+
+ def path_to_jar(self):
+ if self._jar:
+ if not os.path.exists(self._jar):
+ url = urllib.parse.urlparse(self._jar)
+ if not url.scheme:
+ raise ValueError(
+ 'Unable to parse jar URL "%s". If using a full URL, make sure '
+ 'the scheme is specified. If using a local file path, make sure '
+ 'the file exists; you may have to first build the job server '
+ 'using `./gradlew runners:kafka-streams:job-server:shadowJar`.' %
+ self._jar)
+ return self._jar
+ return self.path_to_beam_jar(
+ ':runners:kafka-streams:job-server:shadowJar')
+
+ def java_arguments(
+ self, job_port, artifact_port, expansion_port, artifacts_dir):
+ return [
+ '--artifacts-dir',
+ artifacts_dir,
+ '--job-port',
+ job_port,
+ '--artifact-port',
+ artifact_port,
+ '--expansion-port',
+ expansion_port
+ ]
diff --git a/settings.gradle.kts b/settings.gradle.kts
index 9a9cb1dd1ac..cd1134d685c 100644
--- a/settings.gradle.kts
+++ b/settings.gradle.kts
@@ -145,6 +145,7 @@ include(":runners:java-job-service")
include(":runners:jet")
include(":runners:kafka-streams")
include(":runners:kafka-streams:proto")
+include(":runners:kafka-streams:job-server")
include(":runners:local-java")
include(":runners:portability:java")
include(":runners:prism")
