[
https://issues.apache.org/jira/browse/FLINK-40618?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Purushottam Sinha updated FLINK-40618:
--------------------------------------
Description:
SourceTopicIntegrityTest is a heavyweight integration test -- it starts a
Testcontainers Kafka broker and a 3-TaskManager MiniCluster and drives
savepoint/restore -- but its *Test suffix places it in the unit-test surefire
execution (forkCount=4, 1GB heap). Its heavyweight siblings in the same
package, KafkaSourceITCase and KafkaSourceMigrationITCase, are already named
*ITCase and run in the integration-test execution (forkCount=2, 2GB heap,
reuseForks=false).
h3. Proposed change
Rename SourceTopicIntegrityTest -> SourceTopicIntegrityITCase so it is picked
up by the integration-test phase, consistent with those siblings. This reduces
the number of concurrent Testcontainers-Kafka + MiniCluster forks and gives the
test more heap and a fresh JVM per class. This is a test-classification cleanup
justified by the precedence of KafkaSourceITCase / KafkaSourceMigrationITCase.
h3. Relationship to the observed flakiness (FLINK-40620)
The test has been seen failing intermittently on CI with "Timeout waiting for
100 records in topic SourceTopicIntegrityTest_sink-topic after PT2M". Note this
is NOT a delivery-starvation timeout: the sink actually receives MORE than the
expected count -- the failing logs show "Found 150 records ... (expected: 100)"
(and up to 200 in other runs). The at-least-once sink redelivers records across
source re-reads/failover, and KafkaSourceTestEnv.waitForRecordsInTopic asserts
an EXACT count (count == expected), which can never recover once the count
overshoots, so it waits out the full PT2M.
The assertion-level fix for that is tracked separately in FLINK-40620
(dalelane, PR https://github.com/apache/flink-connector-kafka/pull/308), which
runs each subscription mode twice and asserts "no records lost" for
at-least-once while keeping the exact count only for exactly-once. This ITCase
move is complementary (reduces parallel resource pressure / matches
convention); it is not itself the fix for the over-count assertion.
Kept separate from FLINK-40620 by agreement on the PRs (PR #307 for this
change).
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. 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.
Summary: Connectors/Kafka: Run SourceTopicIntegrity test in the
integration-test phase (was: Connectors/Kafka:
SourceTopicIntegrityTest.testTopicIntegritySuccess is flaky, times out waiting
for records)
> Connectors/Kafka: Run SourceTopicIntegrity test in the integration-test phase
> -----------------------------------------------------------------------------
>
> 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 is a heavyweight integration test -- it starts a
> Testcontainers Kafka broker and a 3-TaskManager MiniCluster and drives
> savepoint/restore -- but its *Test suffix places it in the unit-test surefire
> execution (forkCount=4, 1GB heap). Its heavyweight siblings in the same
> package, KafkaSourceITCase and KafkaSourceMigrationITCase, are already named
> *ITCase and run in the integration-test execution (forkCount=2, 2GB heap,
> reuseForks=false).
> h3. Proposed change
> Rename SourceTopicIntegrityTest -> SourceTopicIntegrityITCase so it is picked
> up by the integration-test phase, consistent with those siblings. This
> reduces the number of concurrent Testcontainers-Kafka + MiniCluster forks and
> gives the test more heap and a fresh JVM per class. This is a
> test-classification cleanup justified by the precedence of KafkaSourceITCase
> / KafkaSourceMigrationITCase.
> h3. Relationship to the observed flakiness (FLINK-40620)
> The test has been seen failing intermittently on CI with "Timeout waiting for
> 100 records in topic SourceTopicIntegrityTest_sink-topic after PT2M". Note
> this is NOT a delivery-starvation timeout: the sink actually receives MORE
> than the expected count -- the failing logs show "Found 150 records ...
> (expected: 100)" (and up to 200 in other runs). The at-least-once sink
> redelivers records across source re-reads/failover, and
> KafkaSourceTestEnv.waitForRecordsInTopic asserts an EXACT count (count ==
> expected), which can never recover once the count overshoots, so it waits out
> the full PT2M.
> The assertion-level fix for that is tracked separately in FLINK-40620
> (dalelane, PR https://github.com/apache/flink-connector-kafka/pull/308),
> which runs each subscription mode twice and asserts "no records lost" for
> at-least-once while keeping the exact count only for exactly-once. This
> ITCase move is complementary (reduces parallel resource pressure / matches
> convention); it is not itself the fix for the over-count assertion.
> Kept separate from FLINK-40620 by agreement on the PRs (PR #307 for this
> change).
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
