MartijnVisser commented on code in PR #323:
URL:
https://github.com/apache/flink-connector-kafka/pull/323#discussion_r4053530892
##########
flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/dynamic/source/enumerator/DynamicKafkaSourceEnumerator.java:
##########
@@ -801,16 +828,28 @@ private boolean allReadersRegistered() {
}
private void addReaderToClusterEnumerators(int subtaskId) {
- splitAssignmentStrategy.onReaderAdded(subtaskId);
- clusterEnumeratorMap.forEach(
- (cluster, subEnumerator) ->
subEnumerator.addReader(subtaskId));
+ addReadersToClusterEnumerators(Collections.singletonList(subtaskId));
}
private void flushPendingSplitAssignmentsForRegisteredReaders() {
List<Integer> registeredReaders = new
ArrayList<>(enumContext.registeredReaders().keySet());
Collections.sort(registeredReaders);
- for (int readerId : registeredReaders) {
- addReaderToClusterEnumerators(readerId);
+ addReadersToClusterEnumerators(registeredReaders);
+ }
+
+ private void addReadersToClusterEnumerators(List<Integer> readers) {
+ boolean previousFlushInProgress = flushingPendingSplitAssignments;
Review Comment:
Same deferral as FLINK-39712's #300; after the rebase this part goes.
##########
flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/dynamic/source/enumerator/ReaderRecoveryGate.java:
##########
@@ -65,32 +55,31 @@ class ReaderRecoveryGate {
private final Map<Integer, List<DynamicKafkaSourceSplit>>
pendingReportedSplitsByReader =
new HashMap<>();
+ /** Readers whose checkpoint reports are being reconciled before pending
assignments flush. */
+ private final Set<Integer> readersAwaitingSplitReconciliation = new
HashSet<>();
+
/** Readers whose metadata update events were deferred during recovery. */
private final Set<Integer> pendingMetadataUpdateReaders = new HashSet<>();
ReaderRecoveryGate(boolean restoredFromCheckpoint) {
this.initialReaderRegistrationPending = restoredFromCheckpoint;
}
- /** Records splits a reader reported on registration; an empty report is
ignored. */
+ /** Records the current report during initial recovery, including an empty
replacement. */
void recordReportedSplits(int subtaskId, List<DynamicKafkaSourceSplit>
reportedSplits) {
- if (!reportedSplits.isEmpty()) {
+ if (initialReaderRegistrationPending) {
Review Comment:
An empty re-registration inside the recovery window now wipes the earlier
report. That is what a reader restoring from a checkpoint taken in that window
sends, and its splits then stay marked assigned and are never handed out.
Two-split restore: main assigns both, this branch none; test in a comment
below. Please keep ignoring empty reports, as #295 does.
--
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]
