[
https://issues.apache.org/jira/browse/FLINK-40622?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18113826#comment-18113826
]
Sylwester Lachiewicz commented on FLINK-40622:
----------------------------------------------
Correction to my earlier comment: the fix has moved from the test into
production code, because the defect is production behaviour rather than test
timing.
A source that verifies topic integrity must not recreate the topic it is
verifying. [PR #311|https://github.com/apache/flink-connector-kafka/pull/311]
now defaults {{allow.auto.create.topics}} to false in {{KafkaSourceBuilder}}
when the integrity check is enabled, through the existing {{maybeOverride(key,
value, false)}} path, so an explicit user value is left alone. Setting the
property on the test's source properties, which is what I described before,
would have hidden the same defect in production: the connector silently
recreates a topic the user deleted and then reports the deletion as a
recreation.
The regression guard is deliberately at two levels, and the reviewer can drop
either: a {{KafkaSourceBuilderTest}} case that runs in milliseconds without a
broker, and a reader-level case in {{SourceTopicIntegrityITCase}} that
exercises a real consumer and needs no savepoint race. Both were checked
against the unfixed builder and fail without it.
Verified locally: {{SourceTopicIntegrityITCase}} 10 tests and
{{KafkaSourceBuilderTest}} 25 tests, 0 failures, run twice, 0 Checkstyle
violations. A single unloaded machine is weak ground for a load-dependent race,
so CI is the stronger signal.
Separately, and not addressed in that PR: {{testTopicIntegrityFailure}} never
cancels its second job, unlike {{testTopicIntegritySuccess}} since
[FLINK-40570|https://issues.apache.org/jira/browse/FLINK-40570]. Adding one
test to the class was enough to exhaust the MiniCluster slots and fail
unrelated cases with {{NoResourceAvailableException}}. That looks worth its own
ticket.
_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)
