[GSoC 2026] Kafka Streams runner: validatesRunner task; Create and Flatten suites green

.github/workflows/beam_KafkaStreamsRunner_FeatureBranch.yml

@@ -70,3 +70,5 @@ jobs:
             ${{ runner.os }}-gradle-kafka-streams-
       - name: Build and test Kafka Streams runner
         run: ./gradlew :runners:kafka-streams:build --no-daemon --stacktrace
+      - name: Run ValidatesRunner suite
+        run: ./gradlew :runners:kafka-streams:validatesRunner --no-daemon --stacktrace

runners/kafka-streams/build.gradle

@@ -16,6 +16,8 @@
  * limitations under the License.
  */
 
+import groovy.json.JsonOutput
+
 plugins { id 'org.apache.beam.module' }
 
 def kafka_version = '3.9.0'
@@ -29,6 +31,10 @@ description = "Apache Beam :: Runners :: Kafka Streams"
 evaluationDependsOn(":sdks:java:core")
 evaluationDependsOn(":runners:core-java")
 
+configurations {
+  validatesRunner
+}
+
 configurations.configureEach {
   resolutionStrategy.eachDependency { details ->
     if (details.requested.group == "org.apache.kafka") {
@@ -68,4 +74,74 @@ dependencies {
   testImplementation library.java.junit
   testImplementation library.java.mockito_core
   testImplementation "org.apache.kafka:kafka-streams-test-utils:$kafka_version"
+
+  // Beam's @ValidatesRunner suite: the test classes come from the SDK core test jar; the runner
+  // (TestKafkaStreamsRunner) and its TopologyTestDriver harness come from this module's test
+  // output and test runtime classpath.
+  validatesRunner project(path: ":sdks:java:core", configuration: "shadowTest")
+  validatesRunner project(project.path)
+  validatesRunner sourceSets.test.output
+  validatesRunner sourceSets.test.runtimeClasspath
+}
+
+
+// Known-failing @ValidatesRunner tests, excluded until the feature they need lands.
+def sickbayTests = [
+  // Needs multi-output executable stages (output-tag dispatch in ExecutableStageProcessor).
+  'org.apache.beam.sdk.transforms.FlattenTest.testFlattenMultiplePCollectionsHavingMultipleConsumers',
+]
+
+tasks.register("validatesRunner", Test) {
+  group = "Verification"
+  description = "Runs the subset of Beam's ValidatesRunner suite the Kafka Streams runner supports."
+  // Never consider up-to-date; the suite is the correctness gate.
+  outputs.upToDateWhen { false }
+  systemProperty "beamTestPipelineOptions",
+      JsonOutput.toJson(["--runner=org.apache.beam.runners.kafka.streams.TestKafkaStreamsRunner"])
+  classpath = configurations.validatesRunner
+  testClassesDirs = files(project(":sdks:java:core").sourceSets.test.output.classesDirs)
+  maxParallelForks 2
+  useJUnit {
+    includeCategories 'org.apache.beam.sdk.testing.ValidatesRunner'
+    // Environment / harness features that need a properly configured external environment.
+    excludeCategories 'org.apache.beam.sdk.testing.UsesExternalService'
+    excludeCategories 'org.apache.beam.sdk.testing.UsesSdkHarnessEnvironment'
+    excludeCategories 'org.apache.beam.sdk.testing.UsesBundleFinalizer'
+    excludeCategories 'org.apache.beam.sdk.testing.UsesJavaExpansionService'
+    excludeCategories 'org.apache.beam.sdk.testing.UsesPythonExpansionService'
+    // Features the runner does not support yet.
+    excludeCategories 'org.apache.beam.sdk.testing.UsesSideInputs'
+    excludeCategories 'org.apache.beam.sdk.testing.UsesStatefulParDo'
+    excludeCategories 'org.apache.beam.sdk.testing.UsesTimersInParDo'
+    excludeCategories 'org.apache.beam.sdk.testing.UsesTimerMap'
+    excludeCategories 'org.apache.beam.sdk.testing.UsesLoopingTimer'
+    excludeCategories 'org.apache.beam.sdk.testing.UsesStrictTimerOrdering'
+    excludeCategories 'org.apache.beam.sdk.testing.UsesProcessingTimeTimers'
+    excludeCategories 'org.apache.beam.sdk.testing.UsesOnWindowExpiration'
+    excludeCategories 'org.apache.beam.sdk.testing.UsesTestStream'
+    excludeCategories 'org.apache.beam.sdk.testing.UsesUnboundedPCollections'
+    excludeCategories 'org.apache.beam.sdk.testing.UsesUnboundedSplittableParDo'
+    excludeCategories 'org.apache.beam.sdk.testing.UsesBoundedSplittableParDo'
+    excludeCategories 'org.apache.beam.sdk.testing.UsesCustomWindowMerging'
+    excludeCategories 'org.apache.beam.sdk.testing.UsesMetricsPusher'
+    excludeCategories 'org.apache.beam.sdk.testing.UsesCommittedMetrics'
+    excludeCategories 'org.apache.beam.sdk.testing.UsesSystemMetrics'
+    excludeCategories 'org.apache.beam.sdk.testing.UsesOrderedListState'
+    excludeCategories 'org.apache.beam.sdk.testing.UsesMultimapState'
+    excludeCategories 'org.apache.beam.sdk.testing.UsesMapState'
+    excludeCategories 'org.apache.beam.sdk.testing.UsesSetState'
+    excludeCategories 'org.apache.beam.sdk.testing.FlattenWithHeterogeneousCoders'
+    excludeCategories 'org.apache.beam.sdk.testing.LargeKeys$Above100MB'
+  }
+  filter {
+    // The suites enabled so far, extended class by class as runner support grows. An explicit
+    // include list is needed because feature gaps like windowing beyond the global window have no
+    // JUnit category to exclude.
+    includeTestsMatching 'org.apache.beam.sdk.transforms.CreateTest'
+    includeTestsMatching 'org.apache.beam.sdk.transforms.FlattenTest'
+    for (String test : sickbayTests) {
+      excludeTestsMatching test
+    }
+    failOnNoMatchingTests = false
+  }
 }

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

@@ -0,0 +1,89 @@
+/*
+ * 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.translation;
+
+import java.io.IOException;
+import java.util.Collections;
+import java.util.List;
+import java.util.NoSuchElementException;
+import org.apache.beam.sdk.coders.ByteArrayCoder;
+import org.apache.beam.sdk.coders.Coder;
+import org.apache.beam.sdk.io.BoundedSource;
+import org.apache.beam.sdk.options.PipelineOptions;
+
+/**
+ * A {@link BoundedSource} with no elements. The runner substitutes it for a {@code Flatten} of zero
+ * PCollections (see {@link KafkaStreamsPipelineTranslator}): reading it produces no data and a
+ * terminal watermark — exactly the semantics of an empty PCollection. The element type is never
+ * observed since no element is ever produced.
+ */
+class EmptyBoundedSource extends BoundedSource<byte[]> {
+
+  @Override
+  public List<? extends BoundedSource<byte[]>> split(
+      long desiredBundleSizeBytes, PipelineOptions options) {
+    return Collections.singletonList(this);
+  }
+
+  @Override
+  public long getEstimatedSizeBytes(PipelineOptions options) {
+    return 0;
+  }
+
+  @Override
+  public BoundedReader<byte[]> createReader(PipelineOptions options) {
+    return new EmptyReader(this);
+  }
+
+  @Override
+  public Coder<byte[]> getOutputCoder() {
+    return ByteArrayCoder.of();
+  }
+
+  /** A reader that is exhausted from the start. */
+  private static final class EmptyReader extends BoundedReader<byte[]> {
+    private final EmptyBoundedSource source;
+
+    EmptyReader(EmptyBoundedSource source) {
+      this.source = source;
+    }
+
+    @Override
+    public boolean start() {
+      return false;
+    }
+
+    @Override
+    public boolean advance() {
+      return false;
+    }
+
+    @Override
+    public byte[] getCurrent() throws NoSuchElementException {
+      throw new NoSuchElementException("EmptyBoundedSource has no elements");
+    }
+
+    @Override
+    public void close() throws IOException {}
+
+    @Override
+    public BoundedSource<byte[]> getCurrentSource() {
+      return source;
+    }
+  }
+}

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

@@ -53,6 +53,7 @@ public void translate(
     // Flatten produces exactly one output PCollection, fed by all of its input PCollections.
     String outputPCollectionId = Iterables.getOnlyElement(transform.getOutputsMap().values());
 
+    Topology topology = context.getTopology();
     Set<String> seenInputs = new HashSet<>();
     List<String> parentProcessors = new ArrayList<>();
     Set<String> upstreamTransformIds = new HashSet<>();
@@ -71,7 +72,6 @@ public void translate(
       upstreamTransformIds.add(parentProcessor);
     }
 
-    Topology topology = context.getTopology();
     topology.addProcessor(
         transformId,
         () -> new FlattenProcessor(transformId, upstreamTransformIds),

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

@@ -30,9 +30,9 @@
  * <p>Adds three nodes to the Kafka Streams {@link Topology}:
  *
  * <ul>
- *   <li>A {@code byte[]} source bound to a dedicated per-application bootstrap topic (see {@link
- *       KafkaStreamsTranslationContext#getImpulseBootstrapTopic()}). Kafka Streams refuses to start
- *       a topology that has no real source topic, so the bootstrap topic exists purely to satisfy
+ *   <li>A {@code byte[]} source bound to a dedicated per-transform bootstrap topic (see {@link
+ *       KafkaStreamsTranslationContext#getImpulseBootstrapTopic}). Kafka Streams refuses to start a
+ *       topology that has no real source topic, so the bootstrap topic exists purely to satisfy
  *       that requirement — records published to it are ignored by {@link ImpulseProcessor}.
  *   <li>The {@link ImpulseProcessor} itself, which schedules a one-shot wall-clock punctuator on
  *       {@code init} and emits a single empty data {@link KStreamsPayload} followed by a terminal
@@ -71,7 +71,7 @@ public void translate(
     Topology topology = context.getTopology();
     String sourceNodeName = transformId + SOURCE_SUFFIX;
     String stateStoreName = transformId + STATE_STORE_SUFFIX;
-    String bootstrapTopic = context.getImpulseBootstrapTopic();
+    String bootstrapTopic = context.getImpulseBootstrapTopic(transformId);
 
     topology.addSource(
         sourceNodeName,

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

@@ -23,15 +23,18 @@
 import org.apache.beam.model.pipeline.v1.RunnerApi;
 import org.apache.beam.runners.fnexecution.provisioning.JobInfo;
 import org.apache.beam.runners.kafka.streams.KafkaStreamsPipelineOptions;
+import org.apache.beam.sdk.util.SerializableUtils;
 import org.apache.beam.sdk.util.construction.NativeTransforms;
 import org.apache.beam.sdk.util.construction.PTransformTranslation;
 import org.apache.beam.sdk.util.construction.graph.ExecutableStage;
 import org.apache.beam.sdk.util.construction.graph.GreedyPipelineFuser;
 import org.apache.beam.sdk.util.construction.graph.PipelineNode;
 import org.apache.beam.sdk.util.construction.graph.QueryablePipeline;
 import org.apache.beam.sdk.util.construction.graph.TrivialNativeTransformExpander;
+import org.apache.beam.vendor.grpc.v1p69p0.com.google.protobuf.ByteString;
 import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.ImmutableMap;
 import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.ImmutableSet;
+import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.Sets;
 
 /**
  * Translates a portable Beam pipeline into a Kafka Streams {@link
@@ -75,7 +78,14 @@ public KafkaStreamsTranslationContext createTranslationContext(
    * yet (e.g. GroupByKey).
    */
   public Set<String> knownUrns() {
-    return urnToTranslator.keySet();
+    // Flatten is deliberately not exposed: the fuser has its own handling for Flatten (unzipping
+    // it into producer stages and re-introducing a runner Flatten via the OutputDeduplicator), and
+    // declaring it native makes TrivialNativeTransformExpander interact badly with multi-branch
+    // flattens whose inputs come from composite expansions (their producing stages get dropped
+    // from the fused pipeline, failing "consumed but never produced" validation). Mirrors the
+    // Flink runner, which likewise keeps Read out of its known URNs for expander reasons.
+    return Sets.difference(
+        urnToTranslator.keySet(), ImmutableSet.of(PTransformTranslation.FLATTEN_TRANSFORM_URN));
   }
 
   /**
@@ -98,10 +108,60 @@ public RunnerApi.Pipeline prepareForTranslation(RunnerApi.Pipeline pipeline) {
     if (alreadyFused) {
       return pipeline;
     }
-    RunnerApi.Pipeline trimmed = TrivialNativeTransformExpander.forKnownUrns(pipeline, knownUrns());
+    RunnerApi.Pipeline withoutEmptyFlattens = replaceEmptyFlattensWithEmptyReads(pipeline);
+    RunnerApi.Pipeline trimmed =
+        TrivialNativeTransformExpander.forKnownUrns(withoutEmptyFlattens, knownUrns());
     return GreedyPipelineFuser.fuse(trimmed).toPipeline();
   }
 
+  /**
+   * Rewrites every {@code Flatten} of zero PCollections into a primitive Read of an {@link
+   * EmptyBoundedSource}. A zero-input Flatten produces an empty PCollection, but it is also a root
+   * of the pipeline graph, and {@link GreedyPipelineFuser} only accepts Impulse or Read roots. The
+   * Read of an empty source has exactly the right semantics — no elements, then the terminal
+   * watermark — and reuses the existing {@link ReadTranslator} path.
+   */
+  private static RunnerApi.Pipeline replaceEmptyFlattensWithEmptyReads(
+      RunnerApi.Pipeline pipeline) {
+    RunnerApi.Pipeline.Builder pipelineBuilder = null;
+    for (Map.Entry<String, RunnerApi.PTransform> entry :
+        pipeline.getComponents().getTransformsMap().entrySet()) {
+      RunnerApi.PTransform transform = entry.getValue();
+      if (!PTransformTranslation.FLATTEN_TRANSFORM_URN.equals(transform.getSpec().getUrn())
+          || !transform.getInputsMap().isEmpty()) {
+        continue;
+      }
+      if (pipelineBuilder == null) {
+        pipelineBuilder = pipeline.toBuilder();
+      }
+      RunnerApi.ReadPayload emptyReadPayload =
+          RunnerApi.ReadPayload.newBuilder()
+              .setIsBounded(RunnerApi.IsBounded.Enum.BOUNDED)
+              .setSource(
+                  RunnerApi.FunctionSpec.newBuilder()
+                      .setUrn(JAVA_SERIALIZED_BOUNDED_SOURCE_URN)
+                      .setPayload(
+                          ByteString.copyFrom(
+                              SerializableUtils.serializeToByteArray(new EmptyBoundedSource()))))
+              .build();
+      pipelineBuilder
+          .getComponentsBuilder()
+          .putTransforms(
+              entry.getKey(),
+              transform
+                  .toBuilder()
+                  .setSpec(
+                      RunnerApi.FunctionSpec.newBuilder()
+                          .setUrn(PTransformTranslation.READ_TRANSFORM_URN)
+                          .setPayload(emptyReadPayload.toByteString()))
+                  .build());
+    }
+    return pipelineBuilder == null ? pipeline : pipelineBuilder.build();
+  }
+
+  /** The URN {@code ReadTranslation} uses for a Java-serialized {@code BoundedSource} payload. */
+  private static final String JAVA_SERIALIZED_BOUNDED_SOURCE_URN = "beam:java:boundedsource:v1";
+
   /**
    * Walks the pipeline in topological order and translates each transform whose URN is supported.
    * Throws {@link UnsupportedOperationException} on the first unsupported URN.

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

@@ -122,9 +122,18 @@ public String getProcessorNameForPCollection(String pCollectionId) {
     return name;
   }
 
-  /** Returns the dedicated bootstrap topic name used by Impulse for this application. */
-  public String getImpulseBootstrapTopic() {
-    return IMPULSE_BOOTSTRAP_TOPIC_PREFIX + pipelineOptions.getApplicationId();
+  /**
+   * Returns the dedicated bootstrap topic name for one Impulse transform. Keyed by transform id
+   * (sanitized to Kafka's legal topic-name character set) because a pipeline can contain several
+   * Impulses (e.g. an empty {@code Create} plus the dummy branch {@code PAssert} adds), and Kafka
+   * Streams rejects registering the same topic on two source nodes.
+   */
+  public String getImpulseBootstrapTopic(String transformId) {
+    String sanitizedTransformId = ILLEGAL_TOPIC_CHARS.matcher(transformId).replaceAll("_");
+    return IMPULSE_BOOTSTRAP_TOPIC_PREFIX
+        + pipelineOptions.getApplicationId()
+        + "_"
+        + sanitizedTransformId;
   }
 
   /**