slachiewicz opened a new pull request, #311: URL: https://github.com/apache/flink-connector-kafka/pull/311
## What is the purpose of the change The deleted-topic cases of `SourceTopicIntegrityTest.testTopicIntegrityFailure` assert that the resumed job fails with `TopicIntegrityException: Topic ... is missing`, and time out instead ([FLINK-40622](https://issues.apache.org/jira/browse/FLINK-40622)). Neither the broker nor the source consumer disables topic auto-creation: `auto.create.topics.enable` is not set on `TestKafkaContainer` and `allow.auto.create.topics` appears nowhere in the repository, so both take their default of `true`. After the restart a restored reader polls its checkpointed partitions and recreates the source topic. The enumerator's integrity check then finds a topic with a fresh id, takes the `failIfRecreated` path and reports `was recreated`, so the awaited message never appears and the wait can only expire. Which side wins is a race between the restored readers and the first discovery pass, which is why the test failed only under load. It also explains why raising the wait from 2s to 20s in FLINK-40589 did not help: the awaited message was never coming. ## Brief change log - `SourceTopicIntegrityTest`: set `allow.auto.create.topics=false` on the source properties, so a deleted topic stays deleted. A comment records why the property is load-bearing. No production code changes. The fork count is left alone; contention changes how often the race is lost, not whether it exists. ## Verifying this change Evidence for the diagnosis, in the order it was established: - Rebuilt from a failing CI job log: a continuous 20 second window of `Topic integrity mismatch ... got <id>` at roughly 20 per second, matching the 50ms discovery interval, which is the timeout window itself. Across that job, 405 mismatch against 10 missing. Passing runs log `is missing in current topics []`, an empty list, so the broker does start empty. - A probe consumer using `assign` plus `poll`, as `KafkaPartitionSplitReader` does, against a missing topic: the topic exists afterwards with the default settings and does not with `allow.auto.create.topics=false`. `poll` itself does not throw, so readers recreate the topic silently. - Injecting such a reader before the resume turns the flake into a deterministic failure of exactly the three `recreateTopic=false` parameterizations, with the same message as CI. The same injected reader with auto-creation disabled passes. Verified: `mvn -pl flink-connector-kafka -Dtest=SourceTopicIntegrityTest test` → 9 tests, 0 failures. `spotless:check` passes under JDK 17. One local machine is a weak reproduction environment for a load-dependent race, so CI on this branch is the stronger signal. ## Does this pull request potentially affect one of the following parts? - Dependencies (does it add or upgrade a dependency): no - The public API, i.e., is any changed class annotated with `@Public(Evolving)`: no - The serializers: no - The runtime per-record code paths (performance sensitive): no ## Documentation - Does this pull request introduce a new feature? no - If yes, how is the feature documented? not applicable *This change was created with AI assistance.* -- 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]
