junaiddshaukat opened a new pull request, #39736:
URL: https://github.com/apache/beam/pull/39736
## 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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
