You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by he...@apache.org on 2025/04/18 16:10:22 UTC
(pulsar) branch branch-4.0 updated: [fix][broker] Fixes Inconsistent ServiceUnitStateData View (ExtensibleLoadManagerImpl only) (#24186)
This is an automated email from the ASF dual-hosted git repository.
heesung pushed a commit to branch branch-4.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
The following commit(s) were added to refs/heads/branch-4.0 by this push:
new 28f7e949d70 [fix][broker] Fixes Inconsistent ServiceUnitStateData View (ExtensibleLoadManagerImpl only) (#24186)
28f7e949d70 is described below
commit 28f7e949d7079b198c05406a8bc4fb21a18216e9
Author: Heesung Sohn <10...@users.noreply.github.com>
AuthorDate: Fri Apr 18 09:08:03 2025 -0700
[fix][broker] Fixes Inconsistent ServiceUnitStateData View (ExtensibleLoadManagerImpl only) (#24186)
(cherry picked from commit c6be44c710141bb55af0af2c0e94ecf2cac40fdd)
---
.../extensions/channel/ServiceUnitStateData.java | 15 ---------
.../channel/ServiceUnitStateDataTest.java | 24 ++++++++++++++
.../org/apache/pulsar/common/util/Backoff.java | 10 +++++-
.../pulsar/metadata/api/MetadataCacheConfig.java | 5 +++
.../metadata/cache/impl/MetadataCacheImpl.java | 8 +++--
.../tableview/impl/MetadataStoreTableViewImpl.java | 11 ++++++-
.../apache/pulsar/metadata/MetadataCacheTest.java | 37 ++++++++++++++++++++++
7 files changed, 91 insertions(+), 19 deletions(-)
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/channel/ServiceUnitStateData.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/channel/ServiceUnitStateData.java
index 4a990ddbc9b..61a1606ba6e 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/channel/ServiceUnitStateData.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/channel/ServiceUnitStateData.java
@@ -75,19 +75,4 @@ public record ServiceUnitStateData(
public static ServiceUnitState state(ServiceUnitStateData data) {
return data == null ? ServiceUnitState.Init : data.state();
}
-
- @Override
- public boolean equals(Object o) {
- if (this == o) {
- return true;
- }
-
- if (o == null || getClass() != o.getClass()) {
- return false;
- }
-
- ServiceUnitStateData that = (ServiceUnitStateData) o;
-
- return versionId == that.versionId;
- }
}
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/extensions/channel/ServiceUnitStateDataTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/extensions/channel/ServiceUnitStateDataTest.java
index 9bfa8ad3ac5..f9a2d106233 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/extensions/channel/ServiceUnitStateDataTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/extensions/channel/ServiceUnitStateDataTest.java
@@ -21,10 +21,13 @@ package org.apache.pulsar.broker.loadbalance.extensions.channel;
import static org.apache.pulsar.broker.loadbalance.extensions.channel.ServiceUnitState.Assigning;
import static org.apache.pulsar.broker.loadbalance.extensions.channel.ServiceUnitState.Owned;
import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertNotEquals;
import static org.testng.Assert.assertNull;
import static org.assertj.core.api.Assertions.assertThat;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
+import java.util.Map;
+import java.util.Optional;
import org.apache.pulsar.common.util.ObjectMapperFactory;
import org.testng.annotations.Test;
@@ -76,4 +79,25 @@ public class ServiceUnitStateDataTest {
ServiceUnitStateData dst = mapper.readValue(json, ServiceUnitStateData.class);
assertEquals(dst, src);
}
+
+ @Test
+ public void equalsTest() {
+ // record ServiceUnitStateData(
+ // ServiceUnitState state, String dstBroker, String sourceBroker,
+ // Map<String, Optional<String>> splitServiceUnitToDestBroker, boolean force,
+ // long timestamp, long versionId) {
+ var d1 = new ServiceUnitStateData(Assigning, "A", "B", Map.of("A", Optional.of("B")), true, 0, 1);
+ var d2 = new ServiceUnitStateData(Assigning, "A", "B", Map.of("A", Optional.of("B")), true, 0, 1);
+ assertEquals(d1, d2);
+ var d3 = new ServiceUnitStateData(Assigning, "C", "B", 1);
+ var d4 = new ServiceUnitStateData(Assigning, "A", "B", Map.of("A", Optional.of("C")), true, 0, 1);
+ assertNotEquals(d1, d3);
+ assertNotEquals(d1, d4);
+
+ var d5 = new ServiceUnitStateData(Assigning, "A", "B", Map.of("A", Optional.of("B")), true, 0, 2);
+ assertNotEquals(d1, d5);
+
+ var d6 = new ServiceUnitStateData(Assigning, "C", "B", Map.of("A", Optional.of("B")), true, 0, 1);
+ assertNotEquals(d1, d6);
+ }
}
\ No newline at end of file
diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/util/Backoff.java b/pulsar-common/src/main/java/org/apache/pulsar/common/util/Backoff.java
index d7381792f60..842d3bbaa97 100644
--- a/pulsar-common/src/main/java/org/apache/pulsar/common/util/Backoff.java
+++ b/pulsar-common/src/main/java/org/apache/pulsar/common/util/Backoff.java
@@ -43,9 +43,13 @@ public class Backoff {
TimeUnit unitMandatoryStop, Clock clock) {
this.initial = unitInitial.toMillis(initial);
this.max = unitMax.toMillis(max);
+ if (initial == 0 && max == 0 && mandatoryStop == 0) {
+ this.mandatoryStopMade = true;
+ }
this.next = this.initial;
this.mandatoryStop = unitMandatoryStop.toMillis(mandatoryStop);
this.clock = clock;
+ this.firstBackoffTimeInMillis = 0;
}
public Backoff(long initial, TimeUnit unitInitial, long max, TimeUnit unitMax, long mandatoryStop,
@@ -91,7 +95,11 @@ public class Backoff {
public void reset() {
this.next = this.initial;
- this.mandatoryStopMade = false;
+ if (initial == 0 && max == 0 && mandatoryStop == 0) {
+ this.mandatoryStopMade = true;
+ } else {
+ this.mandatoryStopMade = false;
+ }
}
public static boolean shouldBackoff(long initialTimestamp, TimeUnit unitInitial, int failedAttempts,
diff --git a/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/api/MetadataCacheConfig.java b/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/api/MetadataCacheConfig.java
index 641d4bde5a2..e5334038290 100644
--- a/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/api/MetadataCacheConfig.java
+++ b/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/api/MetadataCacheConfig.java
@@ -39,6 +39,11 @@ public class MetadataCacheConfig<T> {
.setMax(3, TimeUnit.SECONDS)
.setMandatoryStop(30, TimeUnit.SECONDS);
+ public static final BackoffBuilder NO_RETRY_BACKOFF_BUILDER =
+ new BackoffBuilder().setInitialTime(0, TimeUnit.MILLISECONDS)
+ .setMax(0, TimeUnit.SECONDS)
+ .setMandatoryStop(0, TimeUnit.SECONDS);
+
/**
* Specifies that active entries are eligible for automatic refresh once a fixed duration has
* elapsed after the entry's creation, or the most recent replacement of its value.
diff --git a/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/cache/impl/MetadataCacheImpl.java b/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/cache/impl/MetadataCacheImpl.java
index 66c8388c83b..8607354f15c 100644
--- a/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/cache/impl/MetadataCacheImpl.java
+++ b/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/cache/impl/MetadataCacheImpl.java
@@ -339,8 +339,12 @@ public class MetadataCacheImpl<T> implements MetadataCache<T>, Consumer<Notifica
objCache.synchronous().invalidate(key);
long elapsed = System.currentTimeMillis() - backoff.getFirstBackoffTimeInMillis();
if (backoff.isMandatoryStopMade()) {
- result.completeExceptionally(new TimeoutException(
- String.format("Timeout to update key %s. Elapsed time: %d ms", key, elapsed)));
+ if (backoff.getFirstBackoffTimeInMillis() == 0) {
+ result.completeExceptionally(ex.getCause());
+ } else {
+ result.completeExceptionally(new TimeoutException(
+ String.format("Timeout to update key %s. Elapsed time: %d ms", key, elapsed)));
+ }
return null;
}
final var next = backoff.next();
diff --git a/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/tableview/impl/MetadataStoreTableViewImpl.java b/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/tableview/impl/MetadataStoreTableViewImpl.java
index 4f9aad0ba65..aef23b80399 100644
--- a/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/tableview/impl/MetadataStoreTableViewImpl.java
+++ b/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/tableview/impl/MetadataStoreTableViewImpl.java
@@ -111,6 +111,7 @@ public class MetadataStoreTableViewImpl<T> implements MetadataStoreTableView<T>
MetadataCacheConfig.<T>builder()
.expireAfterWriteMillis(-1)
.refreshAfterWriteMillis(CACHE_REFRESH_FREQUENCY_IN_MILLIS)
+ .retryBackoff(MetadataCacheConfig.NO_RETRY_BACKOFF_BUILDER)
.asyncReloadConsumer(this::consumeAsyncReload)
.build());
store.registerListener(this::handleNotification);
@@ -306,7 +307,15 @@ public class MetadataStoreTableViewImpl<T> implements MetadataStoreTableView<T>
throw new ConflictException(
String.format("Failed to update from old:%s to value:%s", old, value));
}
- }).thenCompose(__ -> doHandleNotification(path)); // immediately notify local tableview
+ }).thenCompose(__ -> doHandleNotification(path)) // immediately notify local tableview
+ .exceptionally(e -> {
+ if (e.getCause() instanceof MetadataStoreException.BadVersionException) {
+ throw FutureUtil.wrapToCompletionException(new ConflictException(
+ String.format("Failed to update to value:%s", value)));
+ }
+
+ throw FutureUtil.wrapToCompletionException(e.getCause());
+ });
}
public CompletableFuture<Void> delete(String key) {
diff --git a/pulsar-metadata/src/test/java/org/apache/pulsar/metadata/MetadataCacheTest.java b/pulsar-metadata/src/test/java/org/apache/pulsar/metadata/MetadataCacheTest.java
index d3f2995a845..96f7b7204ea 100644
--- a/pulsar-metadata/src/test/java/org/apache/pulsar/metadata/MetadataCacheTest.java
+++ b/pulsar-metadata/src/test/java/org/apache/pulsar/metadata/MetadataCacheTest.java
@@ -702,4 +702,41 @@ public class MetadataCacheTest extends BaseMetadataStoreTest {
assertEquals(backoff.getMax(), 3000);
assertEquals(backoff.getMandatoryStop(), 30_000);
}
+
+ @Test
+ public void testNoBackoffMetadataCacheConfig() {
+ final var config = MetadataCacheConfig.builder().retryBackoff(
+ MetadataCacheConfig.NO_RETRY_BACKOFF_BUILDER).build();
+
+ final var backoff = config.getRetryBackoff().create();
+
+ assertEquals(backoff.getInitial(), 0);
+ assertEquals(backoff.getMax(),0);
+ assertEquals(backoff.getMandatoryStop(), 0);
+ assertTrue(backoff.isMandatoryStopMade());
+ assertEquals(backoff.getFirstBackoffTimeInMillis(), 0);
+ assertEquals(backoff.next(), 0);
+ assertEquals(backoff.next(), 0);
+ assertEquals(backoff.next(), 0);
+ assertTrue(backoff.isMandatoryStopMade());
+ assertEquals(backoff.getFirstBackoffTimeInMillis(), 0);
+
+ backoff.reduceToHalf();
+ assertTrue(backoff.isMandatoryStopMade());
+ assertEquals(backoff.getFirstBackoffTimeInMillis(), 0);
+ assertEquals(backoff.next(), 0);
+ assertEquals(backoff.next(), 0);
+ assertEquals(backoff.next(), 0);
+ assertTrue(backoff.isMandatoryStopMade());
+ assertEquals(backoff.getFirstBackoffTimeInMillis(), 0);
+
+ backoff.reset();
+ assertTrue(backoff.isMandatoryStopMade());
+ assertEquals(backoff.getFirstBackoffTimeInMillis(), 0);
+ assertEquals(backoff.next(), 0);
+ assertEquals(backoff.next(), 0);
+ assertEquals(backoff.next(), 0);
+ assertTrue(backoff.isMandatoryStopMade());
+ assertEquals(backoff.getFirstBackoffTimeInMillis(), 0);
+ }
}