[
https://issues.apache.org/jira/browse/FLINK-40398?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18105065#comment-18105065
]
Efrat Levitan commented on FLINK-40398:
---------------------------------------
Copilot analysis:
The failure is in
{{{}DynamicKafkaSourceReaderTest.testHandleSourceEventRetainsRemovedClusterOffsetsUntilExpired{}}}:
* Error: {{NoSuchElementException: No value present}}
* Location:
{{flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/dynamic/source/reader/DynamicKafkaSourceReaderTest.java:289-293}}
h3. What is happening
The test expects the reader state to still contain a split for the removed
cluster:
Java
{{DynamicKafkaSourceSplit retainedCluster0Split =
reader.snapshotState(-1).stream()
.filter(split ->
split.getKafkaClusterId().equals(kafkaClusterId0))
.findFirst()
.orElseThrow();}}
But after the metadata update that removes {{{}kafkaClusterId0{}}},
{{snapshotState(-1)}} no longer contains any split for that cluster, so
{{findFirst()}} fails.
h3. Root cause
The test setup assumes removed-cluster offsets are retained until expiry, but
the reader’s behavior in this run indicates the removed cluster split is being
dropped immediately instead of being marked retained.
This is consistent with the surrounding logic in the same test:
Java
{{reader.handleSourceEvents( new
MetadataUpdateEvent(Collections.singleton(shrunkKafkaStream)));}}
After that update, the test expects:
* cluster0 split retained with {{isRetained() == true}}
* {{retainedUntilMs > System.currentTimeMillis()}}
* snapshot contains both the retained cluster0 split and the active cluster1
split
That expectation is not being met.
h3. Suggested fix
The reader should keep removed-cluster splits in state as retained splits until
the retention timeout expires. In practice, the fix likely belongs in
{{DynamicKafkaSourceReader.handleSourceEvents(...)}} or the logic it delegates
to when processing metadata removals.
The behavior should be:
# Detect when a cluster disappears from metadata.
# Mark all splits for that cluster as retained.
# Assign {{{}retainedUntilMs = now +
STREAM_METADATA_REMOVED_CLUSTER_RETENTION_MS{}}}.
# Keep retained splits in {{snapshotState(...)}} until expiry.
# Once metadata reappears before expiry, reactivate the retained split.
h3. Code-level suggestion
A safe implementation pattern is:
Java
{{if (!activeClusters.contains(split.getKafkaClusterId())) \{
split = split.retainUntil(System.currentTimeMillis() + retentionMs);
}}}
and ensure the split is still included in reader state rather than removed.
If there is already a retention path, check that it is not filtered out before
{{snapshotState()}} or when rebuilding the split assignment after metadata
changes.
h3. Why this is the right fix
The test is specifically verifying retention semantics for removed clusters,
and the failure shows the retained split is missing altogether. So the issue is
not with the assertion itself; it’s with preserving split state across metadata
removal.
h3. Relevant files
* Failing test:
{{flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/dynamic/source/reader/DynamicKafkaSourceReaderTest.java}}
* Metadata parsing helper:
{{flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/testutils/YamlFileMetadataService.java}}
* Workflow/job context: {{.github/workflows/push_pr.yml}}
h3. Summary
Fix the dynamic source reader so removed-cluster splits are retained in state
until the configured retention expires. The test failure shows the split is
being removed too early.
> flink-connector-kafka
> DynamicKafkaSourceReaderTest.testHandleSourceEventRetainsRemovedClusterOffsetsUntilExpired
> flakiness
> --------------------------------------------------------------------------------------------------------------------------
>
> Key: FLINK-40398
> URL: https://issues.apache.org/jira/browse/FLINK-40398
> Project: Flink
> Issue Type: Bug
> Components: Connectors / Kafka
> Affects Versions: 2.3.0
> Reporter: Efrat Levitan
> Priority: Major
>
> Seems related to https://issues.apache.org/jira/browse/FLINK-39837,
> [~lnbest0707] could you ptal?
> https://github.com/apache/flink-connector-kafka/pull/294
> https://github.com/apache/flink-connector-kafka/actions/runs/31738004176/job/94574215905?pr=294
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
