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 2026/10/16 22:31:41 UTC
(pulsar) 18/24: [fix][broker] Prevent bundle split hangs on executor rejection (#26905)
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-4.0-backports-2026-10-17
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 94e33637c74dc7285e37d275c7fc4a1e542fe085
Author: sinan liu <li...@gmail.com>
AuthorDate: Fri Oct 16 06:56:54 2026 +0800
[fix][broker] Prevent bundle split hangs on executor rejection (#26905)
(cherry picked from commit 588a4253b290319ccf142a1517825473f109d8f6)
---
.../pulsar/broker/namespace/NamespaceService.java | 45 ++++--
.../NamespaceServiceSplitFailureTest.java | 178 +++++++++++++++++++++
2 files changed, 208 insertions(+), 15 deletions(-)
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/namespace/NamespaceService.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/namespace/NamespaceService.java
index caef15d803c7..d3cd40b4a2c1 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/namespace/NamespaceService.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/namespace/NamespaceService.java
@@ -45,6 +45,7 @@ import java.util.Set;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
@@ -1059,8 +1060,7 @@ public class NamespaceService implements AutoCloseable {
+ "NamespaceBundle: %s due to %s",
nsname.toString(), bundle.getBundleRange(), ex1.getMessage());
LOG.warn(msg);
- updateFuture.completeExceptionally(
- new ServiceUnitNotReadyException(msg, ex1.getCause()));
+ updateFuture.completeExceptionally(ex1);
return null;
});
} catch (Exception e) {
@@ -1068,7 +1068,7 @@ public class NamespaceService implements AutoCloseable {
"failed to acquire ownership of split bundle for namespace [%s], %s",
nsname.toString(), e.getMessage());
LOG.warn(msg, e);
- updateFuture.completeExceptionally(new ServiceUnitNotReadyException(msg, e));
+ updateFuture.completeExceptionally(e);
}
}).exceptionally(splitException -> {
updateFuture.completeExceptionally(splitException);
@@ -1083,25 +1083,34 @@ public class NamespaceService implements AutoCloseable {
// If success updateNamespaceBundles, then do invalidateBundleCache and unload.
// Else retry splitAndOwnBundleOnceAndRetry.
- updateFuture.whenCompleteAsync((r, t)-> {
+ // Consume the update failure when a retry is scheduled. A whenComplete stage would retain
+ // that failure and incorrectly fail completionFuture in the dispatch-failure handler below.
+ updateFuture.handleAsync((r, t) -> {
if (t != null) {
+ Throwable cause = FutureUtil.unwrapCompletionException(t);
// retry several times on BadVersion
- if ((t.getCause() instanceof MetadataStoreException.BadVersionException)
+ if ((cause instanceof MetadataStoreException.BadVersionException)
&& (counter.decrementAndGet() >= 0)) {
- pulsar.getExecutor().schedule(() -> pulsar.getOrderedExecutor()
- .execute(() -> splitAndOwnBundleOnceAndRetry(
- bundle, unload, counter, completionFuture, splitAlgorithm, boundaries)),
- 100, MILLISECONDS);
- } else if (t instanceof IllegalArgumentException) {
- completionFuture.completeExceptionally(t);
+ pulsar.getExecutor().schedule(() -> {
+ try {
+ pulsar.getOrderedExecutor().execute(() -> splitAndOwnBundleOnceAndRetry(
+ bundle, unload, counter, completionFuture, splitAlgorithm, boundaries));
+ } catch (RejectedExecutionException retryException) {
+ completionFuture.completeExceptionally(new ServiceUnitNotReadyException(
+ "Failed to dispatch namespace bundle split retry for " + bundle,
+ retryException));
+ }
+ }, 100, MILLISECONDS);
+ } else if (cause instanceof IllegalArgumentException) {
+ completionFuture.completeExceptionally(cause);
} else {
// Retry enough, or meet other exception
String msg2 = format(" %s not success update nsBundles, counter %d, reason %s",
- bundle.toString(), counter.get(), t.getMessage());
+ bundle.toString(), counter.get(), cause.getMessage());
LOG.warn(msg2);
- completionFuture.completeExceptionally(new ServiceUnitNotReadyException(msg2));
+ completionFuture.completeExceptionally(new ServiceUnitNotReadyException(msg2, cause));
}
- return;
+ return null;
}
// success updateNamespaceBundles
@@ -1159,7 +1168,13 @@ public class NamespaceService implements AutoCloseable {
completionFuture.completeExceptionally(new ServiceUnitNotReadyException(msg1));
return null;
});
- }, pulsar.getOrderedExecutor());
+ return null;
+ }, pulsar.getOrderedExecutor()).exceptionally(processException -> {
+ completionFuture.completeExceptionally(new ServiceUnitNotReadyException(
+ "Failed to process namespace bundle split completion for " + bundle,
+ FutureUtil.unwrapCompletionException(processException)));
+ return null;
+ });
});
}
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/namespace/NamespaceServiceSplitFailureTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/namespace/NamespaceServiceSplitFailureTest.java
index aee9fbe84f94..ea8f27963cfc 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/namespace/NamespaceServiceSplitFailureTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/namespace/NamespaceServiceSplitFailureTest.java
@@ -20,16 +20,25 @@ package org.apache.pulsar.broker.namespace;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.mockito.Mockito.doCallRealMethod;
import static org.mockito.Mockito.doReturn;
import static org.mockito.Mockito.spy;
+import java.util.ArrayList;
+import java.util.List;
import java.util.Optional;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
+import java.util.concurrent.RejectedExecutionException;
+import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
import lombok.Cleanup;
import org.apache.bookkeeper.common.util.OrderedExecutor;
import org.apache.pulsar.broker.PulsarService;
+import org.apache.pulsar.broker.ServiceConfiguration;
import org.apache.pulsar.broker.loadbalance.extensions.ExtensibleLoadManagerImpl;
+import org.apache.pulsar.broker.loadbalance.impl.ModularLoadManagerImpl;
import org.apache.pulsar.broker.resources.LocalPoliciesResources;
import org.apache.pulsar.broker.resources.PulsarResources;
import org.apache.pulsar.broker.service.BrokerServiceException.ServiceUnitNotReadyException;
@@ -38,10 +47,15 @@ import org.apache.pulsar.common.naming.NamespaceBundle;
import org.apache.pulsar.common.naming.NamespaceBundleFactory;
import org.apache.pulsar.common.naming.NamespaceBundleSplitAlgorithm;
import org.apache.pulsar.common.naming.NamespaceName;
+import org.apache.pulsar.common.policies.data.BundlesData;
import org.apache.pulsar.common.policies.data.LocalPolicies;
import org.apache.pulsar.metadata.api.CacheGetResult;
+import org.apache.pulsar.metadata.api.MetadataStoreConfig;
import org.apache.pulsar.metadata.api.MetadataStoreException;
import org.apache.pulsar.metadata.coordination.impl.CoordinationServiceImpl;
+import org.apache.pulsar.metadata.impl.FaultInjectionMetadataStore;
+import org.apache.pulsar.metadata.impl.FaultInjectionMetadataStore.OperationType;
+import org.apache.pulsar.metadata.impl.MetadataStoreFactoryImpl;
import org.testng.annotations.DataProvider;
import org.testng.annotations.Test;
@@ -94,4 +108,168 @@ public class NamespaceServiceSplitFailureTest {
.hasMessageContaining(failure.getMessage());
assertThat(result).isCompletedExceptionally();
}
+
+ @Test
+ public void testSplitPreservesInvalidBoundaryFailure() throws Exception {
+ @Cleanup
+ SplitContext context = new SplitContext();
+ // The real specified-positions algorithm accepts duplicates. Constructing the resulting
+ // empty bundle throws inside NamespaceBundleFactory.splitBundles's thenApply stage.
+ CompletableFuture<Void> result = context.namespaceService.splitAndOwnBundle(context.bundle, false,
+ NamespaceBundleSplitAlgorithm.SPECIFIED_POSITIONS_DIVIDE_FORCE_ALGO,
+ new ArrayList<>(List.of(1L, 1L)));
+ assertThatThrownBy(() -> result.get(10, TimeUnit.SECONDS))
+ .isInstanceOf(ExecutionException.class)
+ .hasCauseInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("empty key range");
+ assertThat(context.context.getPulsarService().getPulsarResources().getLocalPolicies()
+ .getLocalPoliciesAsync(context.namespace).get(10, TimeUnit.SECONDS).orElseThrow().bundles
+ .getNumBundles()).isEqualTo(1);
+ }
+
+ @Test
+ public void testSplitCompletesWhenCompletionExecutorRejects() throws Exception {
+ @Cleanup
+ SplitContext context = new SplitContext();
+ context.executor.shutdown();
+ assertSplitRejected(context.split());
+ }
+
+ @Test
+ public void testSplitCompletesWhenRetrySchedulerRejects() throws Exception {
+ @Cleanup
+ SplitContext context = new SplitContext();
+ context.failPolicyUpdate(new MetadataStoreException.BadVersionException("split conflict"));
+ context.scheduler.shutdown();
+ assertSplitRejected(context.split());
+ }
+
+ @Test
+ public void testSplitCompletesWhenRetryExecutorRejects() throws Exception {
+ @Cleanup
+ SplitContext context = new SplitContext();
+ context.failPolicyUpdate(new MetadataStoreException.BadVersionException("split conflict"));
+ // Shut down the real ordered executor only after the real scheduler accepts the retry.
+ context.scheduler.beforeRetry = context.executor::shutdown;
+ assertSplitRejected(context.split());
+ }
+
+ @Test
+ public void testSplitRetriesVersionConflictsUntilLimit() throws Exception {
+ @Cleanup
+ SplitContext context = new SplitContext();
+ MetadataStoreException.BadVersionException failure =
+ new MetadataStoreException.BadVersionException("split conflict");
+ AtomicInteger failedWrites = new AtomicInteger();
+ for (int i = 0; i <= NamespaceService.BUNDLE_SPLIT_RETRY_LIMIT; i++) {
+ context.metadataStore.failConditional(failure, (operation, path) -> {
+ if (operation == OperationType.PUT && path.equals(context.policiesPath())) {
+ failedWrites.incrementAndGet();
+ return true;
+ }
+ return false;
+ });
+ }
+ CompletableFuture<Void> result = context.split();
+ assertThatThrownBy(() -> result.get(10, TimeUnit.SECONDS))
+ .isInstanceOf(ExecutionException.class)
+ .hasCauseInstanceOf(ServiceUnitNotReadyException.class)
+ .hasRootCause(failure);
+ assertThat(failedWrites).hasValue(NamespaceService.BUNDLE_SPLIT_RETRY_LIMIT + 1);
+ }
+
+ @Test
+ public void testSplitSucceedsAfterVersionConflict() throws Exception {
+ @Cleanup
+ SplitContext context = new SplitContext();
+ context.failPolicyUpdate(new MetadataStoreException.BadVersionException("split conflict"));
+ context.split().get(10, TimeUnit.SECONDS);
+ PulsarResources resources = context.context.getPulsarService().getPulsarResources();
+ assertThat(resources.getLocalPolicies().getLocalPoliciesAsync(context.namespace)
+ .get(10, TimeUnit.SECONDS).orElseThrow().bundles.getNumBundles()).isEqualTo(2);
+ assertThat(resources.getNamespaceResources().getPoliciesAsync(context.namespace)
+ .get(10, TimeUnit.SECONDS).orElseThrow().bundles.getNumBundles()).isEqualTo(2);
+ }
+
+ private static void assertSplitRejected(CompletableFuture<Void> result) {
+ assertThatThrownBy(() -> result.get(10, TimeUnit.SECONDS))
+ .isInstanceOf(ExecutionException.class)
+ .hasCauseInstanceOf(ServiceUnitNotReadyException.class)
+ .hasRootCauseInstanceOf(RejectedExecutionException.class);
+ }
+
+ private static class RetryScheduler extends ScheduledThreadPoolExecutor {
+ private volatile Runnable beforeRetry;
+
+ RetryScheduler() {
+ super(1);
+ }
+
+ @Override
+ public ScheduledFuture<?> schedule(Runnable command, long delay, TimeUnit unit) {
+ return super.schedule(() -> {
+ if (unit.toMillis(delay) == 100 && beforeRetry != null) {
+ beforeRetry.run();
+ }
+ command.run();
+ }, delay, unit);
+ }
+ }
+
+ private static class SplitContext implements AutoCloseable {
+ private final FaultInjectionMetadataStore metadataStore;
+ private final PulsarTestContext context;
+ private final OrderedExecutor executor;
+ private final RetryScheduler scheduler;
+ private final NamespaceService namespaceService;
+ private final NamespaceName namespace = NamespaceName.get("prop/ns-split-failure");
+ private final NamespaceBundle bundle;
+
+ SplitContext() throws Exception {
+ metadataStore = new FaultInjectionMetadataStore(MetadataStoreFactoryImpl.createExtended("memory:local",
+ MetadataStoreConfig.builder().build()));
+ ServiceConfiguration configuration = new ServiceConfiguration();
+ configuration.setLoadManagerClassName(ModularLoadManagerImpl.class.getName());
+ context = PulsarTestContext.builder().spyByDefault().config(configuration)
+ .localMetadataStore(metadataStore).configurationMetadataStore(metadataStore).build();
+ PulsarService pulsar = context.getPulsarService();
+ executor = OrderedExecutor.newBuilder().numThreads(1).name("split-failure-test").build();
+ scheduler = new RetryScheduler();
+ doReturn(executor).when(pulsar).getOrderedExecutor();
+ doReturn(scheduler).when(pulsar).getExecutor();
+ assertThat(ExtensibleLoadManagerImpl.isLoadManagerExtensionEnabled(pulsar)).isFalse();
+ namespaceService = pulsar.getNamespaceService();
+ bundle = namespaceService.getNamespaceBundleFactory()
+ .getBundle(namespace.toString(), "0x00000000_0xffffffff");
+ pulsar.getPulsarResources().getLocalPolicies().createLocalPoliciesAsync(namespace,
+ new LocalPolicies(BundlesData.builder()
+ .boundaries(List.of("0x00000000", "0xffffffff")).numBundles(1).build(), null, null))
+ .get(10, TimeUnit.SECONDS);
+ }
+
+ CompletableFuture<Void> split() {
+ return namespaceService.splitAndOwnBundle(bundle, false,
+ NamespaceBundleSplitAlgorithm.RANGE_EQUALLY_DIVIDE_ALGO, null);
+ }
+
+ String policiesPath() {
+ return "/admin/local-policies/" + namespace;
+ }
+
+ void failPolicyUpdate(MetadataStoreException failure) {
+ metadataStore.failConditional(failure,
+ (operation, path) -> operation == OperationType.PUT && path.equals(policiesPath()));
+ }
+
+ @Override
+ public void close() throws Exception {
+ PulsarService pulsar = context.getPulsarService();
+ doCallRealMethod().when(pulsar).getOrderedExecutor();
+ doCallRealMethod().when(pulsar).getExecutor();
+ executor.shutdown();
+ scheduler.shutdownNow();
+ context.close();
+ metadataStore.close();
+ }
+ }
}