Savonitar commented on code in PR #300:
URL:
https://github.com/apache/flink-connector-kafka/pull/300#discussion_r4058115506
##########
flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/dynamic/source/enumerator/DynamicKafkaSourceEnumerator.java:
##########
@@ -785,11 +805,17 @@ private boolean tryCompletePendingReaderRegistration() {
return true;
}
- readerRecoveryGate.markInitialRegistrationComplete();
- if (readerRecoveryGate.hasReportedSplits()) {
- reassignReportedSplits();
- } else {
- flushPendingSplitAssignmentsForRegisteredReaders();
+ // Draining the gate clears its pending state before reassignment
finishes.
Review Comment:
Done in 0f703b13. Extracted runWithSplitAssignmentInProgress(Runnable) and
used it for all three assignment windows. The flag is reset in finally and the
completion re-checks preserve their existing ordering. I folded the helper
extraction into the amended commit while rebasing onto main. The assignment and
completion ordering is unchanged.
--
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]
