You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by lh...@apache.org on 2024/10/14 06:52:37 UTC
(pulsar) branch master updated: [improve][broker][PIP-379] Improve hash collision handling by restoring consumer when other leaves (#23447)
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pulsar.git
The following commit(s) were added to refs/heads/master by this push:
new b340a17b8ff [improve][broker][PIP-379] Improve hash collision handling by restoring consumer when other leaves (#23447)
b340a17b8ff is described below
commit b340a17b8ffbe5022b5f0de18cfc2aeb68f00259
Author: Lari Hotari <lh...@users.noreply.github.com>
AuthorDate: Mon Oct 14 09:52:31 2024 +0300
[improve][broker][PIP-379] Improve hash collision handling by restoring consumer when other leaves (#23447)
---
...ConsistentHashingStickyKeyConsumerSelector.java | 95 ++++++++++++++++++----
.../pulsar/broker/service/RemovedHashRanges.java | 17 ++++
...istentHashingStickyKeyConsumerSelectorTest.java | 74 ++++++++++++++++-
.../java/org/apache/pulsar/client/api/Range.java | 18 ++++
.../org/apache/pulsar/client/api/RangeTest.java | 32 ++++++++
5 files changed, 218 insertions(+), 18 deletions(-)
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ConsistentHashingStickyKeyConsumerSelector.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ConsistentHashingStickyKeyConsumerSelector.java
index 2559a02f87d..06bac02782e 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ConsistentHashingStickyKeyConsumerSelector.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ConsistentHashingStickyKeyConsumerSelector.java
@@ -19,6 +19,7 @@
package org.apache.pulsar.broker.service;
import java.util.ArrayList;
+import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.NavigableMap;
@@ -27,6 +28,7 @@ import java.util.TreeMap;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.locks.ReadWriteLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;
+import lombok.ToString;
import lombok.extern.slf4j.Slf4j;
import org.apache.pulsar.client.api.Range;
@@ -40,8 +42,66 @@ public class ConsistentHashingStickyKeyConsumerSelector implements StickyKeyCons
private static final String KEY_SEPARATOR = "\0";
private final ReadWriteLock rwLock = new ReentrantReadWriteLock();
+ /**
+ * Represents a hash ring point entry.
+ */
+ @ToString
+ private static class HashRingPointEntry {
+ Consumer selectedConsumer;
+ private List<Consumer> collidingConsumers;
+
+ /**
+ * Create a hash ring entry with a selected consumer.
+ * @param selectedConsumer the selected consumer
+ */
+ HashRingPointEntry(Consumer selectedConsumer) {
+ this.selectedConsumer = selectedConsumer;
+ this.collidingConsumers = null;
+ }
+
+ /**
+ * Add a colliding consumer to the hash ring entry. Colliding consumers are consumers that have the same hash
+ * ring point. A colliding consumer is selected when the selected consumer is removed from the hash ring.
+ * @param consumer the consumer to add
+ */
+ void addCollidingConsumer(Consumer consumer) {
+ if (collidingConsumers == null) {
+ collidingConsumers = new LinkedList<>();
+ }
+ collidingConsumers.add(consumer);
+ }
+
+ /**
+ * Remove a consumer from the hash ring entry. When the selected consumer is removed, the first colliding
+ * consumer is selected as the new selected consumer and removed from the colliding consumers list.
+ * @param consumer the consumer to remove
+ * @return true if the entry is empty and should be removed from the hash ring
+ */
+ boolean removeConsumer(Consumer consumer) {
+ if (selectedConsumer == consumer) {
+ if (collidingConsumers != null) {
+ selectedConsumer = collidingConsumers.remove(0);
+ if (collidingConsumers.isEmpty()) {
+ collidingConsumers = null;
+ }
+ } else {
+ selectedConsumer = null;
+ }
+ } else if (collidingConsumers != null) {
+ // remove using identity comparison
+ collidingConsumers.removeIf(c -> c == consumer);
+ if (collidingConsumers.isEmpty()) {
+ // remove the list instance when there are no more colliding consumers
+ collidingConsumers = null;
+ }
+ }
+ // return true when the entry is empty and should be removed from the hash ring
+ return selectedConsumer == null;
+ }
+ }
+
// Consistent-Hash ring
- private final NavigableMap<Integer, ConsumerIdentityWrapper> hashRing;
+ private final NavigableMap<Integer, HashRingPointEntry> hashRing;
// Tracks the used consumer name indexes for each consumer name
private final ConsumerNameIndexTracker consumerNameIndexTracker = new ConsumerNameIndexTracker();
@@ -84,15 +144,16 @@ public class ConsistentHashingStickyKeyConsumerSelector implements StickyKeyCons
int consumerNameIndex =
consumerNameIndexTracker.increaseConsumerRefCountAndReturnIndex(consumerIdentityWrapper);
int hash = calculateHashForConsumerAndIndex(consumer, consumerNameIndex, i);
- // When there's a collision, the entry won't be added to the hash ring.
+ // When there's a collision, the entry won't be selected in the hash ring.
// This isn't a problem with the consumerNameIndexTracker solution since the collisions won't align
// for all hash ring points when using the same consumer name. This won't affect the overall
// distribution significantly when the number of hash ring points is sufficiently large (>100).
- ConsumerIdentityWrapper existing = hashRing.putIfAbsent(hash, consumerIdentityWrapper);
+ HashRingPointEntry existing = hashRing.putIfAbsent(hash, new HashRingPointEntry(consumer));
if (existing != null) {
hashPointCollisions++;
- // reduce the ref count which was increased before adding since the consumer was not added
- consumerNameIndexTracker.decreaseConsumerRefCount(consumerIdentityWrapper);
+ // Add the consumer to the colliding consumers list. The first colliding consumer is selected
+ // when the selected consumer is removed from the hash ring.
+ existing.addCollidingConsumer(consumer);
} else {
hashPointsAdded++;
}
@@ -147,9 +208,15 @@ public class ConsistentHashingStickyKeyConsumerSelector implements StickyKeyCons
// Remove all the points that were added for this consumer
for (int i = 0; i < numberOfPoints; i++) {
int hash = calculateHashForConsumerAndIndex(consumer, consumerNameIndex, i);
- if (hashRing.remove(hash, consumerIdentityWrapper)) {
- consumerNameIndexTracker.decreaseConsumerRefCount(consumerIdentityWrapper);
- }
+ hashRing.compute(hash, (k, hashRingPointEntry) -> {
+ assert hashRingPointEntry != null : "hash ring entry wasn't found for hash " + hash;
+ if (hashRingPointEntry.removeConsumer(consumer)) {
+ // Remove the entry from the hash ring when there are no more consumers
+ return null;
+ }
+ return hashRingPointEntry;
+ });
+ consumerNameIndexTracker.decreaseConsumerRefCount(consumerIdentityWrapper);
}
}
if (!addOrRemoveReturnsImpactedConsumersResult) {
@@ -172,12 +239,12 @@ public class ConsistentHashingStickyKeyConsumerSelector implements StickyKeyCons
if (hashRing.isEmpty()) {
return null;
}
- Map.Entry<Integer, ConsumerIdentityWrapper> ceilingEntry = hashRing.ceilingEntry(hash);
+ Map.Entry<Integer, HashRingPointEntry> ceilingEntry = hashRing.ceilingEntry(hash);
if (ceilingEntry != null) {
- return ceilingEntry.getValue().consumer;
+ return ceilingEntry.getValue().selectedConsumer;
} else {
// Handle wrap-around in the hash ring, return the first consumer
- return hashRing.firstEntry().getValue().consumer;
+ return hashRing.firstEntry().getValue().selectedConsumer;
}
} finally {
rwLock.readLock().unlock();
@@ -209,8 +276,8 @@ public class ConsistentHashingStickyKeyConsumerSelector implements StickyKeyCons
int lastKey = -1;
Consumer previousConsumer = null;
Range previousRange = null;
- for (Map.Entry<Integer, ConsumerIdentityWrapper> entry: hashRing.entrySet()) {
- Consumer consumer = entry.getValue().consumer;
+ for (Map.Entry<Integer, HashRingPointEntry> entry: hashRing.entrySet()) {
+ Consumer consumer = entry.getValue().selectedConsumer;
Range range;
if (consumer == previousConsumer) {
// join ranges
@@ -226,7 +293,7 @@ public class ConsistentHashingStickyKeyConsumerSelector implements StickyKeyCons
previousRange = range;
}
// Handle wrap-around
- Consumer firstConsumer = hashRing.firstEntry().getValue().consumer;
+ Consumer firstConsumer = hashRing.firstEntry().getValue().selectedConsumer;
if (lastKey != getKeyHashRange().getEnd()) {
Range range;
if (firstConsumer == previousConsumer && previousRange.getEnd() == lastKey) {
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/RemovedHashRanges.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/RemovedHashRanges.java
index 1833c243f89..d9bd502255b 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/RemovedHashRanges.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/RemovedHashRanges.java
@@ -18,6 +18,7 @@
*/
package org.apache.pulsar.broker.service;
+import java.util.Arrays;
import java.util.List;
import lombok.EqualsAndHashCode;
import lombok.ToString;
@@ -71,4 +72,20 @@ public class RemovedHashRanges {
}
return false;
}
+
+ /**
+ * Checks if all removed ranges are fully contained in the provided list of ranges.
+ */
+ public boolean isFullyContainedInRanges(List<Range> otherRanges) {
+ return Arrays.stream(sortedRanges).allMatch(range ->
+ otherRanges.stream().anyMatch(otherRange -> otherRange.contains(range))
+ );
+ }
+
+ /**
+ * Returns the removed hash ranges as a list of ranges.
+ */
+ public List<Range> asRanges() {
+ return Arrays.asList(sortedRanges);
+ }
}
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ConsistentHashingStickyKeyConsumerSelectorTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ConsistentHashingStickyKeyConsumerSelectorTest.java
index 6752cd7cfab..24e14ca3c9c 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ConsistentHashingStickyKeyConsumerSelectorTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ConsistentHashingStickyKeyConsumerSelectorTest.java
@@ -31,6 +31,7 @@ import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
+import java.util.TreeSet;
import java.util.UUID;
import java.util.stream.Collectors;
import java.util.stream.IntStream;
@@ -475,11 +476,11 @@ public class ConsistentHashingStickyKeyConsumerSelectorTest {
}
@Test
- public void testShouldNotContainMappingChangesWhenConsumersLeaveAndRejoinInSameOrder() {
+ public void testShouldContainMinimalMappingChangesWhenConsumerLeavesAndRejoins() {
final ConsistentHashingStickyKeyConsumerSelector selector =
- new ConsistentHashingStickyKeyConsumerSelector(200, true);
+ new ConsistentHashingStickyKeyConsumerSelector(100, true);
final String consumerName = "consumer";
- final int numOfInitialConsumers = 200;
+ final int numOfInitialConsumers = 10;
List<Consumer> consumers = new ArrayList<>();
for (int i = 0; i < numOfInitialConsumers; i++) {
final Consumer consumer = createMockConsumer(consumerName, "index " + i, i);
@@ -498,8 +499,73 @@ public class ConsistentHashingStickyKeyConsumerSelectorTest {
selector.addConsumer(consumers.get(numOfInitialConsumers / 2));
ConsumerHashAssignmentsSnapshot assignmentsAfter = selector.getConsumerHashAssignmentsSnapshot();
+ int removedRangesSize = assignmentsBefore.diffRanges(assignmentsAfter).keySet().stream()
+ .mapToInt(Range::size)
+ .sum();
+ double allowedremovedRangesPercentage = 1; // 1%
+ int hashRangeSize = selector.getKeyHashRange().size();
+ int allowedremovedRanges = (int) (hashRangeSize * (allowedremovedRangesPercentage / 100.0d));
+ assertThat(removedRangesSize).describedAs("Allow up to %d%% of total hash range size to be impacted",
+ allowedremovedRangesPercentage).isLessThan(allowedremovedRanges);
+ }
- assertThat(assignmentsBefore.resolveImpactedConsumers(assignmentsAfter).getRemovedHashRanges()).isEmpty();
+ @Test
+ public void testShouldNotSwapExistingConsumers() {
+ final ConsistentHashingStickyKeyConsumerSelector selector =
+ new ConsistentHashingStickyKeyConsumerSelector(200, true);
+ final String consumerName = "consumer";
+ final int consumerCount = 100;
+ List<Consumer> consumers = new ArrayList<>();
+ for (int i = 0; i < consumerCount; i++) {
+ final Consumer consumer = createMockConsumer(consumerName + i, "index " + i, i);
+ consumers.add(consumer);
+ selector.addConsumer(consumer);
+ }
+ ConsumerHashAssignmentsSnapshot assignmentsBefore = selector.getConsumerHashAssignmentsSnapshot();
+ for (int i = 0; i < consumerCount; i++) {
+ Consumer consumer = consumers.get(i);
+
+ // remove consumer
+ selector.removeConsumer(consumer);
+
+ ConsumerHashAssignmentsSnapshot assignmentsAfter = selector.getConsumerHashAssignmentsSnapshot();
+ assertThat(assignmentsBefore.resolveImpactedConsumers(assignmentsAfter).getRemovedHashRanges())
+ .describedAs(
+ "when a consumer is removed, the removed hash ranges should only be from "
+ + "the removed consumer")
+ .containsOnlyKeys(consumer);
+ assignmentsBefore = assignmentsAfter;
+
+ // add consumer back
+ selector.addConsumer(consumer);
+
+ assignmentsAfter = selector.getConsumerHashAssignmentsSnapshot();
+ List<Range> addedConsumerRanges = assignmentsAfter.getRangesByConsumer().get(consumer);
+
+ Map<Consumer, RemovedHashRanges> removedHashRanges =
+ assignmentsBefore.resolveImpactedConsumers(assignmentsAfter).getRemovedHashRanges();
+ ConsumerHashAssignmentsSnapshot finalAssignmentsBefore = assignmentsBefore;
+ assertThat(removedHashRanges).allSatisfy((c, removedHashRange) -> {
+ assertThat(removedHashRange
+ .isFullyContainedInRanges(finalAssignmentsBefore.getRangesByConsumer().get(c)))
+ .isTrue();
+ assertThat(removedHashRange
+ .isFullyContainedInRanges(addedConsumerRanges))
+ .isTrue();
+ }).describedAs("when a consumer is added back, all removed hash ranges should be ones "
+ + "that are moved from existing consumers to the new consumer.");
+
+ List<Range> allRemovedRanges =
+ ConsumerHashAssignmentsSnapshot.mergeOverlappingRanges(
+ removedHashRanges.entrySet().stream().map(Map.Entry::getValue)
+ .map(RemovedHashRanges::asRanges)
+ .flatMap(List::stream).collect(Collectors.toCollection(TreeSet::new)));
+ assertThat(allRemovedRanges)
+ .describedAs("all removed ranges should be the same as the ranges of the added consumer")
+ .containsExactlyElementsOf(addedConsumerRanges);
+
+ assignmentsBefore = assignmentsAfter;
+ }
}
@Test
diff --git a/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/Range.java b/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/Range.java
index cbca1ef8f06..3db225330d0 100644
--- a/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/Range.java
+++ b/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/Range.java
@@ -94,10 +94,28 @@ public class Range implements Comparable<Range> {
return result;
}
+ /**
+ * Check if the value is in the range.
+ * @param value
+ * @return true if the value is in the range.
+ */
public boolean contains(int value) {
return value >= start && value <= end;
}
+ /**
+ * Check if the range is fully contained in the other range.
+ * @param otherRange
+ * @return true if the range is fully contained in the other range.
+ */
+ public boolean contains(Range otherRange) {
+ return start <= otherRange.start && end >= otherRange.end;
+ }
+
+ /**
+ * Get the size of the range.
+ * @return the size of the range.
+ */
public int size() {
return end - start + 1;
}
diff --git a/pulsar-client/src/test/java/org/apache/pulsar/client/api/RangeTest.java b/pulsar-client/src/test/java/org/apache/pulsar/client/api/RangeTest.java
index 50168221fea..bb443498a16 100644
--- a/pulsar-client/src/test/java/org/apache/pulsar/client/api/RangeTest.java
+++ b/pulsar-client/src/test/java/org/apache/pulsar/client/api/RangeTest.java
@@ -87,6 +87,38 @@ public class RangeTest {
Assert.assertFalse(range.contains(6));
}
+ @Test
+ public void testContainsRange() {
+ Range range = Range.of(5, 10);
+
+ // Test ranges that are fully contained
+ Assert.assertTrue(range.contains(Range.of(6, 8)));
+
+ Assert.assertTrue(range.contains(Range.of(5, 10)));
+
+ Assert.assertTrue(range.contains(Range.of(5, 5)));
+
+ Assert.assertTrue(range.contains(Range.of(5, 8)));
+
+ Assert.assertTrue(range.contains(Range.of(10, 10)));
+
+ Assert.assertTrue(range.contains(Range.of(8, 10)));
+
+ // Test ranges that are not fully contained
+ Assert.assertFalse(range.contains(Range.of(1, 5)));
+
+ Assert.assertFalse(range.contains(Range.of(1, 4)));
+
+ Assert.assertFalse(range.contains(Range.of(1, 10)));
+
+ Assert.assertFalse(range.contains(Range.of(1, 11)));
+
+ Assert.assertFalse(range.contains(Range.of(10, 12)));
+
+ Assert.assertFalse(range.contains(Range.of(11, 20)));
+ }
+
+
@Test
public void testSize() {
Range range = Range.of(0, 0);