[jira] [Created] (FLINK-40618) Connectors/Kafka: SourceTopicIntegrityTest.testTopicIntegritySuccess is flaky, times out waiting for records

["Purushottam Sinha (Jira)" <ji...@apache.org>]

Purushottam Sinha created FLINK-40618:
-----------------------------------------

             Summary: Connectors/Kafka: SourceTopicIntegrityTest.testTopicIntegritySuccess is flaky, times out waiting for records
                 Key: FLINK-40618
                 URL: https://issues.apache.org/jira/browse/FLINK-40618
             Project: Flink
          Issue Type: Bug
            Reporter: Purushottam Sinha


SourceTopicIntegrityTest.testTopicIntegritySuccess intermittently fails on CI with a timeout waiting for records in the sink topic, on the very first job submission (before any savepoint/restart/recreation happens in the test itself).

Observed 3 times across unrelated branches/PRs in a 3-day window (2026-09-07 to 2026-09-09):
- https://github.com/apache/flink-connector-kafka/actions/runs/34200093784/job/102428596113 (PR #304, a NOTICE-copyright-only change -- no code touched)
- https://github.com/apache/flink-connector-kafka/actions/runs/34335588824
- https://github.com/apache/flink-connector-kafka/actions/runs/34200551453 (push to main)

All three show the identical chain in the log:

ERROR TopicIntegrityProvider - Topic integrity mismatch: expected topic Id of SourceTopicIntegrityTest_source-topic to be X, got Y
Caused by: TopicIntegrityException: Topic SourceTopicIntegrityTest_source-topic was recreated
...
[ERROR] SourceTopicIntegrityTest.testTopicIntegritySuccess(SourceSubscriptionMode, MiniCluster) -- Time elapsed: ~130s <<< ERROR!
java.util.concurrent.TimeoutException: Timeout waiting for 100 records in topic SourceTopicIntegrityTest_sink-topic after PT2M

Root cause theory: TopicIntegrityProvider.failIfRecreated (flink-connector-kafka/src/main/java/.../enumerator/metadata/TopicIntegrityProvider.java) treats any single topic-id mismatch between polls as a permanent recreation and fails immediately, with no debounce. The test polls every 50ms (SourceTopicIntegrityTest.DISCOVERY_INTERVAL), which is aggressive enough that a transient metadata read on a freshly-started Testcontainers Kafka broker (cold controller election / topic just created) can be misread as a real recreation on CI under load. Since this happens on the very first job submission in the test (before the test itself ever touches the topic again), the mismatch cannot be a genuine recreation.

This is separate from FLINK-40589 (already fixed), which addressed flakiness in the sibling method testTopicIntegrityFailure (a too-short assertion wait), not this testTopicIntegritySuccess timeout.

Proposed fix: require a topic-id mismatch to be confirmed on two consecutive polls before throwing TopicIntegrityException in TopicIntegrityProvider.failIfRecreated, so a single transient metadata blip does not fail the job. A lower-risk test-only mitigation would be to increase SourceTopicIntegrityTest.DISCOVERY_INTERVAL from 50ms to reduce poll pressure on a freshly-created topic.



--
This message was sent by Atlassian Jira
(v8.20.10#820010)