junaiddshaukat opened a new pull request, #39546: URL: https://github.com/apache/beam/pull/39546
## Summary Part of #18479. Makes the runner able to run against a real Kafka cluster, and adds the integration test that proves it. Until now everything ran through `TopologyTestDriver`, which fakes the topics and never builds a `KafkaStreams` application — so the production path had never actually been executed. Running it found two bugs that the test driver cannot structurally catch. This is also the groundwork for two things that were going to need it anyway: running the ValidatesRunner suite from another SDK through the job server, and the failure-recovery and scaling benchmark. ## Creating the runner's topics The runner shuffles through topics it names itself — a bootstrap topic per Impulse and per primitive Read, and a repartition topic per GroupByKey. Kafka Streams creates the internal topics it manages, but these are declared with explicit names through `addSource` and `addSink`, so to Kafka Streams they are ordinary user topics: it does not create them, and refuses to start with `MissingSourceTopicException` when a source topic is missing. Leaving it to the broker's `auto.create.topics.enable` is not a fix either — it is off on many clusters, and a topic auto-created on first fetch gets the broker's default partition count instead of the pipeline's. `KafkaStreamsTopicManager` creates them with an `AdminClient` before the application starts. It only creates topics carrying one of the runner's own prefixes; anything else in the topology is a topic the user named, and creating those implicitly would hide a misconfiguration behind an empty topic. A topic that appears between the check and the create — another instance of the same job starting concurrently — is treated as success. Two new options come with it: `topicPartitions` (the parallelism the shuffled parts of a pipeline can reach) and `topicReplicationFactor`. ## Two bugs the test driver could not catch **Options validation rejected every valid job.** `run` validated the whole `KafkaStreamsPipelineOptions` interface, which inherits `@Required jobEndpoint` from `PortablePipelineOptions`: ``` java.lang.IllegalArgumentException: Missing required value for [...getJobEndpoint(), "Job service endpoint to use..."] ``` `jobEndpoint` is a client-side option — the address a client uses to reach the job server. This code runs on the job server, executing a pipeline that has already been submitted, so it does not apply. Flink's equivalent `PortablePipelineRunner` does not validate here either. Now only the options this runner actually reads are checked, which keeps the clear error that motivated the original call. **The state listener was registered after the application had started.** ``` java.lang.IllegalStateException: Can only set StateListener before calling start(). Current state is: REBALANCING ``` The result object registers a `KafkaStreams` state listener in its constructor, but it was constructed after `start()` had been called, and Kafka Streams only accepts a listener while the application is still in `CREATED`. This threw on every real startup, and the listener is what backs `waitUntilFinish()` and `cancel()`. The result is now built before the application starts, and the ordering requirement is documented on its constructor. ## Testing `KafkaStreamsRunnerBrokerIT` runs `Impulse -> ParDo -> GroupByKey -> ParDo` through the production `KafkaStreamsPipelineRunner` against a Kafka container, and polls the pipeline's metrics until the group counter reaches the expected value. That covers the runner creating its own topics, records round-tripping through a real repartition topic, exactly-once processing, the state stores' changelog, and the application lifecycle. It uses testcontainers rather than an embedded broker because `library.java.testcontainers_kafka` is already a declared Beam dependency and is what KafkaIO's own tests use. It needs Docker, so it is not part of the default build — the default `test` task excludes it and a `brokerIntegrationTest` task runs it: ``` ./gradlew :runners:kafka-streams:brokerIntegrationTest # 1 test, passes in ~9s once the container is up ./gradlew :runners:kafka-streams:validatesRunner # 49 tests ./gradlew :runners:kafka-streams:build # 72 unit tests, spotless + checker + errorprone ``` Also adds `PortableWindowingStrategyTest`, which checks that the windowing the runner reconstructs comes from the language-neutral windowing strategy in the proto — the standard `beam:window_fn:*` URNs every SDK emits — rather than from anything Java-specific. That was a question raised on the windowing PR about whether the runner would work for a pipeline submitted from another SDK. -- 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]
