[GSoC 2026] Kafka Streams runner: portable ValidatesRunner suite for Python

[junaiddshaukat]

Summary

Part of #18479. 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 — until now that was an argument from the design rather than something demonstrated.

29 tests pass, 46 are skipped, none fail. Every skip names the issue that would implement the feature it needs, so the file also reads as a capability statement that CI keeps honest.

docker run -d -p 9092:9092 --name beam-kafka apache/kafka:4.0.0
./gradlew :sdks:python:test-suites:portable:py312:kafkaStreamsValidatesRunner

Unlike the Flink and Spark suites this one needs a broker, because the runner executes on Kafka rather than on a cluster of its own. It is therefore not wired into an aggregate build; --bootstrap_servers defaults to localhost:9092 and can be pointed elsewhere with -PkafkaStreamsBootstrapServers=....

Three bugs it found

Each of these breaks an ordinary pipeline, and none was reachable from the Java suite.

Failures were reported as "unknown error". Kafka Streams moves the client to ERROR and keeps the exception to itself. Nothing registered an uncaught exception handler, so run() returned a failed result carrying no reason and the job service had nothing to report. The runner now keeps the first failure and rethrows it, which is what the job service turns into the job's error message. This one mattered beyond itself: it is what made the other two diagnosable.

State store names were not sanitized for Kafka's topic rules. Repartition and bootstrap topic names already were, but Kafka Streams names a persistent store's changelog topic after the store, and the store names embedded raw transform ids:

InvalidTopicException: Topic name is invalid:
  '...-CombinePerKey(MeanCombineFn)/Group-state-changelog'

Any pipeline with a transform whose name contains a character a topic may not — which is most Python pipelines — failed at runtime. Six store names across GroupByKey, Impulse and Read were affected. The Java suite could not see this because TopologyTestDriver never creates topics, and the broker integration tests use hand-written transform names like emit.

portableMetrics() returned an empty result, so a pipeline from another SDK saw no metrics at all even though the same values were already available to a Java one. It now reports what the SDK harness measured, as attempted only — deliberately not also as committed, since these values are not tied to the commit of the records that produced them (#39635).

What the skips say

Side inputs (#39628), stateful ParDo and timers (#39629), merging windows (#39630), splittable DoFn (#39631), TestStream (#39632), committed metrics (#39635), bundle finalization (#18479).

Two are worth calling out because they were not obvious:

• CombineGlobally expands to a stage with side inputs, so it is unsupported, while CombinePerKey works through GroupByKey. That distinction was not in the documentation.
• A Python pipeline's sources cannot go through the runner's primitive Read path, which carries a serialized Java source. Reading them needs splittable DoFn rather than anything specific to Read.

Testing

./gradlew :runners:kafka-streams:build                    # 99 unit tests, spotless + checker + errorprone
./gradlew :runners:kafka-streams:validatesRunner          # 59 tests
./gradlew :runners:kafka-streams:brokerIntegrationTest    # 4 tests

All green, plus the 29 portable tests above.

A note on how the skip list was arrived at, since it is the part worth doubting. It is empirical rather than predicted: I ran the suite, read each failure, and reproduced the ones I did not immediately believe as standalone pipelines before deciding they were missing features rather than bugs. Two failures I first took for correctness bugs turned out to be Sessions — merging windows — which the Java suite sickbays for the same reason.

The list was then checked by running the whole suite with every skip stripped out and diffing what actually fails against what is skipped. That is the only way to know a skip list is not quietly claiming less than the runner does, and it caught one: test_sdf was skipped on the assumption that it needed splittable DoFn, and it passes.

[github-actions]

Checks are failing. Will not request review until checks are succeeding. If you'd like to override that behavior, comment assign set of reviewers