[
https://issues.apache.org/jira/browse/FLINK-40615?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18113707#comment-18113707
]
Martijn Visser commented on FLINK-40615:
----------------------------------------
Another SourceTopicIntegrityTest.testTopicIntegritySuccess[3] failure on
FLINK-40137's PR #289, JDK 21:
https://github.com/apache/flink-connector-kafka/actions/runs/34343298307/job/102764298078
{code}
[ERROR]
org.apache.flink.connector.kafka.source.SourceTopicIntegrityTest.testTopicIntegritySuccess(SourceSubscriptionMode,
MiniCluster)[3] -- Time elapsed: 132.1 s <<< ERROR!
java.util.concurrent.TimeoutException: Timeout waiting for 100 records in topic
SourceTopicIntegrityTest_sink-topic after PT2M
at
org.apache.flink.core.testutils.CommonTestUtils.waitUtil(CommonTestUtils.java:218)
at
org.apache.flink.connector.kafka.testutils.KafkaSourceTestEnv.waitForRecordsInTopic(KafkaSourceTestEnv.java:309)
at
org.apache.flink.connector.kafka.source.SourceTopicIntegrityTest.testTopicIntegritySuccess(SourceTopicIntegrityTest.java:248)
{code}
The poll log shows 200 records in the sink topic for the whole wait (40 polls
of "Found 200 records in topic SourceTopicIntegrityTest_sink-topic (expected:
100)"), the pattern described in FLINK-40620.
> KafkaSourceEnumerator assigns duplicate splits when partition discovery
> interval is shorter than split initialization time
> --------------------------------------------------------------------------------------------------------------------------
>
> Key: FLINK-40615
> URL: https://issues.apache.org/jira/browse/FLINK-40615
> Project: Flink
> Issue Type: Bug
> Components: Connectors / Kafka
> Reporter: Sylwester Lachiewicz
> Assignee: Sylwester Lachiewicz
> Priority: Major
> Labels: pull-request-available
>
> {{KafkaSourceEnumerator.getPartitionChange()}} deduplicates discovered
> partitions only against {{assignedSplits}} and
> {{pendingPartitionSplitAssignment}}. It does *not* account for partitions
> whose split initialization ({{initializePartitionSplits}}) is currently
> in-flight on the worker thread.
> When periodic discovery fires faster than {{initializePartitionSplits}}
> completes (which performs blocking {{listOffsets}} RPCs via AdminClient),
> {{checkPartitionChanges}} treats the same partitions as "new" again and
> dispatches a second {{initializePartitionSplits}} call.
> *Impact*: Duplicate splits are assigned to readers, causing duplicate record
> consumption. Users with {{partition.discovery.interval.ms}} set lower than
> the broker's metadata response time can silently get duplicate records.
> *Observed in CI*:
> {{SourceTopicIntegrityTest.testTopicIntegritySuccess[PATTERN]}} with
> {{DISCOVERY_INTERVAL=50ms}} — found 120 records instead of 100 due to
> partitions being assigned twice. See [run 34200551453, job
> 101978105599|https://github.com/apache/flink-connector-kafka/actions/runs/34200551453/job/101978105599].
> *Root cause timeline*:
> # Discovery callable returns 10 partitions → {{checkPartitionChanges}}
> dispatches {{initializePartitionSplits}} to worker (call A)
> # 50ms later, discovery fires again → {{getPartitionChange}} finds same 10
> partitions absent from {{assignedSplits}}/{{pendingPartitionSplitAssignment}}
> (call A still running) → dispatches duplicate {{initializePartitionSplits}}
> (call B)
> # Both calls' handlers run → partitions assigned twice → readers produce
> duplicate records
> *Fix*: Track in-flight partitions in a {{Set<TopicPartition>
> partitionsBeingInitialized}} field. Populated before dispatching
> {{callAsync}} in {{checkPartitionChanges}}, cleared in the handler wrapper
> regardless of success/failure, and excluded from {{getPartitionChange}}
> deduplication.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
