This is an automated email from the ASF dual-hosted git repository.
lianetm pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/trunk by this push:
new 673107f49d7 KAFKA-20778: Avoid unneeded rebalance on classic consumer
when replica unavailable (#22768)
673107f49d7 is described below
commit 673107f49d7cb01249d9214cfea6e112d8876a39
Author: Lianet Magrans <[email protected]>
AuthorDate: Wed Jul 8 10:47:39 2026 -0400
KAFKA-20778: Avoid unneeded rebalance on classic consumer when replica
unavailable (#22768)
Improve to avoid unnecessary double rebalance on the classic consumer
when a replica is bounced.
Before this PR, when a replica is bounced it resolves to an empty node
(no host) with a null rack on the consumer metadta view. The consumer's
metadata comparison treated that as a rack change (rack1 to null) and
triggered a rebalance, then a second one when the broker came back (null
to rack1). This affects the classic consumer, every assignor, when
client.rack is configured.
This PR improves the rack comparison to track racks by replica id and
treat an unavailable replica (empty node) as having an unknown rack
rather than a changed one: two metadata snapshots are equivalent if they
expose the same set of racks for available replicas, tolerating a
missing/unknown rack only when the replica it belongs to is unavailable.
Reviewers: David Jacot <[email protected]>
---
.../consumer/internals/ConsumerCoordinator.java | 86 +++++++++---
.../internals/ConsumerCoordinatorTest.java | 154 +++++++++++++++++++--
2 files changed, 207 insertions(+), 33 deletions(-)
diff --git
a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinator.java
b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinator.java
index 476fe8bbd2c..fb0ffe11771 100644
---
a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinator.java
+++
b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinator.java
@@ -75,7 +75,6 @@ import org.slf4j.Logger;
import java.nio.ByteBuffer;
import java.util.ArrayList;
-import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
@@ -1619,7 +1618,29 @@ public final class ConsumerCoordinator extends
AbstractCoordinator {
}
boolean matches(MetadataSnapshot other) {
- return version == other.version ||
partitionsPerTopic.equals(other.partitionsPerTopic);
+ if (version == other.version)
+ return true;
+
+ // A rebalance is only needed if the metadata changed in a way
that could change the
+ // assignment: a subscribed topic was added or removed, a topic's
partition count
+ // changed, or the racks of a partition's replicas changed. But
rack differences that are
+ // only due to a replica broker being temporarily offline (host
and rack become unknown)
+ // are tolerated, so that a broker bounce does not trigger
unnecessary rebalances.
+ // See PartitionRackInfo#equivalentTo.
+ if (partitionsPerTopic.size() != other.partitionsPerTopic.size())
+ return false;
+
+ for (Map.Entry<String, List<PartitionRackInfo>> entry :
partitionsPerTopic.entrySet()) {
+ List<PartitionRackInfo> partitionRacks = entry.getValue();
+ List<PartitionRackInfo> otherPartitionRacks =
other.partitionsPerTopic.get(entry.getKey());
+ if (otherPartitionRacks == null || otherPartitionRacks.size()
!= partitionRacks.size())
+ return false;
+ for (int i = 0; i < partitionRacks.size(); i++) {
+ if
(!partitionRacks.get(i).equivalentTo(otherPartitionRacks.get(i)))
+ return false;
+ }
+ }
+ return true;
}
@Override
@@ -1656,36 +1677,61 @@ public final class ConsumerCoordinator extends
AbstractCoordinator {
}
private static class PartitionRackInfo {
- private final Set<String> racks;
+ // Racks of the online 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> onlineRacks;
+ // Ids of the offline 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 offline is not mistaken for a rack change.
+ private final Set<Integer> offlineReplicas;
PartitionRackInfo(Optional<String> clientRack, PartitionInfo
partition) {
+ Map<Integer, String> onlineRacks = new HashMap<>();
+ Set<Integer> offlineReplicas = 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())
+ offlineReplicas.add(replica.id());
+ else if (replica.rack() != null)
+ onlineRacks.put(replica.id(), replica.rack());
+ }
}
+ this.onlineRacks = onlineRacks;
+ this.offlineReplicas = offlineReplicas;
}
- @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 offline (same replica id, rack unknown rather than
changed),
+ * so a temporarily offline 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);
}
- @Override
- public int hashCode() {
- return Objects.hash(racks);
+ /**
+ * Returns true if every online rack in this PartitionRackInfo
snapshot is either still
+ * present in {@code other}, or belongs to a replica that is offline
in {@code other}
+ * (its rack is unknown there, rather than changed).
+ */
+ private boolean racksAccountedForIn(PartitionRackInfo other) {
+ for (Map.Entry<Integer, String> replicaRack :
onlineRacks.entrySet()) {
+ if (!other.onlineRacks.containsValue(replicaRack.getValue()) &&
+ !other.offlineReplicas.contains(replicaRack.getKey()))
{
+ return false;
+ }
+ }
+ return true;
}
@Override
public String toString() {
- return racks.isEmpty() ? "NO_RACKS" : "racks=" + racks;
+ String racks = onlineRacks.isEmpty() ? "NO_RACKS" : "racks=" +
onlineRacks;
+ return offlineReplicas.isEmpty() ? racks : racks + ",
offlineReplicas=" + offlineReplicas;
}
}
diff --git
a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinatorTest.java
b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinatorTest.java
index c9d050c161d..d98d2d5fd70 100644
---
a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinatorTest.java
+++
b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinatorTest.java
@@ -1545,8 +1545,104 @@ public abstract class ConsumerCoordinatorTest {
true, false);
}
+ @Test
+ public void testRackAwareConsumerNoRebalanceWhenReplicaBounced() {
+ // A broker hosting a replica is bounced: it first goes offline (drops
out of the
+ // live-broker list, while replica 1 remains in the partition replicas
with an unknown
+ // rack) and then comes back online on the same replica. Neither
transition is a rack
+ // change, so no rebalance is expected at any point.
+ List<String> racks = List.of("rack-a", "rack-b", "rack-c");
+ RackAwareAssignor assignor = new RackAwareAssignor(protocol);
+ createRackAwareCoordinator("rack-a", assignor);
+
+ Map<String, List<List<Integer>>> replicas = Map.of(topic1,
+ List.of(List.of(0, 1), List.of(1, 2), List.of(2, 0)));
+ MetadataResponse allBrokersUp = rackAwareMetadata(6, racks, Set.of(),
replicas);
+ MetadataResponse broker1Down = rackAwareMetadata(6, racks, Set.of(1),
replicas);
+
+ subscribeAndJoinAsLeader(Optional.of("rack-a"), assignor,
allBrokersUp);
+
+ // Broker 1 goes down.
+ client.updateMetadata(broker1Down);
+ coordinator.poll(time.timer(0));
+ assertEquals(0, client.requests().size());
+
+ // Broker 1 comes back.
+ client.updateMetadata(allBrokersUp);
+ coordinator.poll(time.timer(0));
+ assertEquals(0, client.requests().size());
+ }
+
+ @Test
+ public void
testRackAwareConsumerRebalanceWithReplicaReassignedToOfflineBroker() {
+ // Partition 0 is reassigned from broker 0 (rack-a) to broker 6, which
is not in the
+ // live-broker list. This is a replica change, not just an offline
broker, so a
+ // rebalance is expected even though broker 6's rack is unknown.
+ verifyRackAwareConsumerRebalance(
+ List.of(List.of(0, 1), List.of(1, 2), List.of(2, 0)),
+ List.of(List.of(6, 1), List.of(1, 2), List.of(2, 0)),
+ true, true);
+ }
+
+ @Test
+ public void testRackAwareConsumerRebalanceWithNewRackWhileBrokerOffline() {
+ // The assignment was taken while broker 1 was offline. A replica on a
new rack (broker 5,
+ // rack-c) is added later: the offline broker must not mask that real
rack change.
+ verifyRackAwareConsumerRebalance(
+ List.of(List.of(0, 1), List.of(1, 2), List.of(2, 0)),
+ List.of(List.of(0, 1, 5), List.of(1, 2), List.of(2, 0)),
+ Set.of(1), Set.of(),
+ true, true);
+ }
+
+ @Test
+ public void testRackAwareConsumerRebalanceWhenOnlineReplicaLosesRack() {
+ // Broker 1 stays online (in the live-broker list) but loses its rack
config. Unlike an
+ // offline broker, its rack really disappeared from the replicas of
partitions 0
+ // and 1, so a rebalance is expected.
+ RackAwareAssignor assignor = new RackAwareAssignor(protocol);
+ createRackAwareCoordinator("rack-a", assignor);
+ Map<String, List<List<Integer>>> replicas = Map.of(topic1,
+ List.of(List.of(0, 1), List.of(1, 2), List.of(2, 0)));
+ MetadataResponse response1 = rackAwareMetadata(
+ List.of("rack-a", "rack-b", "rack-c", "rack-a", "rack-b",
"rack-c"),
+ Set.of(), replicas);
+ MetadataResponse response2 = rackAwareMetadata(
+ Arrays.asList("rack-a", null, "rack-c", "rack-a", "rack-b",
"rack-c"),
+ Set.of(), replicas);
+ verifyRebalanceWithMetadataChange(Optional.of("rack-a"), assignor,
response1, response2, true);
+ }
+
+ @Test
+ public void
testRackAwareConsumerNoRebalanceWhenRedundantReplicaLosesRack() {
+ // Broker 3 loses its rack config, but partition 0 still has a replica
on the same rack
+ // (broker 0, rack-a), so the racks visible to the assignor are
unchanged and no
+ // rebalance is expected.
+ RackAwareAssignor assignor = new RackAwareAssignor(protocol);
+ createRackAwareCoordinator("rack-a", assignor);
+ Map<String, List<List<Integer>>> replicas = Map.of(topic1,
+ List.of(List.of(0, 3), List.of(1, 2), List.of(2, 0)));
+ MetadataResponse response1 = rackAwareMetadata(
+ List.of("rack-a", "rack-b", "rack-c", "rack-a", "rack-b",
"rack-c"),
+ Set.of(), replicas);
+ MetadataResponse response2 = rackAwareMetadata(
+ Arrays.asList("rack-a", "rack-b", "rack-c", null, "rack-b",
"rack-c"),
+ Set.of(), replicas);
+ verifyRebalanceWithMetadataChange(Optional.of("rack-a"), assignor,
response1, response2, false);
+ }
+
+ private void verifyRackAwareConsumerRebalance(List<List<Integer>>
partitionReplicas1,
+ List<List<Integer>>
partitionReplicas2,
+ boolean rackAwareConsumer,
+ boolean expectRebalance) {
+ verifyRackAwareConsumerRebalance(partitionReplicas1,
partitionReplicas2,
+ Set.of(), Set.of(), rackAwareConsumer, expectRebalance);
+ }
+
private void verifyRackAwareConsumerRebalance(List<List<Integer>>
partitionReplicas1,
List<List<Integer>>
partitionReplicas2,
+ Set<Integer> downNodes1,
+ Set<Integer> downNodes2,
boolean rackAwareConsumer,
boolean expectRebalance) {
List<String> racks = Arrays.asList("rack-a", "rack-b", "rack-c");
@@ -1558,23 +1654,24 @@ public abstract class ConsumerCoordinatorTest {
createRackAwareCoordinator(consumerRackId, assignor);
}
- MetadataResponse metadataResponse1 = rackAwareMetadata(6, racks,
Collections.singletonMap(topic1, partitionReplicas1));
- MetadataResponse metadataResponse2 = rackAwareMetadata(6, racks,
Collections.singletonMap(topic1, partitionReplicas2));
+ MetadataResponse metadataResponse1 = rackAwareMetadata(6, racks,
downNodes1, Collections.singletonMap(topic1, partitionReplicas1));
+ MetadataResponse metadataResponse2 = rackAwareMetadata(6, racks,
downNodes2, Collections.singletonMap(topic1, partitionReplicas2));
verifyRebalanceWithMetadataChange(Optional.ofNullable(consumerRackId),
assignor, metadataResponse1, metadataResponse2, expectRebalance);
}
- private void verifyRebalanceWithMetadataChange(Optional<String> rackId,
- MockPartitionAssignor
partitionAssignor,
- MetadataResponse
metadataResponse1,
- MetadataResponse
metadataResponse2,
- boolean expectRebalance) {
+ /**
+ * Subscribes to {@code topic1} and {@code topic2} and completes a first
rebalance as the
+ * group leader, with the given metadata as the initial cluster state, so
that the
+ * assignment metadata snapshot is taken from it.
+ */
+ private void subscribeAndJoinAsLeader(Optional<String> rackId,
+ MockPartitionAssignor
partitionAssignor,
+ MetadataResponse initialMetadata) {
final String consumerId = "leader";
final List<String> topics = Arrays.asList(topic1, topic2);
- final List<TopicPartition> partitions =
metadataResponse1.topicMetadata().stream()
- .flatMap(t -> t.partitionMetadata().stream().map(p -> new
TopicPartition(t.topic(), p.partition())))
- .collect(Collectors.toList());
+ final List<TopicPartition> partitions =
partitionsFromMetadata(initialMetadata);
subscriptions.subscribe(Set.copyOf(topics),
Optional.of(rebalanceListener));
- client.updateMetadata(metadataResponse1);
+ client.updateMetadata(initialMetadata);
coordinator.maybeUpdateSubscriptionMetadata();
client.prepareResponse(groupCoordinatorResponse(node, Errors.NONE));
@@ -1594,6 +1691,25 @@ public abstract class ConsumerCoordinatorTest {
assertEquals(0, rebalanceListener.revokedCount);
assertNull(rebalanceListener.revoked);
assertEquals(1, rebalanceListener.assignedCount);
+ }
+
+ private static List<TopicPartition>
partitionsFromMetadata(MetadataResponse metadata) {
+ return metadata.topicMetadata().stream()
+ .flatMap(t -> t.partitionMetadata().stream().map(p -> new
TopicPartition(t.topic(), p.partition())))
+ .collect(Collectors.toList());
+ }
+
+ private void verifyRebalanceWithMetadataChange(Optional<String> rackId,
+ MockPartitionAssignor
partitionAssignor,
+ MetadataResponse
metadataResponse1,
+ MetadataResponse
metadataResponse2,
+ boolean expectRebalance) {
+ final String consumerId = "leader";
+ final List<String> topics = Arrays.asList(topic1, topic2);
+ final List<TopicPartition> partitions =
partitionsFromMetadata(metadataResponse1);
+ Map<String, List<String>> initialSubscription =
singletonMap(consumerId, topics);
+
+ subscribeAndJoinAsLeader(rackId, partitionAssignor, metadataResponse1);
// Change metadata to trigger rebalance.
client.updateMetadata(metadataResponse2);
@@ -4113,10 +4229,22 @@ public abstract class ConsumerCoordinatorTest {
private static MetadataResponse rackAwareMetadata(int numNodes,
List<String> racks,
+ Set<Integer> downNodes,
Map<String,
List<List<Integer>>> partitionReplicas) {
- final List<Node> nodes = new ArrayList<>(numNodes);
+ List<String> nodeRacks = new ArrayList<>(numNodes);
for (int i = 0; i < numNodes; i++)
- nodes.add(new Node(i, "localhost", 1969 + i, racks.get(i %
racks.size())));
+ nodeRacks.add(racks.get(i % racks.size()));
+ return rackAwareMetadata(nodeRacks, downNodes, partitionReplicas);
+ }
+
+ private static MetadataResponse rackAwareMetadata(List<String> nodeRacks,
+ Set<Integer> downNodes,
+ Map<String,
List<List<Integer>>> partitionReplicas) {
+ final List<Node> nodes = new ArrayList<>(nodeRacks.size());
+ for (int i = 0; i < nodeRacks.size(); i++) {
+ if (!downNodes.contains(i))
+ nodes.add(new Node(i, "localhost", 1969 + i,
nodeRacks.get(i)));
+ }
List<MetadataResponse.TopicMetadata> topicMetadata = new ArrayList<>();
for (Map.Entry<String, List<List<Integer>>> topicPartitionCountEntry :
partitionReplicas.entrySet()) {
