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 5f1658f8b5e [GSoC 2026] Kafka Streams runner: portable ValidatesRunner
suite for Python (#39736)
5f1658f8b5e is described below
commit 5f1658f8b5e0842aaedd12612a0cb44ae5fd3219
Author: M Junaid Shaukat <[email protected]>
AuthorDate: Thu Aug 13 13:34:54 2026 +0500
[GSoC 2026] Kafka Streams runner: portable ValidatesRunner suite for Python
(#39736)
* [GSoC 2026] Kafka Streams runner: portable ValidatesRunner suite for
Python
Runs Beam's portable ValidatesRunner suite against the runner, which is what
shows it works for a pipeline that was not written in Java. 29 tests pass
and
46 are skipped, each skip naming the issue for the feature it needs.
* [GSoC 2026] Kafka Streams runner: integration test for two instances over
three partitions
---
.../kafka/streams/KafkaStreamsPipelineRunner.java | 32 ++-
.../KafkaStreamsPortablePipelineResult.java | 17 +-
.../streams/translation/GroupByKeyTranslator.java | 12 +-
.../streams/translation/ImpulseTranslator.java | 3 +-
.../KafkaStreamsTranslationContext.java | 17 ++
.../kafka/streams/translation/ReadTranslator.java | 6 +-
.../KafkaStreamsPortablePipelineResultTest.java | 96 +++++++
.../kafka/streams/KafkaStreamsRunnerBrokerIT.java | 66 ++++-
.../portability/kafka_streams_runner_test.py | 285 +++++++++++++++++++++
sdks/python/test-suites/portable/common.gradle | 35 +++
sdks/python/tox.ini | 5 +
11 files changed, 558 insertions(+), 16 deletions(-)
diff --git
a/runners/kafka-streams/src/main/java/org/apache/beam/runners/kafka/streams/KafkaStreamsPipelineRunner.java
b/runners/kafka-streams/src/main/java/org/apache/beam/runners/kafka/streams/KafkaStreamsPipelineRunner.java
index 93e1058121e..de08480a61d 100644
---
a/runners/kafka-streams/src/main/java/org/apache/beam/runners/kafka/streams/KafkaStreamsPipelineRunner.java
+++
b/runners/kafka-streams/src/main/java/org/apache/beam/runners/kafka/streams/KafkaStreamsPipelineRunner.java
@@ -18,6 +18,8 @@
package org.apache.beam.runners.kafka.streams;
import java.util.Properties;
+import java.util.UUID;
+import java.util.concurrent.atomic.AtomicReference;
import org.apache.beam.model.pipeline.v1.RunnerApi;
import org.apache.beam.runners.fnexecution.provisioning.JobInfo;
import org.apache.beam.runners.jobsubmission.PortablePipelineResult;
@@ -27,6 +29,7 @@ import
org.apache.beam.runners.kafka.streams.translation.KafkaStreamsTranslation
import org.apache.kafka.streams.KafkaStreams;
import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.streams.Topology;
+import org.apache.kafka.streams.errors.StreamsUncaughtExceptionHandler;
import org.checkerframework.checker.nullness.qual.Nullable;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -77,6 +80,19 @@ public class KafkaStreamsPipelineRunner implements
PortablePipelineRunner {
topology.describe());
KafkaStreams kafkaStreams = new KafkaStreams(topology,
streamsConfig(jobInfo));
+ // Kafka Streams reports a failed task by moving the client to ERROR and
keeping the exception
+ // to itself, which left a failed job with nothing to say beyond "unknown
error". Hold on to the
+ // first failure so this method can rethrow it: the job service turns what
run() throws into the
+ // job's error message.
+ AtomicReference<@Nullable Throwable> failure = new AtomicReference<>();
+ kafkaStreams.setUncaughtExceptionHandler(
+ throwable -> {
+ failure.compareAndSet(null, throwable);
+ LOG.error("Pipeline {} failed", jobInfo.jobId(), throwable);
+ // The pipeline is a job with an owner waiting on it, not a service
to keep alive, so a
+ // failure stops the client rather than replacing the thread and
carrying on.
+ return
StreamsUncaughtExceptionHandler.StreamThreadExceptionResponse.SHUTDOWN_CLIENT;
+ });
// Build the result before starting: it registers a state listener, and
Kafka Streams only
// accepts one while the application is still in the CREATED state.
KafkaStreamsPortablePipelineResult result =
@@ -110,6 +126,12 @@ public class KafkaStreamsPipelineRunner implements
PortablePipelineRunner {
// stream threads and the joins it does would throw straight back out of
an interrupted one.
closeInBackground(kafkaStreams, jobInfo.jobId(), "the job was
cancelled");
}
+ Throwable thrown = failure.get();
+ if (thrown != null) {
+ // Thrown rather than returned as a failed result: the job service reads
the state of what is
+ // returned, but only what is thrown carries a reason the user can act
on.
+ throw new RuntimeException("Pipeline " + jobInfo.jobId() + " failed",
thrown);
+ }
return result;
}
@@ -146,7 +168,15 @@ public class KafkaStreamsPipelineRunner implements
PortablePipelineRunner {
props.put(StreamsConfig.APPLICATION_ID_CONFIG,
pipelineOptions.getApplicationId());
props.put(StreamsConfig.STATE_DIR_CONFIG, pipelineOptions.getStateDir());
props.put(StreamsConfig.PROCESSING_GUARANTEE_CONFIG,
StreamsConfig.EXACTLY_ONCE_V2);
- props.put(StreamsConfig.CLIENT_ID_CONFIG, jobInfo.jobId());
+ // The job id identifies the pipeline, which every instance of it shares,
so on its own it does
+ // not identify an instance. Kafka Streams names threads, consumers and
metrics after the client
+ // id, so two workers running the same job would produce logs and JMX
metrics that cannot be
+ // told
+ // apart — in a deployment whose whole point is that you add workers.
Keeping the job id as the
+ // prefix leaves the pipeline recognizable; the suffix is what makes each
worker distinct, and
+ // is
+ // what Kafka Streams does by default when no client id is set.
+ props.put(StreamsConfig.CLIENT_ID_CONFIG, jobInfo.jobId() + "-" +
UUID.randomUUID());
return props;
}
}
diff --git
a/runners/kafka-streams/src/main/java/org/apache/beam/runners/kafka/streams/KafkaStreamsPortablePipelineResult.java
b/runners/kafka-streams/src/main/java/org/apache/beam/runners/kafka/streams/KafkaStreamsPortablePipelineResult.java
index 81e80b66e30..29c07917986 100644
---
a/runners/kafka-streams/src/main/java/org/apache/beam/runners/kafka/streams/KafkaStreamsPortablePipelineResult.java
+++
b/runners/kafka-streams/src/main/java/org/apache/beam/runners/kafka/streams/KafkaStreamsPortablePipelineResult.java
@@ -26,8 +26,6 @@ import
org.apache.beam.runners.jobsubmission.PortablePipelineResult;
import org.apache.beam.sdk.metrics.MetricResults;
import org.apache.kafka.streams.KafkaStreams;
import org.joda.time.Duration;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
/**
* Result of executing a portable pipeline as a {@link KafkaStreams}
application.
@@ -38,9 +36,6 @@ import org.slf4j.LoggerFactory;
*/
class KafkaStreamsPortablePipelineResult implements PortablePipelineResult {
- private static final Logger LOG =
- LoggerFactory.getLogger(KafkaStreamsPortablePipelineResult.class);
-
private final KafkaStreams kafkaStreams;
// The job's metrics accumulator, shared by reference with the topology's
stage processors, which
// update it as the SDK harness reports bundle metrics.
@@ -126,8 +121,16 @@ class KafkaStreamsPortablePipelineResult implements
PortablePipelineResult {
@Override
public JobApi.MetricResults portableMetrics() throws
UnsupportedOperationException {
- LOG.debug("portableMetrics() not yet implemented in the Kafka Streams
runner");
- return JobApi.MetricResults.newBuilder().build();
+ // How a pipeline from another SDK reads its metrics. The job service asks
for these once the
+ // job is terminal and returns them over the job API. Without it a Python
pipeline saw no
+ // metrics at all, even though the same values were already available to a
Java one.
+ //
+ // Reported as attempted only, and deliberately not also as committed: the
values are what the
+ // SDK harness reported per bundle, which is not tied to the commit of the
records that produced
+ // them. Committed metrics are https://github.com/apache/beam/issues/39635.
+ return JobApi.MetricResults.newBuilder()
+ .addAllAttempted(metricsContainerStepMap.getMonitoringInfos())
+ .build();
}
private static State mapState(KafkaStreams.State state) {
diff --git
a/runners/kafka-streams/src/main/java/org/apache/beam/runners/kafka/streams/translation/GroupByKeyTranslator.java
b/runners/kafka-streams/src/main/java/org/apache/beam/runners/kafka/streams/translation/GroupByKeyTranslator.java
index c0e69302386..a62d937170e 100644
---
a/runners/kafka-streams/src/main/java/org/apache/beam/runners/kafka/streams/translation/GroupByKeyTranslator.java
+++
b/runners/kafka-streams/src/main/java/org/apache/beam/runners/kafka/streams/translation/GroupByKeyTranslator.java
@@ -102,10 +102,14 @@ class GroupByKeyTranslator implements
PTransformTranslator {
String shuffleName = transformId + SHUFFLE_SUFFIX;
String sinkName = transformId + SINK_SUFFIX;
String sourceName = transformId + SOURCE_SUFFIX;
- String stateStoreName = transformId + STATE_STORE_SUFFIX;
- String holdsIndexStoreName = transformId + HOLDS_INDEX_STORE_SUFFIX;
- String timerStoreName = transformId + TIMER_STORE_SUFFIX;
- String timerIndexStoreName = transformId + TIMER_INDEX_STORE_SUFFIX;
+ String stateStoreName =
+ KafkaStreamsTranslationContext.getStoreName(transformId,
STATE_STORE_SUFFIX);
+ String holdsIndexStoreName =
+ KafkaStreamsTranslationContext.getStoreName(transformId,
HOLDS_INDEX_STORE_SUFFIX);
+ String timerStoreName =
+ KafkaStreamsTranslationContext.getStoreName(transformId,
TIMER_STORE_SUFFIX);
+ String timerIndexStoreName =
+ KafkaStreamsTranslationContext.getStoreName(transformId,
TIMER_INDEX_STORE_SUFFIX);
String repartitionTopic =
repartitionTopic(transformId,
context.getPipelineOptions().getApplicationId());
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 3daf7782362..29bd6e6bd9d 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
@@ -70,7 +70,8 @@ class ImpulseTranslator implements PTransformTranslator {
Topology topology = context.getTopology();
String sourceNodeName = transformId + SOURCE_SUFFIX;
- String stateStoreName = transformId + STATE_STORE_SUFFIX;
+ String stateStoreName =
+ KafkaStreamsTranslationContext.getStoreName(transformId,
STATE_STORE_SUFFIX);
String bootstrapTopic = context.getImpulseBootstrapTopic(transformId);
topology.addSource(
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 904bd8a71e3..4a546367716 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
@@ -192,4 +192,21 @@ public class KafkaStreamsTranslationContext {
+ "_"
+ sanitizedTransformId;
}
+
+ /**
+ * Returns the name of a state store belonging to a transform.
+ *
+ * <p>The transform id is sanitized to Kafka's legal topic-name characters
even though a store
+ * name is not itself a topic: Kafka Streams names a persistent store's
changelog topic after the
+ * store, so a transform whose name contains a character a topic may not —
which is ordinary,
+ * {@code CombinePerKey(MeanCombineFn)/Group} is a Beam transform name —
would fail at runtime
+ * when the changelog is created.
+ *
+ * <p>Two transform ids differing only in characters that are replaced would
sanitize to one name.
+ * Kafka Streams rejects a store name that is already taken when the
topology is built, so that
+ * surfaces as a failure to start rather than as two transforms quietly
sharing state.
+ */
+ public static String getStoreName(String transformId, String suffix) {
+ return ILLEGAL_TOPIC_CHARS.matcher(transformId).replaceAll("_") + suffix;
+ }
}
diff --git
a/runners/kafka-streams/src/main/java/org/apache/beam/runners/kafka/streams/translation/ReadTranslator.java
b/runners/kafka-streams/src/main/java/org/apache/beam/runners/kafka/streams/translation/ReadTranslator.java
index 049f29651eb..69006661ee6 100644
---
a/runners/kafka-streams/src/main/java/org/apache/beam/runners/kafka/streams/translation/ReadTranslator.java
+++
b/runners/kafka-streams/src/main/java/org/apache/beam/runners/kafka/streams/translation/ReadTranslator.java
@@ -126,7 +126,8 @@ class ReadTranslator implements PTransformTranslator {
Topology topology = context.getTopology();
String sourceNodeName = transformId + SOURCE_SUFFIX;
- String stateStoreName = transformId + STATE_STORE_SUFFIX;
+ String stateStoreName =
+ KafkaStreamsTranslationContext.getStoreName(transformId,
STATE_STORE_SUFFIX);
String bootstrapTopic = context.getReadBootstrapTopic(transformId);
SerializablePipelineOptions options =
new SerializablePipelineOptions(context.getPipelineOptions());
@@ -185,7 +186,8 @@ class ReadTranslator implements PTransformTranslator {
Topology topology = context.getTopology();
String sourceNodeName = transformId + SOURCE_SUFFIX;
- String stateStoreName = transformId + STATE_STORE_SUFFIX;
+ String stateStoreName =
+ KafkaStreamsTranslationContext.getStoreName(transformId,
STATE_STORE_SUFFIX);
String bootstrapTopic = context.getReadBootstrapTopic(transformId);
SerializablePipelineOptions options =
new SerializablePipelineOptions(context.getPipelineOptions());
diff --git
a/runners/kafka-streams/src/test/java/org/apache/beam/runners/kafka/streams/KafkaStreamsPortablePipelineResultTest.java
b/runners/kafka-streams/src/test/java/org/apache/beam/runners/kafka/streams/KafkaStreamsPortablePipelineResultTest.java
new file mode 100644
index 00000000000..2c44765151b
--- /dev/null
+++
b/runners/kafka-streams/src/test/java/org/apache/beam/runners/kafka/streams/KafkaStreamsPortablePipelineResultTest.java
@@ -0,0 +1,96 @@
+/*
+ * 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.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.hasItem;
+import static org.hamcrest.Matchers.hasProperty;
+import static org.hamcrest.Matchers.is;
+import static org.hamcrest.Matchers.not;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+import org.apache.beam.model.jobmanagement.v1.JobApi;
+import org.apache.beam.runners.core.metrics.MetricsContainerImpl;
+import org.apache.beam.runners.core.metrics.MetricsContainerStepMap;
+import org.apache.beam.sdk.metrics.MetricName;
+import org.apache.kafka.streams.KafkaStreams;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.JUnit4;
+
+/**
+ * Tests for {@link KafkaStreamsPortablePipelineResult}, in particular the
metrics an SDK other than
+ * Java reads its results through.
+ */
+@RunWith(JUnit4.class)
+public class KafkaStreamsPortablePipelineResultTest {
+
+ private static final String STEP = "a-stage";
+ private static final String NAMESPACE = "ns";
+ private static final String COUNTER = "elements";
+
+ private static KafkaStreams idleClient() {
+ KafkaStreams kafkaStreams = mock(KafkaStreams.class);
+ // The result registers a state listener and checks the current state, so
it has to have one.
+ when(kafkaStreams.state()).thenReturn(KafkaStreams.State.CREATED);
+ return kafkaStreams;
+ }
+
+ @Test
+ public void portableMetricsReportWhatTheHarnessMeasured() {
+ MetricsContainerStepMap stepMap = new MetricsContainerStepMap();
+ MetricsContainerImpl container = stepMap.getContainer(STEP);
+ container.getCounter(MetricName.named(NAMESPACE, COUNTER)).inc(7);
+
+ KafkaStreamsPortablePipelineResult result =
+ new KafkaStreamsPortablePipelineResult(idleClient(), stepMap, () ->
{});
+
+ JobApi.MetricResults metrics = result.portableMetrics();
+
+ // A pipeline from another SDK reads these over the job API; before they
were reported the list
+ // was empty and a Python pipeline saw no metrics at all.
+ assertThat(
+ metrics.getAttemptedList(),
+ hasItem(hasProperty("urn", is("beam:metric:user:sum_int64:v1"))));
+ assertThat(metrics.getAttemptedCount(), is(not(0)));
+ }
+
+ @Test
+ public void portableMetricsAreNotReportedAsCommitted() {
+ // The values are what the SDK harness reported per bundle, which is not
tied to the commit of
+ // the records that produced them, so claiming them as committed would be
wrong.
+ // See https://github.com/apache/beam/issues/39635.
+ MetricsContainerStepMap stepMap = new MetricsContainerStepMap();
+ stepMap.getContainer(STEP).getCounter(MetricName.named(NAMESPACE,
COUNTER)).inc(1);
+
+ KafkaStreamsPortablePipelineResult result =
+ new KafkaStreamsPortablePipelineResult(idleClient(), stepMap, () ->
{});
+
+ assertThat(result.portableMetrics().getCommittedCount(), is(0));
+ }
+
+ @Test
+ public void aPipelineThatMeasuredNothingReportsNothing() {
+ KafkaStreamsPortablePipelineResult result =
+ new KafkaStreamsPortablePipelineResult(
+ idleClient(), new MetricsContainerStepMap(), () -> {});
+
+ assertThat(result.portableMetrics().getAttemptedCount(), is(0));
+ }
+}
diff --git
a/runners/kafka-streams/src/test/java/org/apache/beam/runners/kafka/streams/KafkaStreamsRunnerBrokerIT.java
b/runners/kafka-streams/src/test/java/org/apache/beam/runners/kafka/streams/KafkaStreamsRunnerBrokerIT.java
index b060561eaab..acc4130b821 100644
---
a/runners/kafka-streams/src/test/java/org/apache/beam/runners/kafka/streams/KafkaStreamsRunnerBrokerIT.java
+++
b/runners/kafka-streams/src/test/java/org/apache/beam/runners/kafka/streams/KafkaStreamsRunnerBrokerIT.java
@@ -21,7 +21,14 @@ import static org.hamcrest.CoreMatchers.is;
import static org.hamcrest.MatcherAssert.assertThat;
import java.nio.file.Files;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
import java.util.UUID;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
import org.apache.beam.model.pipeline.v1.RunnerApi;
import org.apache.beam.runners.fnexecution.provisioning.JobInfo;
import org.apache.beam.sdk.Pipeline;
@@ -123,11 +130,22 @@ public class KafkaStreamsRunnerBrokerIT {
}
private KafkaStreamsPipelineOptions options(int topicPartitions) {
+ return options(topicPartitions, "ks-broker-it-" + UUID.randomUUID());
+ }
+
+ /**
+ * Options for one runner instance.
+ *
+ * <p>Two instances of the same job share an application id — that is what
puts them in one
+ * consumer group and so splits the work between them — but each needs its
own state directory,
+ * since the local stores are per instance.
+ */
+ private KafkaStreamsPipelineOptions options(int topicPartitions, String
applicationId) {
KafkaStreamsPipelineOptions options =
PipelineOptionsFactory.create().as(KafkaStreamsPipelineOptions.class);
options.setRunner(CrashingRunner.class);
options.setBootstrapServers(kafka.getBootstrapServers());
- options.setApplicationId("ks-broker-it-" + UUID.randomUUID());
+ options.setApplicationId(applicationId);
options.setInternalParallelism(topicPartitions);
options
.as(PortablePipelineOptions.class)
@@ -275,6 +293,52 @@ public class KafkaStreamsRunnerBrokerIT {
assertThat(counterValue(result), is(1L));
}
+ @Test
+ public void twoInstancesShareThreePartitions() throws Exception {
+ // Everything else here runs one instance, which leaves the thing the
runner exists for
+ // untested: the work being split between instances by Kafka's own group
membership.
+ //
+ // Three partitions across two instances is deliberate. It does not
divide, so the instances
+ // take an unequal share, and a watermark aggregator on either of them has
to hear from all
+ // three upstream partitions — some of which are being produced by the
other instance — before
+ // it may let its watermark advance. If the reports were tied to the
instance that produced
+ // them rather than to the partition, this is the shape that would break.
+ String applicationId = "ks-broker-it-" + UUID.randomUUID();
+ List<PipelineResult> results = Collections.synchronizedList(new
ArrayList<>());
+ ExecutorService instances = Executors.newFixedThreadPool(2);
+ try {
+ List<Future<?>> running = new ArrayList<>();
+ for (int instance = 0; instance < 2; instance++) {
+ KafkaStreamsPipelineOptions options = options(3, applicationId);
+ Pipeline pipeline = Pipeline.create(options);
+ buildChainedPipeline(pipeline);
+ running.add(
+ instances.submit(
+ () -> {
+ // run() blocks until its instance has finished, so each
needs its own thread.
+ results.add(runPipeline(pipeline, options));
+ }));
+ }
+ for (Future<?> future : running) {
+ // Fails rather than hangs if an instance never finishes — which is
the interesting way for
+ // this to go wrong, since an instance only stops once every processor
it owns is done.
+ future.get(TIMEOUT.getMillis(), TimeUnit.MILLISECONDS);
+ }
+ } finally {
+ instances.shutdownNow();
+ }
+
+ // The pipeline collapses everything onto one key, so exactly one group
comes out of the second
+ // GroupByKey however the partitions were shared. Each instance counts
what it processed, so the
+ // total across both is what has to be one: a group counted twice would
mean the instances had
+ // both claimed the same partition's data.
+ long groups = 0;
+ for (PipelineResult result : results) {
+ groups += counterValue(result);
+ }
+ assertThat(groups, is(1L));
+ }
+
/**
* Polls the pipeline's metrics until the counter reaches {@code expected}
or the timeout hits.
*/
diff --git
a/sdks/python/apache_beam/runners/portability/kafka_streams_runner_test.py
b/sdks/python/apache_beam/runners/portability/kafka_streams_runner_test.py
new file mode 100644
index 00000000000..9285e0fd3d3
--- /dev/null
+++ b/sdks/python/apache_beam/runners/portability/kafka_streams_runner_test.py
@@ -0,0 +1,285 @@
+#
+# 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.
+#
+
+# pytype: skip-file
+
+import argparse
+import logging
+import shlex
+import unittest
+import uuid
+from shutil import rmtree
+from tempfile import mkdtemp
+
+import pytest
+
+from apache_beam.options.pipeline_options import KafkaStreamsRunnerOptions
+from apache_beam.options.pipeline_options import PortableOptions
+from apache_beam.runners.portability import job_server
+from apache_beam.runners.portability import portable_runner
+from apache_beam.runners.portability import portable_runner_test
+from apache_beam.utils import subprocess_server
+
+# Runs Beam's portable ValidatesRunner suite against the Kafka Streams runner,
which is what shows
+# the runner works for a pipeline that was not written in Java.
+#
+# Needs a Kafka broker, unlike the Flink and Spark suites, because the runner
executes on Kafka
+# rather than on a cluster of its own. Point it at one with
--bootstrap_servers.
+#
+# Run as
+#
+# pytest kafka_streams_runner_test.py[::TestClass::test_case] \
+# --test-pipeline-options="--bootstrap_servers=localhost:9092"
+
+_LOGGER = logging.getLogger(__name__)
+
+
+class KafkaStreamsRunnerTest(portable_runner_test.PortableRunnerTest):
+ _use_grpc = True
+ _use_subprocesses = True
+
+ expansion_port = None
+ kafka_streams_job_server_jar = None
+ bootstrap_servers = 'localhost:9092'
+ environment_type = 'LOOPBACK'
+ environment_options = None
+
+ @pytest.fixture(autouse=True)
+ def parse_options(self, request):
+ if not request.config.option.test_pipeline_options:
+ raise unittest.SkipTest(
+ 'Skipping because --test-pipeline-options is not specified.')
+ test_pipeline_options = request.config.option.test_pipeline_options
+ parser = argparse.ArgumentParser(add_help=True)
+ parser.add_argument(
+ '--kafka_streams_job_server_jar',
+ help='Job server jar to submit jobs.',
+ action='store')
+ parser.add_argument(
+ '--bootstrap_servers',
+ default='localhost:9092',
+ help='Kafka the runner executes on, and creates its own topics in.')
+ parser.add_argument(
+ '--environment_type',
+ default='LOOPBACK',
+ choices=['DOCKER', 'PROCESS', 'LOOPBACK'],
+ help='Set the environment type for running user code. DOCKER runs '
+ 'user code in a container. PROCESS runs user code in '
+ 'automatically started processes. LOOPBACK runs user code on '
+ 'the same process that originally submitted the job.')
+ parser.add_argument(
+ '--environment_option',
+ '--environment_options',
+ dest='environment_options',
+ action='append',
+ default=None,
+ help=(
+ 'Environment configuration for running the user code. '
+ 'Recognized options depend on --environment_type.'))
+ known_args, unknown_args = parser.parse_known_args(
+ shlex.split(test_pipeline_options))
+ if unknown_args:
+ _LOGGER.warning('Discarding unrecognized arguments %s' % unknown_args)
+ self.set_kafka_streams_job_server_jar(
+ known_args.kafka_streams_job_server_jar or
+ job_server.JavaJarJobServer.path_to_beam_jar(
+ ':runners:kafka-streams:job-server:shadowJar'))
+ type(self).bootstrap_servers = known_args.bootstrap_servers
+ self.environment_type = known_args.environment_type
+ self.environment_options = known_args.environment_options
+
+ @classmethod
+ def _subprocess_command(cls, job_port, expansion_port):
+ # Created and used by the job server; removed here so the job server makes
it itself.
+ tmp_dir = mkdtemp(prefix='kafkastreamstest')
+
+ cls.expansion_port = expansion_port
+
+ try:
+ return [
+ subprocess_server.JavaHelper.get_java(),
+ '-jar',
+ cls.kafka_streams_job_server_jar,
+ '--artifacts-dir',
+ tmp_dir,
+ '--job-port',
+ str(job_port),
+ '--artifact-port',
+ '0',
+ '--expansion-port',
+ str(expansion_port),
+ ]
+ finally:
+ rmtree(tmp_dir)
+
+ @classmethod
+ def get_runner(cls):
+ return portable_runner.PortableRunner()
+
+ @classmethod
+ def get_expansion_service(cls):
+ return 'localhost:%s' % cls.expansion_port
+
+ @classmethod
+ def set_kafka_streams_job_server_jar(cls, kafka_streams_job_server_jar):
+ cls.kafka_streams_job_server_jar = kafka_streams_job_server_jar
+
+ def create_options(self):
+ options = super().create_options()
+ options.view_as(PortableOptions).environment_type = self.environment_type
+ options.view_as(
+ PortableOptions).environment_options = self.environment_options
+
+ kafka_streams_options = options.view_as(KafkaStreamsRunnerOptions)
+ kafka_streams_options.bootstrap_servers = self.bootstrap_servers
+ # A fresh application id per pipeline. The id names the consumer group and
the runner's own
+ # topics, so reusing one would have a test resume another test's offsets
and read its data.
+ kafka_streams_options.application_id = 'beam-vr-%s' % uuid.uuid4()
+ return options
+
+ # ---------------------------------------------------------------------------
+ # Features the runner does not support yet. Each skip points at the issue
that
+ # would implement it, so this list doubles as the runner's capability gaps.
+ # ---------------------------------------------------------------------------
+
+ def test_pardo_side_inputs(self):
+ raise unittest.SkipTest("https://github.com/apache/beam/issues/39628";)
+
+ def test_pardo_windowed_side_inputs(self):
+ raise unittest.SkipTest("https://github.com/apache/beam/issues/39628";)
+
+ def test_flattened_side_input(self):
+ raise unittest.SkipTest("https://github.com/apache/beam/issues/39628";)
+
+ def test_multimap_side_input(self):
+ raise unittest.SkipTest("https://github.com/apache/beam/issues/39628";)
+
+ def test_multimap_multiside_input(self):
+ raise unittest.SkipTest("https://github.com/apache/beam/issues/39628";)
+
+ def test_multimap_side_input_type_coercion(self):
+ raise unittest.SkipTest("https://github.com/apache/beam/issues/39628";)
+
+ def test_pardo_unfusable_side_inputs(self):
+ raise unittest.SkipTest("https://github.com/apache/beam/issues/39628";)
+
+ def test_pardo_state_only(self):
+ raise unittest.SkipTest("https://github.com/apache/beam/issues/39629";)
+
+ def test_pardo_timers(self):
+ raise unittest.SkipTest("https://github.com/apache/beam/issues/39629";)
+
+ def test_pardo_timers_clear(self):
+ raise unittest.SkipTest("https://github.com/apache/beam/issues/39629";)
+
+ def test_pardo_state_timers(self):
+ raise unittest.SkipTest("https://github.com/apache/beam/issues/39629";)
+
+ def test_pardo_state_timers_non_standard_coder(self):
+ raise unittest.SkipTest("https://github.com/apache/beam/issues/39629";)
+
+ def test_windowed_pardo_state_timers(self):
+ raise unittest.SkipTest("https://github.com/apache/beam/issues/39629";)
+
+ def test_pardo_dynamic_timer(self):
+ raise unittest.SkipTest("https://github.com/apache/beam/issues/39629";)
+
+ def test_custom_merging_window(self):
+ raise unittest.SkipTest("https://github.com/apache/beam/issues/39630";)
+
+ def test_custom_window_type(self):
+ raise unittest.SkipTest("https://github.com/apache/beam/issues/39630";)
+
+ def test_sdf_with_watermark_tracking(self):
+ raise unittest.SkipTest("https://github.com/apache/beam/issues/39631";)
+
+ def test_sdf_with_sdf_initiated_checkpointing(self):
+ raise unittest.SkipTest("https://github.com/apache/beam/issues/39631";)
+
+ def test_sdf_synthetic_source(self):
+ raise unittest.SkipTest("https://github.com/apache/beam/issues/39631";)
+
+ def test_sdf_with_dofn_as_watermark_estimator(self):
+ raise unittest.SkipTest("https://github.com/apache/beam/issues/39631";)
+
+ def test_callbacks_with_exception(self):
+ raise unittest.SkipTest("https://github.com/apache/beam/issues/18479";)
+
+ def test_register_finalizations(self):
+ raise unittest.SkipTest("https://github.com/apache/beam/issues/18479";)
+
+ def test_batch_pardo_fusion_break(self):
+ # CombineGlobally expands to a stage with side inputs.
+ raise unittest.SkipTest("https://github.com/apache/beam/issues/39628";)
+
+ def test_batch_to_element_pardo(self):
+ # CombineGlobally expands to a stage with side inputs.
+ raise unittest.SkipTest("https://github.com/apache/beam/issues/39628";)
+
+ def test_gbk_side_input(self):
+ raise unittest.SkipTest("https://github.com/apache/beam/issues/39628";)
+
+ def test_pack_combiners(self):
+ # The packed combiners are CombineGlobally, which needs side inputs.
+ raise unittest.SkipTest("https://github.com/apache/beam/issues/39628";)
+
+ def test_pardo_side_input_dependencies(self):
+ raise unittest.SkipTest("https://github.com/apache/beam/issues/39628";)
+
+ def test_pardo_unfusable_side_inputs_with_separation(self):
+ raise unittest.SkipTest("https://github.com/apache/beam/issues/39628";)
+
+ def test_pardo_state_with_custom_key_coder(self):
+ raise unittest.SkipTest("https://github.com/apache/beam/issues/39629";)
+
+ def test_pardo_et_timer_with_no_firing(self):
+ raise unittest.SkipTest("https://github.com/apache/beam/issues/39629";)
+
+ def test_pardo_et_timer_with_no_reset(self):
+ raise unittest.SkipTest("https://github.com/apache/beam/issues/39629";)
+
+ def test_pardo_et_timer_with_no_reset_and_no_clear(self):
+ raise unittest.SkipTest("https://github.com/apache/beam/issues/39629";)
+
+ def test_windowing(self):
+ # Sessions, which are merging windows.
+ raise unittest.SkipTest("https://github.com/apache/beam/issues/39630";)
+
+ def test_windowed_combine_per_key(self):
+ # The fixed and sliding parts pass; the sessions part does not, sessions
being merging windows.
+ raise unittest.SkipTest("https://github.com/apache/beam/issues/39630";)
+
+ def test_reshuffle_after_custom_window(self):
+ raise unittest.SkipTest("https://github.com/apache/beam/issues/39630";)
+
+ def test_metrics(self):
+ # The runner reports attempted values, which do reach a Python pipeline;
this asserts on
+ # committed ones.
+ raise unittest.SkipTest("https://github.com/apache/beam/issues/39635";)
+
+ def test_read(self):
+ # The runner reads sources through the deprecated primitive Read, which
carries a serialized
+ # Java source; a Python pipeline's source is a Python object, so reading
one needs splittable
+ # DoFn support rather than anything specific to Read.
+ raise unittest.SkipTest("https://github.com/apache/beam/issues/39631";)
+
+ # Inherits all other tests from PortableRunnerTest.
+
+
+if __name__ == '__main__':
+ logging.getLogger().setLevel(logging.INFO)
+ unittest.main()
diff --git a/sdks/python/test-suites/portable/common.gradle
b/sdks/python/test-suites/portable/common.gradle
index 17bf9989f28..c8c789ca96d 100644
--- a/sdks/python/test-suites/portable/common.gradle
+++ b/sdks/python/test-suites/portable/common.gradle
@@ -174,6 +174,41 @@ tasks.register("sparkValidatesRunner") {
dependsOn 'sparkCompatibilityMatrixLOOPBACK'
}
+// Unlike the Flink and Spark suites, this one needs a Kafka broker: the
runner executes on Kafka
+// rather than on a cluster of its own. It is therefore not wired into any
aggregate build. Run it
+// against a broker, e.g.
+//
+// docker run -d -p 9092:9092 --name beam-kafka apache/kafka:4.0.0
+// ./gradlew
:sdks:python:test-suites:portable:py312:kafkaStreamsValidatesRunner
+//
+// The broker defaults to localhost:9092; point it elsewhere with
+// -PkafkaStreamsBootstrapServers=host:port.
+//
+// LOOPBACK only, unlike the other runners: it is the environment the suite
has been run in, and a
+// task for an environment nobody has tried would claim more than is known.
+def createKafkaStreamsRunnerTestTask() {
+ def taskName = "kafkaStreamsCompatibilityMatrixLOOPBACK"
+ // Not resolvable until runtime, as for the Spark job server above, so the
path is spelled out.
+ def jobServerJar =
+
"${rootDir}/runners/kafka-streams/job-server/build/libs/beam-runners-kafka-streams-job-server-${version}.jar"
+ def bootstrapServers =
+ project.findProperty('kafkaStreamsBootstrapServers') ?: 'localhost:9092'
+ def options =
+ "--kafka_streams_job_server_jar=${jobServerJar}
--environment_type=LOOPBACK" +
+ " --bootstrap_servers=${bootstrapServers}"
+ def task = toxTask(taskName, 'kafka-streams-runner-test', options)
+ task.configure {
+ dependsOn ':runners:kafka-streams:job-server:shadowJar'
+ }
+ return task
+}
+
+createKafkaStreamsRunnerTestTask()
+
+tasks.register("kafkaStreamsValidatesRunner") {
+ dependsOn 'kafkaStreamsCompatibilityMatrixLOOPBACK'
+}
+
def createPrismRunnerTestTask(String workerType) {
def taskName = "prismCompatibilityMatrix${workerType}"
diff --git a/sdks/python/tox.ini b/sdks/python/tox.ini
index 6dab85083a0..beab0399d50 100644
--- a/sdks/python/tox.ini
+++ b/sdks/python/tox.ini
@@ -330,6 +330,11 @@ extras = test
commands =
bash {toxinidir}/scripts/pytest_validates_runner.sh {envname}
{toxinidir}/apache_beam/runners/portability/spark_runner_test.py {posargs}
+[testenv:kafka-streams-runner-test]
+extras = test
+commands =
+ bash {toxinidir}/scripts/pytest_validates_runner.sh {envname}
{toxinidir}/apache_beam/runners/portability/kafka_streams_runner_test.py
{posargs}
+
[testenv:prism-runner-test]
extras = test
commands =
