MartijnVisser commented on code in PR #295:
URL:
https://github.com/apache/flink-connector-kafka/pull/295#discussion_r4056853450
##########
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:
You are right, I reproduced it. With the metadata refresh outstanding, the
restored sub enumerator pushed a newly discovered partition to the reader whose
report was still in the pending map, so the merge is needed and not just
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]
