[
https://issues.apache.org/jira/browse/FLINK-40618?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
ASF GitHub Bot updated FLINK-40618:
-----------------------------------
Labels: pull-request-available (was: )
> 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
> Priority: Major
> Labels: pull-request-available
>
> SourceTopicIntegrityTest.testTopicIntegritySuccess intermittently fails on CI
> with a timeout waiting for records in the sink topic:
> [ERROR]
> SourceTopicIntegrityTest.testTopicIntegritySuccess(SourceSubscriptionMode,
> MiniCluster) <<< ERROR!
> java.util.concurrent.TimeoutException: Timeout waiting for 100 records in
> topic SourceTopicIntegrityTest_sink-topic after PT2M
> at SourceTopicIntegrityTest.java:248
> 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 production 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)
> h3. Root cause (from detailed log analysis of run 34200093784)
> The timeout is at the FIRST waitForRecordsInTopic call
> (SourceTopicIntegrityTest.java:248), before the test does any
> savepoint/restart/recreation. Reconstructing the timeline by timestamp:
> * testTopicIntegritySuccess[3] actually times out at 13:03:56 (its 2-minute
> wait runs 13:01:56 -> 13:03:56).
> * During that window the pipeline starts up cleanly: all 10 partitions
> discovered, splits assigned to all 9 readers, split fetchers and producers
> started. There is NO source-topic integrity failure in this window.
> * Records never complete the round-trip to the sink within PT2M, so the wait
> times out.
> The real signal is Kafka broker instability/overload, not topic-integrity
> logic. In this single class run:
> * ~3,043 Kafka "Connection to node ... could not be established" errors, plus
> "Kafka server timed out" and "Timed out waiting for a node assignment. Call:
> listTopics".
> * ~62 "Starting KafkaServer" events, because every @BeforeEach/@AfterEach
> restarts the ENTIRE Kafka container cluster (9 parametrized cases x
> setup+teardown, and testTopicIntegrityFailure adds an extra mid-test
> restart). Each restart gets a new broker port.
> * Confirmed concurrent contention: a DynamicKafkaSourceEnumerator ("Tested
> Source") runs in parallel inside the failing window. Surefire runs the
> unit-test phase with forkCount=4 (separate JVMs), so multiple
> Testcontainers-Kafka + MiniCluster forks execute at once and contend for
> Docker/CPU/IO on the runner.
> Note: the "TopicIntegrityException: Topic ... was recreated" lines that
> appear in the log are a red herring for this failure -- they all occur AFTER
> 13:03:56 and belong to the sibling testTopicIntegrityFailure variants (which
> recreate the topic on purpose, recreateTopic=true) and to teardown, not to
> the timed-out success run.
> This is separate from FLINK-40589 (already fixed), which addressed a
> too-short assertion wait in the sibling method testTopicIntegrityFailure.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
