dajac commented on code in PR #22768:
URL: https://github.com/apache/kafka/pull/22768#discussion_r3536299563
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinator.java:
##########
@@ -1656,36 +1677,61 @@ private ConsumerCoordinatorMetrics(MetricsLedger
metrics, String metricGrpPrefix
}
private static class PartitionRackInfo {
- private final Set<String> racks;
+ // Racks of the available replicas, keyed by replica id. Replicas on
brokers without a
+ // configured rack are not tracked, since they cannot affect
rack-aware assignment.
+ private final Map<Integer, String> knownRacks;
+ // Ids of the replicas whose broker is not in the current live-broker
list (they resolve
+ // to an empty node), so their rack is unknown. Tracked by id so that
a broker that is
+ // temporarily unavailable is not mistaken for a rack change.
+ private final Set<Integer> unknownReplicas;
PartitionRackInfo(Optional<String> clientRack, PartitionInfo
partition) {
+ Map<Integer, String> knownRacks = new HashMap<>();
+ Set<Integer> unknownReplicas = new HashSet<>();
if (clientRack.isPresent() && partition.replicas() != null) {
- racks =
Arrays.stream(partition.replicas()).map(Node::rack).collect(Collectors.toSet());
- } else {
- racks = Collections.emptySet();
+ for (Node replica : partition.replicas()) {
+ if (replica.isEmpty())
+ unknownReplicas.add(replica.id());
+ else if (replica.rack() != null)
+ knownRacks.put(replica.id(), replica.rack());
+ }
}
+ this.knownRacks = knownRacks;
+ this.unknownReplicas = unknownReplicas;
}
- @Override
- public boolean equals(Object o) {
- if (this == o) {
- return true;
- }
- if (!(o instanceof PartitionRackInfo)) {
- return false;
- }
- PartitionRackInfo rackInfo = (PartitionRackInfo) o;
- return Objects.equals(racks, rackInfo.racks);
+ /**
+ * Returns true if the two snapshots expose the same set of racks for
assignment purposes.
+ * Replica changes that leave the set of racks unchanged (e.g. a
replica moved to a
+ * different broker on one of the same racks) are not considered a
change. A rack that is
+ * present in one snapshot but missing from the other is tolerated
only when the replica
+ * it belongs to is unavailable (same replica id, rack unknown rather
than changed),
+ * so a temporarily unavailable broker is not a rack change, while
+ * a rack that is actually added or removed still is.
+ */
+ boolean equivalentTo(PartitionRackInfo other) {
+ return racksAccountedForIn(other) &&
other.racksAccountedForIn(this);
Review Comment:
Do we really need to check both ways here?
##########
clients/src/test/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinatorTest.java:
##########
@@ -1545,8 +1545,104 @@ public void
testRackAwareConsumerRebalanceWithNewReplicasOnSameRacks() {
true, false);
}
+ @Test
+ public void testRackAwareConsumerNoRebalanceWhenReplicaBounced() {
+ // A broker hosting a replica is bounced: it first drops out of the
live-broker list
+ // (replica 1 remains in the partition replicas, with an unknown rack)
and then comes
+ // back on the same replica. Neither transition is a rack change, so
no rebalance is
+ // expected at any point.
+ List<String> racks = Arrays.asList("rack-a", "rack-b", "rack-c");
Review Comment:
nit: Should we use `List.of` and `Map.of` where possible?
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinator.java:
##########
@@ -1656,36 +1677,61 @@ private ConsumerCoordinatorMetrics(MetricsLedger
metrics, String metricGrpPrefix
}
private static class PartitionRackInfo {
- private final Set<String> racks;
+ // Racks of the available replicas, keyed by replica id. Replicas on
brokers without a
+ // configured rack are not tracked, since they cannot affect
rack-aware assignment.
+ private final Map<Integer, String> knownRacks;
+ // Ids of the replicas whose broker is not in the current live-broker
list (they resolve
+ // to an empty node), so their rack is unknown. Tracked by id so that
a broker that is
+ // temporarily unavailable is not mistaken for a rack change.
+ private final Set<Integer> unknownReplicas;
Review Comment:
nit: I wonder whether online and offline would be a better terminology. Not
sure...
--
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]
