[
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, 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.
was:
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.
> 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, 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.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
