[GSoC 2026] Add Kafka Streams runner skeleton module + portable entry points

CHANGES.md

@@ -71,6 +71,7 @@
 
 ## New Features / Improvements
 
+* (Java) Added a `runners/kafka-streams` Gradle module with portable job server and runner entry points; translation fails fast with an explicit unsupported-URN message until transforms are implemented ([#38465](https://github.com/apache/beam/issues/38465)).
 * Capability introduces an indicator for aggregations and timers firing during a pipeline drain, allowing users and sinks to recognize and appropriately handle potentially incomplete or partial data ([#36884](https://github.com/apache/beam/issues/36884)).
 * Added support for setting disk provisioned IOPS and throughput in Dataflow runner via `--diskProvisionedIops` and `--diskProvisionedThroughputMibps` pipeline options (Java/Go/Python) ([#38349](https://github.com/apache/beam/issues/38349)).
 * TriggerStateMachineRunner changes from BitSetCoder to SentinelBitSetCoder to

build.gradle.kts

@@ -278,6 +278,7 @@ tasks.register("javaPreCommit") {
   dependsOn(":runners:java-fn-execution:build")
   dependsOn(":runners:java-job-service:build")
   dependsOn(":runners:jet:build")
+  dependsOn(":runners:kafka-streams:build")
   dependsOn(":runners:local-java:build")
   dependsOn(":runners:portability:java:build")
   dependsOn(":runners:prism:java:build")

runners/kafka-streams/build.gradle

@@ -0,0 +1,62 @@
+/*
+ * 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.
+ */
+
+plugins { id 'org.apache.beam.module' }
+
+def kafka_version = '3.9.0'
+
+applyJavaNature(
+    automaticModuleName: 'org.apache.beam.runners.kafka.streams',
+)
+
+description = "Apache Beam :: Runners :: Kafka Streams"
+
+evaluationDependsOn(":sdks:java:core")
+evaluationDependsOn(":runners:core-java")
+
+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 {
+  compileOnly project(":sdks:java:build-tools")
+  permitUnusedDeclared project(":sdks:java:build-tools")
+
+  implementation project(path: ":sdks:java:core", configuration: "shadow")
+  implementation project(path: ":model:pipeline", configuration: "shadow")
+  implementation project(":runners:core-java")
+  permitUnusedDeclared project(":runners:core-java")
+  implementation project(":runners:java-fn-execution")
+  implementation project(":runners:java-job-service")
+  implementation project(":runners:portability:java")
+  implementation project(path: ":sdks:java:extensions:google-cloud-platform-core")
+  implementation library.java.args4j
+  implementation library.java.joda_time
+  implementation library.java.slf4j_api
+  implementation library.java.vendored_grpc_1_69_0
+  implementation library.java.vendored_guava_32_1_2_jre
+  implementation "org.apache.kafka:kafka-clients:$kafka_version"
+  implementation "org.apache.kafka:kafka-streams:$kafka_version"
+  permitUnusedDeclared "org.apache.kafka:kafka-clients:$kafka_version"
+  permitUnusedDeclared "org.apache.kafka:kafka-streams:$kafka_version"
+}

runners/kafka-streams/src/main/java/org/apache/beam/runners/kafka/streams/KafkaStreamsJobInvoker.java

@@ -0,0 +1,99 @@
+/*
+ * 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.
+ */
+package org.apache.beam.runners.kafka.streams;
+
+import java.util.UUID;
+import org.apache.beam.model.pipeline.v1.RunnerApi;
+import org.apache.beam.runners.fnexecution.provisioning.JobInfo;
+import org.apache.beam.runners.jobsubmission.JobInvocation;
+import org.apache.beam.runners.jobsubmission.JobInvoker;
+import org.apache.beam.runners.jobsubmission.PortablePipelineRunner;
+import org.apache.beam.sdk.util.construction.PipelineOptionsTranslation;
+import org.apache.beam.vendor.grpc.v1p69p0.com.google.protobuf.Struct;
+import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.util.concurrent.ListeningExecutorService;
+import org.checkerframework.checker.nullness.qual.Nullable;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/** Job invoker for the Kafka Streams portable runner. */
+@SuppressWarnings({
+  "nullness" // TODO(https://github.com/apache/beam/issues/20497)
+})
+public class KafkaStreamsJobInvoker extends JobInvoker {
+
+  private static final Logger LOG = LoggerFactory.getLogger(KafkaStreamsJobInvoker.class);
+
+  public static KafkaStreamsJobInvoker create(
+      KafkaStreamsJobServerDriver.KafkaStreamsServerConfiguration serverConfig) {
+    return new KafkaStreamsJobInvoker(serverConfig);
+  }
+
+  private final KafkaStreamsJobServerDriver.KafkaStreamsServerConfiguration serverConfig;
+
+  protected KafkaStreamsJobInvoker(
+      KafkaStreamsJobServerDriver.KafkaStreamsServerConfiguration serverConfig) {
+    super("kafka-streams-runner-job-invoker-%d");
+    this.serverConfig = serverConfig;
+  }
+
+  @Override
+  protected JobInvocation invokeWithExecutor(
+      RunnerApi.Pipeline pipeline,
+      Struct options,
+      @Nullable String retrievalToken,
+      ListeningExecutorService executorService) {
+
+    LOG.trace(
+        "Parsing pipeline options (job server {}:{})",
+        serverConfig.getHost(),
+        serverConfig.getPort());
+    KafkaStreamsPipelineOptions kafkaStreamsOptions =
+        PipelineOptionsTranslation.fromProto(options).as(KafkaStreamsPipelineOptions.class);
+
+    String invocationId =
+        String.format("%s_%s", kafkaStreamsOptions.getJobName(), UUID.randomUUID().toString());
+
+    PortablePipelineRunner pipelineRunner = new KafkaStreamsPipelineRunner(kafkaStreamsOptions);
+    kafkaStreamsOptions.setRunner(null);
+
+    LOG.info("Invoking job {} with pipeline runner {}", invocationId, pipelineRunner);
+    return createJobInvocation(
+        invocationId,
+        retrievalToken,
+        executorService,
+        pipeline,
+        kafkaStreamsOptions,
+        pipelineRunner);
+  }
+
+  protected JobInvocation createJobInvocation(
+      String invocationId,
+      String retrievalToken,
+      ListeningExecutorService executorService,
+      RunnerApi.Pipeline pipeline,
+      KafkaStreamsPipelineOptions kafkaStreamsOptions,
+      PortablePipelineRunner pipelineRunner) {
+    JobInfo jobInfo =
+        JobInfo.create(
+            invocationId,
+            kafkaStreamsOptions.getJobName(),
+            retrievalToken,
+            PipelineOptionsTranslation.toProto(kafkaStreamsOptions));
+    return new JobInvocation(jobInfo, executorService, pipeline, pipelineRunner);
+  }
+}

runners/kafka-streams/src/main/java/org/apache/beam/runners/kafka/streams/KafkaStreamsJobServerDriver.java

@@ -0,0 +1,109 @@
+/*
+ * 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.
+ */
+package org.apache.beam.runners.kafka.streams;
+
+import org.apache.beam.runners.jobsubmission.JobServerDriver;
+import org.apache.beam.sdk.extensions.gcp.options.GcsOptions;
+import org.apache.beam.sdk.fn.server.ServerFactory;
+import org.apache.beam.sdk.io.FileSystems;
+import org.apache.beam.sdk.options.PipelineOptions;
+import org.apache.beam.sdk.options.PipelineOptionsFactory;
+import org.kohsuke.args4j.CmdLineException;
+import org.kohsuke.args4j.CmdLineParser;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/** Driver that starts a Beam job server for the Kafka Streams portable runner. */
+@SuppressWarnings({
+  "nullness" // TODO(https://github.com/apache/beam/issues/20497)
+})
+public class KafkaStreamsJobServerDriver extends JobServerDriver {
+
+  private static final Logger LOG = LoggerFactory.getLogger(KafkaStreamsJobServerDriver.class);
+
+  /** Runner-specific configuration for the job server process. */
+  public static class KafkaStreamsServerConfiguration extends ServerConfiguration {}
+
+  public static void main(String[] args) throws Exception {
+    PipelineOptions options = PipelineOptionsFactory.create();
+    options.as(GcsOptions.class).setGcsUploadBufferSizeBytes(1024 * 1024);
+    FileSystems.setDefaultPipelineOptions(options);
+    fromParams(args).run();
+  }
+
+  private static void printUsage(CmdLineParser parser) {
+    System.err.println(
+        String.format(
+            "Usage: java %s arguments...", KafkaStreamsJobServerDriver.class.getSimpleName()));
+    parser.printUsage(System.err);
+    System.err.println();
+  }
+
+  public static KafkaStreamsServerConfiguration parseArgs(String[] args) {
+    KafkaStreamsServerConfiguration configuration = new KafkaStreamsServerConfiguration();
+    CmdLineParser parser = new CmdLineParser(configuration);
+    try {
+      parser.parseArgument(args);
+    } catch (CmdLineException e) {
+      LOG.error("Unable to parse command line arguments.", e);
+      printUsage(parser);
+      throw new IllegalArgumentException("Unable to parse command line arguments.", e);
+    }
+    return configuration;
+  }
+
+  /** Used by tests and tooling to construct a driver from command-line parameters. */
+  public static KafkaStreamsJobServerDriver fromParams(String[] args) {
+    return fromConfig(parseArgs(args));
+  }
+
+  public static KafkaStreamsJobServerDriver fromConfig(
+      KafkaStreamsServerConfiguration configuration) {
+    return create(
+        configuration,
+        createJobServerFactory(configuration),
+        createArtifactServerFactory(configuration),
+        () -> KafkaStreamsJobInvoker.create(configuration));
+  }
+
+  public static KafkaStreamsJobServerDriver fromConfig(
+      KafkaStreamsServerConfiguration configuration, JobInvokerFactory jobInvokerFactory) {
+    return create(
+        configuration,
+        createJobServerFactory(configuration),
+        createArtifactServerFactory(configuration),
+        jobInvokerFactory);
+  }
+
+  private static KafkaStreamsJobServerDriver create(
+      KafkaStreamsServerConfiguration configuration,
+      ServerFactory jobServerFactory,
+      ServerFactory artifactServerFactory,
+      JobInvokerFactory jobInvokerFactory) {
+    return new KafkaStreamsJobServerDriver(
+        configuration, jobServerFactory, artifactServerFactory, jobInvokerFactory);
+  }
+
+  private KafkaStreamsJobServerDriver(
+      KafkaStreamsServerConfiguration configuration,
+      ServerFactory jobServerFactory,
+      ServerFactory artifactServerFactory,
+      JobInvokerFactory jobInvokerFactory) {
+    super(configuration, jobServerFactory, artifactServerFactory, jobInvokerFactory);
+  }
+}

runners/kafka-streams/src/main/java/org/apache/beam/runners/kafka/streams/KafkaStreamsPipelineOptions.java

@@ -0,0 +1,76 @@
+/*
+ * 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.
+ */
+package org.apache.beam.runners.kafka.streams;
+
+import java.nio.file.Paths;
+import org.apache.beam.sdk.options.Default;
+import org.apache.beam.sdk.options.DefaultValueFactory;
+import org.apache.beam.sdk.options.Description;
+import org.apache.beam.sdk.options.PipelineOptions;
+import org.apache.beam.sdk.options.PortablePipelineOptions;
+
+/** Pipeline options for the Kafka Streams runner. */
+public interface KafkaStreamsPipelineOptions extends PortablePipelineOptions {
+
+  @Description("Comma-separated list of host:port Kafka brokers used by the Kafka Streams client.")
+  @Default.String("localhost:9092")
+  String getBootstrapServers();
+
+  void setBootstrapServers(String bootstrapServers);
+
+  @Description(
+      "Kafka Streams application.id (must be unique for each distinct topology using the same "
+          + "input topics in a Kafka cluster).")
+  @Default.String("beam-kafka-streams-runner")
+  String getApplicationId();
+
+  void setApplicationId(String applicationId);
+
+  @Description(
+      "Kafka Streams processing.guarantee setting, for example at_least_once or exactly_once_v2.")
+  @Default.String("exactly_once_v2")
+  String getProcessingGuarantee();
+
+  void setProcessingGuarantee(String processingGuarantee);
+
+  @Description("Soft cap on the number of elements per bundle.")
+  @Default.Integer(1000)
+  int getMaxBundleSize();
+
+  void setMaxBundleSize(int maxBundleSize);
+
+  @Description("Soft cap on bundle wall-clock duration in milliseconds.")
+  @Default.Integer(1000)
+  int getMaxBundleTimeMs();
+
+  void setMaxBundleTimeMs(int maxBundleTimeMs);
+
+  @Description("Directory where Kafka Streams stores local state.")
+  @Default.InstanceFactory(StateDirDefaultFactory.class)
+  String getStateDir();
+
+  void setStateDir(String stateDir);
+
+  /** Default {@link #getStateDir()} under the JVM temp directory. */
+  class StateDirDefaultFactory implements DefaultValueFactory<String> {
+    @Override
+    public String create(PipelineOptions options) {
+      return Paths.get(System.getProperty("java.io.tmpdir"), "beam-kafka-streams-state").toString();
+    }
+  }
+}