Savonitar commented on code in PR #295:
URL:
https://github.com/apache/flink-connector-kafka/pull/295#discussion_r4054656814
##########
flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/dynamic/source/enumerator/ReaderRecoveryGate.java:
##########
@@ -72,11 +75,42 @@ class ReaderRecoveryGate {
this.initialReaderRegistrationPending = restoredFromCheckpoint;
}
- /** Records splits a reader reported on registration; an empty report is
ignored. */
+ /** Merges a reader's registration report with pending splits; an empty
report is ignored. */
void recordReportedSplits(int subtaskId, List<DynamicKafkaSourceSplit>
reportedSplits) {
if (!reportedSplits.isEmpty()) {
- pendingReportedSplitsByReader.put(subtaskId, new
ArrayList<>(reportedSplits));
+ pendingReportedSplitsByReader.put(
+ subtaskId,
+ mergeReportedSplits(
+ pendingReportedSplitsByReader.get(subtaskId),
reportedSplits));
+ }
+ }
+
+ /** Returns a copy of the pending reports without draining them. */
+ Map<Integer, List<DynamicKafkaSourceSplit>> snapshotReportedSplits() {
+ Map<Integer, List<DynamicKafkaSourceSplit>> snapshot = new HashMap<>();
+ pendingReportedSplitsByReader.forEach(
+ (readerId, splits) -> snapshot.put(readerId, new
ArrayList<>(splits)));
+ return snapshot;
+ }
+
+ /**
+ * Pending entries can come from earlier registrations or remapped
checkpoint state. Merging by
+ * split id, preferring the current report, avoids both losing and
duplicating splits.
Review Comment:
There is a concrete case where the sets differ: a restored sub-enumerator
can discover a new partition of an existing topic before the first asynchronous
dynamic-metadata refresh completes. The reader buffers the new split B while
previously reported split A still awaits reassignment, so a checkpoint can
contain A in the enumerator’s pending map and B in reader state. On restore,
both sets must be preserved. I verified this with a deterministic
connector-level diagnostic and checked the runtime checkpoint ordering against
Flink 2.2.1. Updated the comment in 4ee2c4a1 to explain this distinction:
merging the sets is necessary, preferring the current report for overlapping
split IDs is defensive.
--
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]
