MartijnVisser commented on code in PR #295:
URL:
https://github.com/apache/flink-connector-kafka/pull/295#discussion_r4053603633
##########
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:
Can this see two different non-empty reports for one reader? A reader with a
pending entry holds no splits, so its next report is empty or the same set. If
it is only defensive, say so here.
--
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]
