From 3d2439507ab39d73493b6d961524d4b62501afcd Mon Sep 17 00:00:00 2001 From: junaiddshaukat Date: Fri, 7 Aug 2026 18:57:45 +0500 Subject: [PATCH 1/2] [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/job-server packages the runner into one shaded jar, so the job server can be launched without a Beam source tree, in the same shape as the Flink and Spark job server modules. It carries an SLF4J binding: the Spark module excludes one because Spark supplies its own, and copying that left the jar starting up and then logging nothing, which is unhelpful for a process a user runs in the foreground to watch their pipeline. On the Python side 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. Beam's Python is not built in my environment, so each link of the chain was checked separately rather than assumed: the jar starts all three services on the arguments the Python runner passes, those arguments are all accepted by JobServerDriver.ServerConfiguration, and against an installed Beam with these files grafted in the runner resolves, the options parse, the environment defaults to LOOPBACK and java_arguments produces those flags. A Python pipeline running end to end is the part still unverified. --- runners/kafka-streams/job-server/build.gradle | 72 ++++++++++++ .../apache_beam/options/pipeline_options.py | 20 ++++ .../portability/kafka_streams_runner.py | 106 ++++++++++++++++++ settings.gradle.kts | 1 + 4 files changed, 199 insertions(+) create mode 100644 runners/kafka-streams/job-server/build.gradle create mode 100644 sdks/python/apache_beam/runners/portability/kafka_streams_runner.py diff --git a/runners/kafka-streams/job-server/build.gradle b/runners/kafka-streams/job-server/build.gradle new file mode 100644 index 000000000000..45d50cd1a788 --- /dev/null +++ b/runners/kafka-streams/job-server/build.gradle @@ -0,0 +1,72 @@ +/* + * 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" + +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 2533083f7e7e..b0d0bdfbaa64 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 @@ def _add_argparse_args(cls, parser): ' 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_runner.py b/sdks/python/apache_beam/runners/portability/kafka_streams_runner.py new file mode 100644 index 000000000000..a2d043ca174e --- /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 9a9cb1dd1acb..cd1134d685c9 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") From 2f8fefaabd33cb8a2dc244ec2bc04b0b2d50a4f5 Mon Sep 17 00:00:00 2001 From: junaiddshaukat Date: Sat, 8 Aug 2026 16:58:35 +0500 Subject: [PATCH 2/2] [GSoC 2026] Kafka Streams runner: test the Python job server wiring Covers what the Python wrapper is responsible for: that job servers are cached across runner instances and keyed on the options that actually change their configuration, that the jar is resolved from the job server module, and that the driver is launched with arguments it accepts. Follows spark_java_job_server_test.py. --- .../kafka_streams_java_job_server_test.py | 123 ++++++++++++++++++ 1 file changed, 123 insertions(+) create mode 100644 sdks/python/apache_beam/runners/portability/kafka_streams_java_job_server_test.py 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 000000000000..fa785781ad49 --- /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()