[
https://issues.apache.org/jira/browse/FLINK-40622?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18113812#comment-18113812
]
Sylwester Lachiewicz commented on FLINK-40622:
----------------------------------------------
Root cause is topic auto-creation by the source's own readers, not fork
contention.
Neither side disables it: {{auto.create.topics.enable}} is not set on
{{TestKafkaContainer}} and {{allow.auto.create.topics}} appears nowhere in the
repository, so both take their default of true. After the restart a restored
reader polls its checkpointed partitions and recreates the source topic. The
integrity check then finds a topic with a fresh id, takes {{failIfRecreated}}
and reports "was recreated", so the awaited "is missing" never appears and the
wait can only expire.
Which side wins is a race between the restored readers and the first discovery
pass. That is why it only shows under load, and why raising the wait from 2s to
20s in [FLINK-40589|https://issues.apache.org/jira/browse/FLINK-40589] could
not help: the awaited message was never coming.
Evidence, in the order it was established:
* Rebuilt from the job logs in this ticket: a continuous 20 second window of
"Topic integrity mismatch ... got <id>" at roughly 20 per second, matching the
50ms discovery interval, which is the timeout window itself. Your counts line
up with this reading: the 405 "was recreated" are the failing wait spinning,
while passing runs log "is missing in current topics []" with an empty list, so
the broker does start empty.
* A probe consumer using {{assign}} plus {{poll}}, as
{{KafkaPartitionSplitReader}} does, against a missing topic: the topic exists
afterwards with default settings and does not with
{{allow.auto.create.topics=false}}. {{poll}} itself does not throw, so readers
recreate the topic silently.
* Injecting such a reader before the resume turns the flake into a
deterministic failure of exactly the three {{recreateTopic=false}}
parameterizations, with the same message as CI. The same injected reader with
auto-creation disabled passes.
Fix in [PR #311|https://github.com/apache/flink-connector-kafka/pull/311]: set
{{allow.auto.create.topics=false}} on the test's source properties so a deleted
topic stays deleted. Draft until CI is green. Locally the class is 9 tests, 0
failures.
This does not argue against
[FLINK-40618|https://issues.apache.org/jira/browse/FLINK-40618]; lowering the
fork count changes how often the race is lost, not whether it exists.
Worth a separate ticket: production readers carry the same default, so the
connector can recreate a topic the user deleted, which reports as "was
recreated" rather than "is missing".
_This comment was created with AI assistance._
> SourceTopicIntegrityTest.testTopicIntegrityFailure times out waiting for the
> job to fail with "Topic ... is missing"
> --------------------------------------------------------------------------------------------------------------------
>
> Key: FLINK-40622
> URL: https://issues.apache.org/jira/browse/FLINK-40622
> Project: Flink
> Issue Type: Bug
> Components: Connectors / Kafka
> Affects Versions: kafka-5.1.0
> Reporter: Martijn Visser
> Priority: Major
> Labels: pull-request-available, test-stability
>
> Three occurrences, one before and two after FLINK-40589 raised the wait from
> 2 s to 20 s:
> -
> https://github.com/apache/flink-connector-kafka/actions/runs/34116575758/job/101724687796
> (main, 2026-09-07, JDK 17, parameterization [2], 45.43 s)
> -
> https://github.com/apache/flink-connector-kafka/actions/runs/34343298307/job/102637207151
> (PR #289, 2026-09-09, JDK 17, [1], 48.86 s)
> -
> https://github.com/apache/flink-connector-kafka/actions/runs/34392134730/job/102602782562
> (PR #305, 2026-09-09, JDK 11, [2], 63.22 s)
> {code}
> [ERROR]
> org.apache.flink.connector.kafka.source.SourceTopicIntegrityTest.testTopicIntegrityFailure(SourceSubscriptionMode,
> boolean, MiniCluster)[1] -- Time elapsed: 48.86 s <<< ERROR!
> java.util.concurrent.TimeoutException: Waiting for job to fail with
> org.apache.flink.connector.kafka.source.enumerator.metadata.TopicIntegrityException:
> Topic SourceTopicIntegrityTest_source-topic is missing
> at
> org.apache.flink.core.testutils.CommonTestUtils.waitUtil(CommonTestUtils.java:218)
> at
> org.apache.flink.connector.kafka.source.SourceTopicIntegrityTest.testTopicIntegrityFailure(SourceTopicIntegrityTest.java:206)
> {code}
> In each job log the SourceCoordinator reports the job failing repeatedly with
> the other integrity message during the class run, for example (job
> 102637207151):
> {code}
> 22:27:48,944 [SourceCoordinator-Source: kafka-source] ERROR
> org.apache.flink.runtime.source.coordinator.SourceCoordinatorContext [] -
> Exception while handling result from async call in SourceCoordinator-Source:
> kafka-source. Triggering job failover.
> ...
>
> Caused by:
> org.apache.flink.connector.kafka.source.enumerator.metadata.TopicIntegrityException:
> Topic SourceTopicIntegrityTest_source-topic was recreated
> {code}
> Counts of "Caused by: ... TopicIntegrityException" lines per job log, all
> parameterizations of the class included: 102637207151: "was recreated" 405,
> "is missing" 4; 101724687796: 10 and 6; 102602782562: 404 and 8. The class
> runs in the unit-test surefire execution with forkCount=4 (FLINK-40618), so
> the log interleaves other forks.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
