StephanEwen commented on a change in pull request #15461:
URL: https://github.com/apache/flink/pull/15461#discussion_r605744868
##########
File path:
flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/enumerator/KafkaSourceEnumerator.java
##########
@@ -258,42 +265,52 @@ private void addPartitionSplitChangeToPendingAssignments(
}
// This method should only be invoked in the coordinator executor thread.
- private void assignPendingPartitionSplits() {
+ private void assignPendingPartitionSplits(Set<Integer> pendingReaders) {
Map<Integer, List<KafkaPartitionSplit>> incrementalAssignment = new
HashMap<>();
- pendingPartitionSplitAssignment.forEach(
- (ownerReader, pendingSplits) -> {
- if (!pendingSplits.isEmpty()
- &&
context.registeredReaders().containsKey(ownerReader)) {
- // The owner reader is ready, assign the split to the
owner reader.
- incrementalAssignment
- .computeIfAbsent(ownerReader, r -> new
ArrayList<>())
- .addAll(pendingSplits);
- }
- });
- if (incrementalAssignment.isEmpty()) {
- // No assignment is made.
- return;
+
+ // Check if there's any pending splits for given readers
+ for (int pendingReader : pendingReaders) {
+ checkReaderRegistered(pendingReader);
+
+ // Remove pending assignment for the reader
+ final Set<KafkaPartitionSplit> pendingAssignmentForReader =
+ pendingPartitionSplitAssignment.remove(pendingReader);
+
+ if (pendingAssignmentForReader != null &&
!pendingAssignmentForReader.isEmpty()) {
+ // Put pending assignment into incremental assignment
+ incrementalAssignment
+ .computeIfAbsent(pendingReader, ArrayList::new)
Review comment:
I think this lambda is wrong here. This lambda represents `(key) ->
factory` here.
Because the key is an integer here, this matches the `new ArrayList(int
size)` constructor, and it will create the array list with a capacity equal to
the key. So if the key is 2000 (reader number 2000), it will create an array
list with a capacity of 2000.
--
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.
For queries about this service, please contact Infrastructure at:
[email protected]
