From c1d6c5efd1453eb1efc5150634a927c17c3b51d5 Mon Sep 17 00:00:00 2001 From: junaiddshaukat Date: Sun, 19 Jul 2026 11:02:53 +0500 Subject: [PATCH 1/3] [GSoC 2026] Kafka Streams runner: validatesRunner task; Create and Flatten suites green Adds the validatesRunner gradle task running Beam's @ValidatesRunner tests from the SDK core test jar against TestKafkaStreamsRunner (category excludes for unsupported features, plus a sickbay list). CreateTest passes 5/5 and FlattenTest 13/13, with one test sickbayed pending multi-output executable stages. Enabling the suite forced three runner fixes: - Impulse bootstrap topics are now per-transform: a pipeline can contain two Impulses (an empty Create plus the dummy branch PAssert adds), and Kafka Streams rejects registering one topic on two source nodes. - Flatten is no longer in knownUrns(): declaring it native made TrivialNativeTransformExpander corrupt the fused graph of a multi-branch Flatten fed by composite expansions ("consumed but never produced" from the fuser's own validation). Mirrors the Flink runner keeping Read out of its known URNs for expander reasons. - A Flatten of zero PCollections is rewritten before fusion into a primitive Read of a new EmptyBoundedSource: a zero-input Flatten is a root of the pipeline graph and GreedyPipelineFuser only accepts Impulse or Read roots, and an empty read has exactly the empty-PCollection semantics (no elements, then the terminal watermark). --- runners/kafka-streams/build.gradle | 71 +++++++++++++++ .../translation/EmptyBoundedSource.java | 89 +++++++++++++++++++ .../translation/FlattenTranslator.java | 2 +- .../translation/ImpulseTranslator.java | 8 +- .../KafkaStreamsPipelineTranslator.java | 64 ++++++++++++- .../KafkaStreamsTranslationContext.java | 15 +++- 6 files changed, 239 insertions(+), 10 deletions(-) create mode 100644 runners/kafka-streams/src/main/java/org/apache/beam/runners/kafka/streams/translation/EmptyBoundedSource.java diff --git a/runners/kafka-streams/build.gradle b/runners/kafka-streams/build.gradle index a794299cb608..56df15240fe9 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,69 @@ 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 { + 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 000000000000..14bee0c3cfd0 --- /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 { + + @Override + public List> split( + long desiredBundleSizeBytes, PipelineOptions options) { + return Collections.singletonList(this); + } + + @Override + public long getEstimatedSizeBytes(PipelineOptions options) { + return 0; + } + + @Override + public BoundedReader createReader(PipelineOptions options) { + return new EmptyReader(this); + } + + @Override + public Coder getOutputCoder() { + return ByteArrayCoder.of(); + } + + /** A reader that is exhausted from the start. */ + private static final class EmptyReader extends BoundedReader { + 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 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 8f6ce2b546dd..3ac4c1304daa 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 @@ 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 seenInputs = new HashSet<>(); List parentProcessors = new ArrayList<>(); Set upstreamTransformIds = new HashSet<>(); @@ -71,7 +72,6 @@ public void translate( 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 a90987ba6383..79d8c3cf577f 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 @@ *

Adds three nodes to the Kafka Streams {@link Topology}: * *

    - *
  • 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 + *
  • 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}. *
  • 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, 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 a9e26ecd6412..a71434bfd8d0 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 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.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 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 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 89a2d9825cbc..ec1b3f26aded 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 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; } /** From 1245da4a64137b2bee8fd304deed477df26ef191 Mon Sep 17 00:00:00 2001 From: junaiddshaukat Date: Sun, 19 Jul 2026 11:16:20 +0500 Subject: [PATCH 2/3] Run the validatesRunner suite in the feature-branch CI workflow --- .github/workflows/beam_KafkaStreamsRunner_FeatureBranch.yml | 2 ++ 1 file changed, 2 insertions(+) diff --git a/.github/workflows/beam_KafkaStreamsRunner_FeatureBranch.yml b/.github/workflows/beam_KafkaStreamsRunner_FeatureBranch.yml index 4ec7b0778cc2..e290c9beee1a 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 From 1c18c67713ca11c8e6a40b2c0312641c315adde4 Mon Sep 17 00:00:00 2001 From: junaiddshaukat Date: Sun, 19 Jul 2026 17:23:08 +0500 Subject: [PATCH 3/3] Scope the validatesRunner task to the enabled suites The task previously ran the whole ValidatesRunner category, including suites that have not been triaged yet (windowed GroupByKey, PAssert singleton views, metric types beyond counters). Feature gaps like non-global windowing have no JUnit category to exclude, so the task now declares the enabled test classes explicitly -- CreateTest and FlattenTest -- and the list grows class by class as runner support does. --- runners/kafka-streams/build.gradle | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/runners/kafka-streams/build.gradle b/runners/kafka-streams/build.gradle index 56df15240fe9..81099560ba79 100644 --- a/runners/kafka-streams/build.gradle +++ b/runners/kafka-streams/build.gradle @@ -134,6 +134,11 @@ tasks.register("validatesRunner", Test) { 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 }