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 diff --git a/runners/kafka-streams/build.gradle b/runners/kafka-streams/build.gradle index a794299cb608..81099560ba79 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 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; } /**