[GSoC 2026] Kafka Streams runner: portable ValidatesRunner suite for Python

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

@@ -18,6 +18,8 @@
 package org.apache.beam.runners.kafka.streams;
 
 import java.util.Properties;
+import java.util.UUID;
+import java.util.concurrent.atomic.AtomicReference;
 import org.apache.beam.model.pipeline.v1.RunnerApi;
 import org.apache.beam.runners.fnexecution.provisioning.JobInfo;
 import org.apache.beam.runners.jobsubmission.PortablePipelineResult;
@@ -27,6 +29,7 @@
 import org.apache.kafka.streams.KafkaStreams;
 import org.apache.kafka.streams.StreamsConfig;
 import org.apache.kafka.streams.Topology;
+import org.apache.kafka.streams.errors.StreamsUncaughtExceptionHandler;
 import org.checkerframework.checker.nullness.qual.Nullable;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -77,6 +80,19 @@ public PortablePipelineResult run(RunnerApi.Pipeline pipeline, JobInfo jobInfo)
         topology.describe());
 
     KafkaStreams kafkaStreams = new KafkaStreams(topology, streamsConfig(jobInfo));
+    // Kafka Streams reports a failed task by moving the client to ERROR and keeping the exception
+    // to itself, which left a failed job with nothing to say beyond "unknown error". Hold on to the
+    // first failure so this method can rethrow it: the job service turns what run() throws into the
+    // job's error message.
+    AtomicReference<@Nullable Throwable> failure = new AtomicReference<>();
+    kafkaStreams.setUncaughtExceptionHandler(
+        throwable -> {
+          failure.compareAndSet(null, throwable);
+          LOG.error("Pipeline {} failed", jobInfo.jobId(), throwable);
+          // The pipeline is a job with an owner waiting on it, not a service to keep alive, so a
+          // failure stops the client rather than replacing the thread and carrying on.
+          return StreamsUncaughtExceptionHandler.StreamThreadExceptionResponse.SHUTDOWN_CLIENT;
+        });
     // Build the result before starting: it registers a state listener, and Kafka Streams only
     // accepts one while the application is still in the CREATED state.
     KafkaStreamsPortablePipelineResult result =
@@ -110,6 +126,12 @@ public PortablePipelineResult run(RunnerApi.Pipeline pipeline, JobInfo jobInfo)
       // stream threads and the joins it does would throw straight back out of an interrupted one.
       closeInBackground(kafkaStreams, jobInfo.jobId(), "the job was cancelled");
     }
+    Throwable thrown = failure.get();
+    if (thrown != null) {
+      // Thrown rather than returned as a failed result: the job service reads the state of what is
+      // returned, but only what is thrown carries a reason the user can act on.
+      throw new RuntimeException("Pipeline " + jobInfo.jobId() + " failed", thrown);
+    }
     return result;
   }
 
@@ -146,7 +168,15 @@ private Properties streamsConfig(JobInfo jobInfo) {
     props.put(StreamsConfig.APPLICATION_ID_CONFIG, pipelineOptions.getApplicationId());
     props.put(StreamsConfig.STATE_DIR_CONFIG, pipelineOptions.getStateDir());
     props.put(StreamsConfig.PROCESSING_GUARANTEE_CONFIG, StreamsConfig.EXACTLY_ONCE_V2);
-    props.put(StreamsConfig.CLIENT_ID_CONFIG, jobInfo.jobId());
+    // The job id identifies the pipeline, which every instance of it shares, so on its own it does
+    // not identify an instance. Kafka Streams names threads, consumers and metrics after the client
+    // id, so two workers running the same job would produce logs and JMX metrics that cannot be
+    // told
+    // apart — in a deployment whose whole point is that you add workers. Keeping the job id as the
+    // prefix leaves the pipeline recognizable; the suffix is what makes each worker distinct, and
+    // is
+    // what Kafka Streams does by default when no client id is set.
+    props.put(StreamsConfig.CLIENT_ID_CONFIG, jobInfo.jobId() + "-" + UUID.randomUUID());
     return props;
   }
 }

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

@@ -26,8 +26,6 @@
 import org.apache.beam.sdk.metrics.MetricResults;
 import org.apache.kafka.streams.KafkaStreams;
 import org.joda.time.Duration;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
 
 /**
  * Result of executing a portable pipeline as a {@link KafkaStreams} application.
@@ -38,9 +36,6 @@
  */
 class KafkaStreamsPortablePipelineResult implements PortablePipelineResult {
 
-  private static final Logger LOG =
-      LoggerFactory.getLogger(KafkaStreamsPortablePipelineResult.class);
-
   private final KafkaStreams kafkaStreams;
   // The job's metrics accumulator, shared by reference with the topology's stage processors, which
   // update it as the SDK harness reports bundle metrics.
@@ -126,8 +121,16 @@ public MetricResults metrics() {
 
   @Override
   public JobApi.MetricResults portableMetrics() throws UnsupportedOperationException {
-    LOG.debug("portableMetrics() not yet implemented in the Kafka Streams runner");
-    return JobApi.MetricResults.newBuilder().build();
+    // How a pipeline from another SDK reads its metrics. The job service asks for these once the
+    // job is terminal and returns them over the job API. Without it a Python pipeline saw no
+    // metrics at all, even though the same values were already available to a Java one.
+    //
+    // Reported as attempted only, and deliberately not also as committed: the values are what the
+    // SDK harness reported per bundle, which is not tied to the commit of the records that produced
+    // them. Committed metrics are https://github.com/apache/beam/issues/39635.
+    return JobApi.MetricResults.newBuilder()
+        .addAllAttempted(metricsContainerStepMap.getMonitoringInfos())
+        .build();
   }
 
   private static State mapState(KafkaStreams.State state) {

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

@@ -102,10 +102,14 @@ public void translate(
     String shuffleName = transformId + SHUFFLE_SUFFIX;
     String sinkName = transformId + SINK_SUFFIX;
     String sourceName = transformId + SOURCE_SUFFIX;
-    String stateStoreName = transformId + STATE_STORE_SUFFIX;
-    String holdsIndexStoreName = transformId + HOLDS_INDEX_STORE_SUFFIX;
-    String timerStoreName = transformId + TIMER_STORE_SUFFIX;
-    String timerIndexStoreName = transformId + TIMER_INDEX_STORE_SUFFIX;
+    String stateStoreName =
+        KafkaStreamsTranslationContext.getStoreName(transformId, STATE_STORE_SUFFIX);
+    String holdsIndexStoreName =
+        KafkaStreamsTranslationContext.getStoreName(transformId, HOLDS_INDEX_STORE_SUFFIX);
+    String timerStoreName =
+        KafkaStreamsTranslationContext.getStoreName(transformId, TIMER_STORE_SUFFIX);
+    String timerIndexStoreName =
+        KafkaStreamsTranslationContext.getStoreName(transformId, TIMER_INDEX_STORE_SUFFIX);
     String repartitionTopic =
         repartitionTopic(transformId, context.getPipelineOptions().getApplicationId());
 

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

@@ -70,7 +70,8 @@ public void translate(
 
     Topology topology = context.getTopology();
     String sourceNodeName = transformId + SOURCE_SUFFIX;
-    String stateStoreName = transformId + STATE_STORE_SUFFIX;
+    String stateStoreName =
+        KafkaStreamsTranslationContext.getStoreName(transformId, STATE_STORE_SUFFIX);
     String bootstrapTopic = context.getImpulseBootstrapTopic(transformId);
 
     topology.addSource(

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

@@ -192,4 +192,21 @@ public String getReadBootstrapTopic(String transformId) {
         + "_"
         + sanitizedTransformId;
   }
+
+  /**
+   * Returns the name of a state store belonging to a transform.
+   *
+   * <p>The transform id is sanitized to Kafka's legal topic-name characters even though a store
+   * name is not itself a topic: Kafka Streams names a persistent store's changelog topic after the
+   * store, so a transform whose name contains a character a topic may not — which is ordinary,
+   * {@code CombinePerKey(MeanCombineFn)/Group} is a Beam transform name — would fail at runtime
+   * when the changelog is created.
+   *
+   * <p>Two transform ids differing only in characters that are replaced would sanitize to one name.
+   * Kafka Streams rejects a store name that is already taken when the topology is built, so that
+   * surfaces as a failure to start rather than as two transforms quietly sharing state.
+   */
+  public static String getStoreName(String transformId, String suffix) {
+    return ILLEGAL_TOPIC_CHARS.matcher(transformId).replaceAll("_") + suffix;
+  }
 }

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

@@ -126,7 +126,8 @@ private <T, CheckpointT extends UnboundedSource.CheckpointMark> void addUnbounde
 
     Topology topology = context.getTopology();
     String sourceNodeName = transformId + SOURCE_SUFFIX;
-    String stateStoreName = transformId + STATE_STORE_SUFFIX;
+    String stateStoreName =
+        KafkaStreamsTranslationContext.getStoreName(transformId, STATE_STORE_SUFFIX);
     String bootstrapTopic = context.getReadBootstrapTopic(transformId);
     SerializablePipelineOptions options =
         new SerializablePipelineOptions(context.getPipelineOptions());
@@ -185,7 +186,8 @@ private <T> void addReadNodes(
 
     Topology topology = context.getTopology();
     String sourceNodeName = transformId + SOURCE_SUFFIX;
-    String stateStoreName = transformId + STATE_STORE_SUFFIX;
+    String stateStoreName =
+        KafkaStreamsTranslationContext.getStoreName(transformId, STATE_STORE_SUFFIX);
     String bootstrapTopic = context.getReadBootstrapTopic(transformId);
     SerializablePipelineOptions options =
         new SerializablePipelineOptions(context.getPipelineOptions());

runners/kafka-streams/src/test/java/org/apache/beam/runners/kafka/streams/KafkaStreamsPortablePipelineResultTest.java

@@ -0,0 +1,96 @@
+/*
+ * 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 static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.hasItem;
+import static org.hamcrest.Matchers.hasProperty;
+import static org.hamcrest.Matchers.is;
+import static org.hamcrest.Matchers.not;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+import org.apache.beam.model.jobmanagement.v1.JobApi;
+import org.apache.beam.runners.core.metrics.MetricsContainerImpl;
+import org.apache.beam.runners.core.metrics.MetricsContainerStepMap;
+import org.apache.beam.sdk.metrics.MetricName;
+import org.apache.kafka.streams.KafkaStreams;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.JUnit4;
+
+/**
+ * Tests for {@link KafkaStreamsPortablePipelineResult}, in particular the metrics an SDK other than
+ * Java reads its results through.
+ */
+@RunWith(JUnit4.class)
+public class KafkaStreamsPortablePipelineResultTest {
+
+  private static final String STEP = "a-stage";
+  private static final String NAMESPACE = "ns";
+  private static final String COUNTER = "elements";
+
+  private static KafkaStreams idleClient() {
+    KafkaStreams kafkaStreams = mock(KafkaStreams.class);
+    // The result registers a state listener and checks the current state, so it has to have one.
+    when(kafkaStreams.state()).thenReturn(KafkaStreams.State.CREATED);
+    return kafkaStreams;
+  }
+
+  @Test
+  public void portableMetricsReportWhatTheHarnessMeasured() {
+    MetricsContainerStepMap stepMap = new MetricsContainerStepMap();
+    MetricsContainerImpl container = stepMap.getContainer(STEP);
+    container.getCounter(MetricName.named(NAMESPACE, COUNTER)).inc(7);
+
+    KafkaStreamsPortablePipelineResult result =
+        new KafkaStreamsPortablePipelineResult(idleClient(), stepMap, () -> {});
+
+    JobApi.MetricResults metrics = result.portableMetrics();
+
+    // A pipeline from another SDK reads these over the job API; before they were reported the list
+    // was empty and a Python pipeline saw no metrics at all.
+    assertThat(
+        metrics.getAttemptedList(),
+        hasItem(hasProperty("urn", is("beam:metric:user:sum_int64:v1"))));
+    assertThat(metrics.getAttemptedCount(), is(not(0)));
+  }
+
+  @Test
+  public void portableMetricsAreNotReportedAsCommitted() {
+    // The values are what the SDK harness reported per bundle, which is not tied to the commit of
+    // the records that produced them, so claiming them as committed would be wrong.
+    // See https://github.com/apache/beam/issues/39635.
+    MetricsContainerStepMap stepMap = new MetricsContainerStepMap();
+    stepMap.getContainer(STEP).getCounter(MetricName.named(NAMESPACE, COUNTER)).inc(1);
+
+    KafkaStreamsPortablePipelineResult result =
+        new KafkaStreamsPortablePipelineResult(idleClient(), stepMap, () -> {});
+
+    assertThat(result.portableMetrics().getCommittedCount(), is(0));
+  }
+
+  @Test
+  public void aPipelineThatMeasuredNothingReportsNothing() {
+    KafkaStreamsPortablePipelineResult result =
+        new KafkaStreamsPortablePipelineResult(
+            idleClient(), new MetricsContainerStepMap(), () -> {});
+
+    assertThat(result.portableMetrics().getAttemptedCount(), is(0));
+  }
+}

runners/kafka-streams/src/test/java/org/apache/beam/runners/kafka/streams/KafkaStreamsRunnerBrokerIT.java

@@ -21,7 +21,14 @@
 import static org.hamcrest.MatcherAssert.assertThat;
 
 import java.nio.file.Files;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
 import java.util.UUID;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
 import org.apache.beam.model.pipeline.v1.RunnerApi;
 import org.apache.beam.runners.fnexecution.provisioning.JobInfo;
 import org.apache.beam.sdk.Pipeline;
@@ -123,11 +130,22 @@ private KafkaStreamsPipelineOptions options() {
   }
 
   private KafkaStreamsPipelineOptions options(int topicPartitions) {
+    return options(topicPartitions, "ks-broker-it-" + UUID.randomUUID());
+  }
+
+  /**
+   * Options for one runner instance.
+   *
+   * <p>Two instances of the same job share an application id — that is what puts them in one
+   * consumer group and so splits the work between them — but each needs its own state directory,
+   * since the local stores are per instance.
+   */
+  private KafkaStreamsPipelineOptions options(int topicPartitions, String applicationId) {
     KafkaStreamsPipelineOptions options =
         PipelineOptionsFactory.create().as(KafkaStreamsPipelineOptions.class);
     options.setRunner(CrashingRunner.class);
     options.setBootstrapServers(kafka.getBootstrapServers());
-    options.setApplicationId("ks-broker-it-" + UUID.randomUUID());
+    options.setApplicationId(applicationId);
     options.setInternalParallelism(topicPartitions);
     options
         .as(PortablePipelineOptions.class)
@@ -275,6 +293,52 @@ public void aBoundedPipelineTerminatesOnItsOwn() throws Exception {
     assertThat(counterValue(result), is(1L));
   }
 
+  @Test
+  public void twoInstancesShareThreePartitions() throws Exception {
+    // Everything else here runs one instance, which leaves the thing the runner exists for
+    // untested: the work being split between instances by Kafka's own group membership.
+    //
+    // Three partitions across two instances is deliberate. It does not divide, so the instances
+    // take an unequal share, and a watermark aggregator on either of them has to hear from all
+    // three upstream partitions — some of which are being produced by the other instance — before
+    // it may let its watermark advance. If the reports were tied to the instance that produced
+    // them rather than to the partition, this is the shape that would break.
+    String applicationId = "ks-broker-it-" + UUID.randomUUID();
+    List<PipelineResult> results = Collections.synchronizedList(new ArrayList<>());
+    ExecutorService instances = Executors.newFixedThreadPool(2);
+    try {
+      List<Future<?>> running = new ArrayList<>();
+      for (int instance = 0; instance < 2; instance++) {
+        KafkaStreamsPipelineOptions options = options(3, applicationId);
+        Pipeline pipeline = Pipeline.create(options);
+        buildChainedPipeline(pipeline);
+        running.add(
+            instances.submit(
+                () -> {
+                  // run() blocks until its instance has finished, so each needs its own thread.
+                  results.add(runPipeline(pipeline, options));
+                }));
+      }
+      for (Future<?> future : running) {
+        // Fails rather than hangs if an instance never finishes — which is the interesting way for
+        // this to go wrong, since an instance only stops once every processor it owns is done.
+        future.get(TIMEOUT.getMillis(), TimeUnit.MILLISECONDS);
+      }
+    } finally {
+      instances.shutdownNow();
+    }
+
+    // The pipeline collapses everything onto one key, so exactly one group comes out of the second
+    // GroupByKey however the partitions were shared. Each instance counts what it processed, so the
+    // total across both is what has to be one: a group counted twice would mean the instances had
+    // both claimed the same partition's data.
+    long groups = 0;
+    for (PipelineResult result : results) {
+      groups += counterValue(result);
+    }
+    assertThat(groups, is(1L));
+  }
+
   /**
    * Polls the pipeline's metrics until the counter reaches {@code expected} or the timeout hits.
    */