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 d9cea589e5b [GSoC 2026] Kafka Streams runner: validatesRunner task;
Create and Flatten suites green (#39380)
d9cea589e5b is described below
commit d9cea589e5b9e2da28730620ca1ffb045607fdbc
Author: M Junaid Shaukat <[email protected]>
AuthorDate: Mon Jul 20 13:12:26 2026 +0500
[GSoC 2026] Kafka Streams runner: validatesRunner task; Create and Flatten
suites green (#39380)
---
.../beam_KafkaStreamsRunner_FeatureBranch.yml | 2 +
runners/kafka-streams/build.gradle | 76 ++++++++++++++++++
.../streams/translation/EmptyBoundedSource.java | 89 ++++++++++++++++++++++
.../streams/translation/FlattenTranslator.java | 2 +-
.../streams/translation/ImpulseTranslator.java | 8 +-
.../KafkaStreamsPipelineTranslator.java | 64 +++++++++++++++-
.../KafkaStreamsTranslationContext.java | 15 +++-
7 files changed, 246 insertions(+), 10 deletions(-)
diff --git a/.github/workflows/beam_KafkaStreamsRunner_FeatureBranch.yml
b/.github/workflows/beam_KafkaStreamsRunner_FeatureBranch.yml
index 4ec7b0778cc..e290c9beee1 100644
--- a/.github/workflows/beam_KafkaStreamsRunner_FeatureBranch.yml
+++ b/.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
diff --git a/runners/kafka-streams/build.gradle
b/runners/kafka-streams/build.gradle
index a794299cb60..81099560ba7 100644
--- a/runners/kafka-streams/build.gradle
+++ b/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
+ }
}
diff --git
a/runners/kafka-streams/src/main/java/org/apache/beam/runners/kafka/streams/translation/EmptyBoundedSource.java
b/runners/kafka-streams/src/main/java/org/apache/beam/runners/kafka/streams/translation/EmptyBoundedSource.java
new file mode 100644
index 00000000000..14bee0c3cfd
--- /dev/null
+++
b/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;
+ }
+ }
+}
diff --git
a/runners/kafka-streams/src/main/java/org/apache/beam/runners/kafka/streams/translation/FlattenTranslator.java
b/runners/kafka-streams/src/main/java/org/apache/beam/runners/kafka/streams/translation/FlattenTranslator.java
index 8f6ce2b546d..3ac4c1304da 100644
---
a/runners/kafka-streams/src/main/java/org/apache/beam/runners/kafka/streams/translation/FlattenTranslator.java
+++
b/runners/kafka-streams/src/main/java/org/apache/beam/runners/kafka/streams/translation/FlattenTranslator.java
@@ -53,6 +53,7 @@ class FlattenTranslator implements PTransformTranslator {
// 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 @@ class FlattenTranslator implements PTransformTranslator {
upstreamTransformIds.add(parentProcessor);
}
- Topology topology = context.getTopology();
topology.addProcessor(
transformId,
() -> new FlattenProcessor(transformId, upstreamTransformIds),
diff --git
a/runners/kafka-streams/src/main/java/org/apache/beam/runners/kafka/streams/translation/ImpulseTranslator.java
b/runners/kafka-streams/src/main/java/org/apache/beam/runners/kafka/streams/translation/ImpulseTranslator.java
index a90987ba638..79d8c3cf577 100644
---
a/runners/kafka-streams/src/main/java/org/apache/beam/runners/kafka/streams/translation/ImpulseTranslator.java
+++
b/runners/kafka-streams/src/main/java/org/apache/beam/runners/kafka/streams/translation/ImpulseTranslator.java
@@ -30,9 +30,9 @@ import org.apache.kafka.streams.state.Stores;
* <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 @@ class ImpulseTranslator implements PTransformTranslator {
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,
diff --git
a/runners/kafka-streams/src/main/java/org/apache/beam/runners/kafka/streams/translation/KafkaStreamsPipelineTranslator.java
b/runners/kafka-streams/src/main/java/org/apache/beam/runners/kafka/streams/translation/KafkaStreamsPipelineTranslator.java
index a9e26ecd641..a71434bfd8d 100644
---
a/runners/kafka-streams/src/main/java/org/apache/beam/runners/kafka/streams/translation/KafkaStreamsPipelineTranslator.java
+++
b/runners/kafka-streams/src/main/java/org/apache/beam/runners/kafka/streams/translation/KafkaStreamsPipelineTranslator.java
@@ -23,6 +23,7 @@ import java.util.Set;
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;
@@ -30,8 +31,10 @@ 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 class KafkaStreamsPipelineTranslator {
* 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 class KafkaStreamsPipelineTranslator {
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.
diff --git
a/runners/kafka-streams/src/main/java/org/apache/beam/runners/kafka/streams/translation/KafkaStreamsTranslationContext.java
b/runners/kafka-streams/src/main/java/org/apache/beam/runners/kafka/streams/translation/KafkaStreamsTranslationContext.java
index 89a2d9825cb..ec1b3f26ade 100644
---
a/runners/kafka-streams/src/main/java/org/apache/beam/runners/kafka/streams/translation/KafkaStreamsTranslationContext.java
+++
b/runners/kafka-streams/src/main/java/org/apache/beam/runners/kafka/streams/translation/KafkaStreamsTranslationContext.java
@@ -122,9 +122,18 @@ public class KafkaStreamsTranslationContext {
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;
}
/**
