[
https://issues.apache.org/jira/browse/FLINK-40618?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Purushottam Sinha updated FLINK-40618:
--------------------------------------
Description:
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.
was:
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, i.e. a different test
class hitting the same shared static Kafka. This lines up with
DynamicKafkaSourceTestHelper / KafkaSourceTestEnv not holding
@ResourceLock("KafkaTestBase").
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.
h3. Proposed fix direction (test infrastructure, not production code)
* Add @ResourceLock("KafkaTestBase") to DynamicKafkaSourceTestHelper /
KafkaSourceTestEnv so Dynamic-Kafka-source tests cannot run concurrently
against the shared static broker. Highest leverage.
* Avoid restarting the whole Kafka cluster on every @BeforeEach/@AfterEach;
reuse a shared broker and only (re)create topics. This removes the ~62-restart
churn that destabilizes the broker.
* Increasing the PT2M waitForRecordsInTopic timeout would only mask the problem
and is not the root cause.
> 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
>
> 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)
