lnbest0707 commented on code in PR #279:
URL:
https://github.com/apache/flink-connector-kafka/pull/279#discussion_r3574336770
##########
flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/dynamic/source/enumerator/DynamicKafkaSourceEnumerator.java:
##########
@@ -833,7 +944,11 @@ private void reassignReportedSplits() {
new TreeMap<>(pendingReportedSplitsByReader).entrySet()) {
int readerId = readerSplits.getKey();
for (DynamicKafkaSourceSplit split : readerSplits.getValue()) {
- if (isSplitActive(split)) {
+ boolean retainedClusterWaitingForHandoff =
+ split.isRetained()
+ && retainedClusterEnumeratorStates.containsKey(
+ split.getKafkaClusterId());
+ if (!retainedClusterWaitingForHandoff && isSplitActive(split))
{
Review Comment:
Good catch, this was a real gap. Commit 2f729d9c keeps whether each recovery
report was retained before clearing the marker. If the same split is reported
as retained and active, recovery now prefers the active report; retained-only
duplicates keep the furthest offset, while two active reports still fail. This
prevents the stale retained copy from becoming a duplicate active owner and
causing a restart loop. I also added
testRecoveryPrefersActiveSplitOverRetainedDuplicate with retained offset 123 on
reader 0 and active offset 456 on reader 1, asserting 456 wins.
##########
flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/dynamic/source/reader/DynamicKafkaSourceReader.java:
##########
@@ -208,19 +213,38 @@ private InputStatus consolidateInputStatus(
@Override
public void addSplits(List<DynamicKafkaSourceSplit> splits) {
logger.info("Adding splits to reader {}: {}",
readerContext.getIndexOfSubtask(), splits);
+ List<DynamicKafkaSourceSplit> activeSplits = new ArrayList<>();
+ long currentTimeMillis = System.currentTimeMillis();
for (DynamicKafkaSourceSplit split : splits) {
+ if (split.isRetained()) {
+ if (removedClusterStateRetentionMs > 0
+ && split.isRetained(currentTimeMillis)
+ && !activeSplitIds.contains(split.splitId())) {
+ retainedSplits.add(split);
+ }
+ continue;
+ }
+
+ retainedSplits.removeIf(
Review Comment:
Yes. When ownership moves, the old reader can keep its dormant retained copy
until TTL; no reader-side assignment reaches that old reader. Commit 2f729d9c
makes recovery explicitly tolerate that checkpoint shape: it discards the
retained duplicate in favor of the active copy, so it is not reassigned and
disappears from subsequent checkpoints. Two genuinely active owners still fail.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
