bowenli86 commented on code in PR #323:
URL:
https://github.com/apache/flink-connector-kafka/pull/323#discussion_r4264604847
##########
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:
Fixed in 6ba3e19a. I kept ignoring empty registration reports and preserved
the pending-report merge from #295. I added your two-split re-registration
sequence with distinct checkpoint offsets; it fails with the original overwrite
behavior and passes with the fix. The latest head passes all 84 focused tests.
##########
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:
Rebased onto main with #295 and #300 included. In 56d070d7 I removed the
duplicate flushingPendingSplitAssignments flag and helper, and reused #300's
splitAssignmentInProgress guard. The bounded recovery and local-failure
ordering tests still pass.
--
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]
