CalvinConfluent commented on code in PR #14312:
URL: https://github.com/apache/kafka/pull/14312#discussion_r1349612450
##########
metadata/src/main/java/org/apache/kafka/controller/PartitionChangeBuilder.java:
##########
@@ -362,6 +408,38 @@ private void setAssignmentChanges(PartitionChangeRecord
record) {
}
}
+ private void populateTargetElr() {
+ // If the ISR is larger or equal to the min ISR, clear the ELR and
lastKnownELR.
+ if (targetIsr.size() >= minISR) {
+ targetElr = Collections.emptyList();
+ targetLastKnownElr = Collections.emptyList();
+ return;
+ }
+
+ Set<Integer> currentIsrSet =
Arrays.stream(partition.isr).boxed().collect(Collectors.toSet());
+ Set<Integer> targetIsrSet =
targetIsr.stream().collect(Collectors.toSet());
+ Set<Integer> currentElrSet =
Arrays.stream(partition.elr).boxed().collect(Collectors.toSet());
+ Set<Integer> currentLastKnownElrSet =
Arrays.stream(partition.lastKnownElr).boxed().collect(Collectors.toSet());
+
+ // Tracking the ELR. The new elr is expected to
+ // 1. Include the current ISR
+ // 2. Exclude the duplicate replicas between elr and target ISR.
+ // 3. Exclude unclean shutdown replicas.
+ // To do that, we first union the current ISR and current elr, then
filter out the target ISR and unclean shutdown
+ // Replicas.
+ Set<Integer> elrCandidates = Utils.union(HashSet::new, currentElrSet,
currentIsrSet);
+ Set<Integer> newElr = elrCandidates.stream()
+ .filter(replica -> !targetIsrSet.contains(replica) &&
(uncleanShutdownReplicas == null || !uncleanShutdownReplicas.contains(replica)))
+ .collect(Collectors.toSet());
+ targetElr = newElr.stream().collect(Collectors.toList());
Review Comment:
Done.
--
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]
