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 87911f00973 [GSoC 2026] Kafka Streams runner:
TestPipeline-dispatchable test runner (PAssert works) (#39362)
87911f00973 is described below
commit 87911f00973cb7ae13e78bd07ce71d9b2395e044
Author: M Junaid Shaukat <[email protected]>
AuthorDate: Fri Jul 17 12:40:09 2026 +0500
[GSoC 2026] Kafka Streams runner: TestPipeline-dispatchable test runner
(PAssert works) (#39362)
---
.../kafka/streams/TestKafkaStreamsRunner.java | 126 +++++++++++++++++++++
.../kafka/streams/TestKafkaStreamsRunnerTest.java | 82 ++++++++++++++
2 files changed, 208 insertions(+)
diff --git
a/runners/kafka-streams/src/test/java/org/apache/beam/runners/kafka/streams/TestKafkaStreamsRunner.java
b/runners/kafka-streams/src/test/java/org/apache/beam/runners/kafka/streams/TestKafkaStreamsRunner.java
new file mode 100644
index 00000000000..1fe4b310789
--- /dev/null
+++
b/runners/kafka-streams/src/test/java/org/apache/beam/runners/kafka/streams/TestKafkaStreamsRunner.java
@@ -0,0 +1,126 @@
+/*
+ * 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 java.io.IOException;
+import java.util.UUID;
+import org.apache.beam.sdk.Pipeline;
+import org.apache.beam.sdk.PipelineResult;
+import org.apache.beam.sdk.PipelineRunner;
+import org.apache.beam.sdk.metrics.MetricResults;
+import org.apache.beam.sdk.options.PipelineOptions;
+import org.apache.beam.sdk.options.PortablePipelineOptions;
+import org.apache.beam.sdk.util.construction.Environments;
+import org.checkerframework.checker.nullness.qual.Nullable;
+import org.joda.time.Duration;
+
+/**
+ * A {@link PipelineRunner} for tests, so {@code TestPipeline}-based suites —
including Beam's
+ * {@code @ValidatesRunner} tests — can run against the Kafka Streams runner
without a broker.
+ *
+ * <p>{@link #run} translates the pipeline and drives it to quiescence through
a {@code
+ * TopologyTestDriver} (via {@link KafkaStreamsTestRunner}), with the SDK
harness in-process
+ * (EMBEDDED environment). The returned {@link PipelineResult} is terminal
({@code DONE}) and
+ * exposes the metrics the harness reported — which is what {@code
+ * TestPipeline.verifyPAssertsSucceeded} uses to check that every {@code
PAssert} in the pipeline
+ * actually ran and succeeded.
+ *
+ * <p>A {@code PAssert} failure inside a DoFn fails the pipeline run — over
the Fn API the DoFn's
+ * error travels as an error string (carrying the assertion text), not as a
Java {@link
+ * AssertionError} object. {@link #run} additionally unwraps and rethrows an
{@link AssertionError}
+ * when one is present in the exception chain (e.g. assertions thrown
runner-side). A {@code
+ * PAssert} that silently never runs is caught by {@code
TestPipeline.verifyPAssertsSucceeded}
+ * comparing the success-counter metric against the number of assertions in
the pipeline.
+ *
+ * <p>Select it with {@code
--runner=org.apache.beam.runners.kafka.streams.TestKafkaStreamsRunner}
+ * in {@code beamTestPipelineOptions}.
+ */
+public final class TestKafkaStreamsRunner extends
PipelineRunner<PipelineResult> {
+
+ private TestKafkaStreamsRunner() {}
+
+ /** Called reflectively by {@link PipelineRunner#fromOptions}. */
+ public static TestKafkaStreamsRunner fromOptions(PipelineOptions options) {
+ return new TestKafkaStreamsRunner();
+ }
+
+ @Override
+ public PipelineResult run(Pipeline pipeline) {
+ // Tests supply generic options; fill in what the Kafka Streams
translation needs. The
+ // pipeline's own options are the ones the translation reads.
+ KafkaStreamsPipelineOptions options =
+ pipeline.getOptions().as(KafkaStreamsPipelineOptions.class);
+ if (options.getApplicationId() == null ||
options.getApplicationId().isEmpty()) {
+ options.setApplicationId("ks-validates-runner-" + UUID.randomUUID());
+ }
+ options
+ .as(PortablePipelineOptions.class)
+ .setDefaultEnvironmentType(Environments.ENVIRONMENT_EMBEDDED);
+
+ MetricResults metrics;
+ try {
+ metrics = KafkaStreamsTestRunner.run(pipeline);
+ } catch (Throwable t) {
+ // A PAssert failure throws an AssertionError inside the harness DoFn,
which reaches us
+ // wrapped in the bundle-processing exception chain. Rethrow the
innermost AssertionError so
+ // the test framework reports the actual assertion.
+ for (Throwable current = t; current != null; current =
current.getCause()) {
+ if (current instanceof AssertionError) {
+ throw (AssertionError) current;
+ }
+ }
+ throw t;
+ }
+ return new TestKafkaStreamsPipelineResult(metrics);
+ }
+
+ /** Terminal result of a test run: state {@code DONE} with the
harness-reported metrics. */
+ private static final class TestKafkaStreamsPipelineResult implements
PipelineResult {
+ private final MetricResults metrics;
+
+ TestKafkaStreamsPipelineResult(MetricResults metrics) {
+ this.metrics = metrics;
+ }
+
+ @Override
+ public State getState() {
+ return State.DONE;
+ }
+
+ @Override
+ public State cancel() throws IOException {
+ throw new UnsupportedOperationException(
+ "A TestKafkaStreamsRunner pipeline has already finished when its
result is returned.");
+ }
+
+ @Override
+ public State waitUntilFinish(@Nullable Duration duration) {
+ return State.DONE;
+ }
+
+ @Override
+ public State waitUntilFinish() {
+ return State.DONE;
+ }
+
+ @Override
+ public MetricResults metrics() {
+ return metrics;
+ }
+ }
+}
diff --git
a/runners/kafka-streams/src/test/java/org/apache/beam/runners/kafka/streams/TestKafkaStreamsRunnerTest.java
b/runners/kafka-streams/src/test/java/org/apache/beam/runners/kafka/streams/TestKafkaStreamsRunnerTest.java
new file mode 100644
index 00000000000..ba62a9d7f49
--- /dev/null
+++
b/runners/kafka-streams/src/test/java/org/apache/beam/runners/kafka/streams/TestKafkaStreamsRunnerTest.java
@@ -0,0 +1,82 @@
+/*
+ * 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.CoreMatchers.containsString;
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.junit.Assert.assertThrows;
+
+import org.apache.beam.sdk.options.PipelineOptions;
+import org.apache.beam.sdk.options.PipelineOptionsFactory;
+import org.apache.beam.sdk.testing.PAssert;
+import org.apache.beam.sdk.testing.TestPipeline;
+import org.apache.beam.sdk.transforms.Count;
+import org.apache.beam.sdk.transforms.Create;
+import org.apache.beam.sdk.values.KV;
+import
org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Throwables;
+import org.junit.Rule;
+import org.junit.Test;
+
+/**
+ * End-to-end tests for {@link TestKafkaStreamsRunner}: a {@link TestPipeline}
with {@link PAssert}
+ * runs against the Kafka Streams runner, and {@code
TestPipeline.verifyPAssertsSucceeded} confirms
+ * through the runner's metrics that every assertion actually executed.
+ *
+ * <p>This is the full {@code @ValidatesRunner} mechanism exercised in-module:
PAssert's {@code
+ * containsInAnyOrder} materializes the actual PCollection through the {@code
GBK + Flatten}
+ * bootstrap path (no side inputs), the assertion DoFn runs in the EMBEDDED
harness, its success
+ * counter travels back through the runner's MetricResults, and a failing
assertion surfaces as an
+ * {@link AssertionError}.
+ */
+public class TestKafkaStreamsRunnerTest {
+
+ private static PipelineOptions options() {
+ PipelineOptions options = PipelineOptionsFactory.create();
+ options.setRunner(TestKafkaStreamsRunner.class);
+ return options;
+ }
+
+ @Rule public final transient TestPipeline pipeline =
TestPipeline.fromOptions(options());
+
+ @Test
+ public void pAssertContainsInAnyOrderSucceeds() {
+ PAssert.that(pipeline.apply("create", Create.of(1, 2,
3))).containsInAnyOrder(3, 2, 1);
+ pipeline.run();
+ }
+
+ @Test
+ public void pAssertOnGroupedResultSucceeds() {
+ PAssert.that(
+ pipeline.apply("create", Create.of("a", "b", "a")).apply("count",
Count.perElement()))
+ .containsInAnyOrder(KV.of("a", 2L), KV.of("b", 1L));
+ pipeline.run();
+ }
+
+ @Test
+ public void failingPAssertFailsTheRun() {
+ // Over the Fn API a DoFn failure travels as an error string, not a Java
AssertionError object,
+ // so the guarantee for a portable runner is that a failing PAssert fails
the pipeline run (and
+ // its message carries the assertion text). A dropped assertion — one that
never runs — would
+ // instead be caught by TestPipeline.verifyPAssertsSucceeded via the
runner's metrics.
+ // Throwable rather than RuntimeException: if the exception chain ever
preserves the DoFn's
+ // AssertionError (an Error), the runner unwraps and rethrows it, which is
also a pass here.
+ PAssert.that(pipeline.apply("create", Create.of(1, 2,
3))).containsInAnyOrder(1, 2, 4);
+ Throwable thrown = assertThrows(Throwable.class, pipeline::run);
+ assertThat(Throwables.getStackTraceAsString(thrown),
containsString("AssertionError"));
+ }
+}
