customMetadata = new HashMap<>();
+ customMetadata.put(EnsemblePlacementPolicyConfig.ENSEMBLE_PLACEMENT_POLICY_CONFIG, policyConfig.encode());
+
+ BookieId bookie1Id = new BookieSocketAddress(BOOKIE1).toBookieId();
+ BookieId bookie2Id = new BookieSocketAddress(BOOKIE2).toBookieId();
+
+ // Must not throw NullPointerException; BKNotEnoughBookiesException is acceptable.
+ isolationPolicy.replaceBookie(2, 2, 2, customMetadata,
+ Arrays.asList(bookie1Id, bookie2Id), bookie2Id, null);
+ }
+
+ /**
+ * Verifies that {@link IsolatedBookieEnsemblePlacementPolicy#getIsolationGroup} treats
+ * {@link ZkIsolatedBookieEnsemblePlacementPolicy} (a subclass) exactly like
+ * {@link IsolatedBookieEnsemblePlacementPolicy} itself when reading isolation groups from
+ * {@link EnsemblePlacementPolicyConfig} properties.
+ *
+ * Legacy Pulsar clusters may have persisted {@code EnsemblePlacementPolicyConfig} entries whose
+ * {@code policyClass} field is set to {@code ZkIsolatedBookieEnsemblePlacementPolicy}. The
+ * {@code isAssignableFrom} check in {@code getIsolationGroup} must recognise this subclass so that
+ * the isolation groups are read from the stored properties rather than falling back to the
+ * policy-level defaults.
+ */
+ @Test
+ public void testGetIsolationGroupWithZkCompatiblePolicyClass() throws Exception {
+ // Group1 → default isolation group configured on the policy.
+ // Group2 → isolation group carried inside the custom metadata (ZkIsolated class).
+ final String defaultGroup = "Group1";
+ final String customGroup = "Group2";
+
+ Map> bookieMapping = new HashMap<>();
+ Map group1 = new HashMap<>();
+ group1.put(BOOKIE1, BookieInfo.builder().rack("rack0").build());
+ group1.put(BOOKIE2, BookieInfo.builder().rack("rack0").build());
+ Map group2 = new HashMap<>();
+ group2.put(BOOKIE3, BookieInfo.builder().rack("rack1").build());
+ group2.put(BOOKIE4, BookieInfo.builder().rack("rack1").build());
+ bookieMapping.put(defaultGroup, group1);
+ bookieMapping.put(customGroup, group2);
+
+ store.put(BookieRackAffinityMapping.BOOKIE_INFO_ROOT_PATH, jsonMapper.writeValueAsBytes(bookieMapping),
+ Optional.empty()).join();
+
+ IsolatedBookieEnsemblePlacementPolicy isolationPolicy = new IsolatedBookieEnsemblePlacementPolicy();
+ ClientConfiguration bkClientConf = new ClientConfiguration();
+ bkClientConf.setProperty(BookieRackAffinityMapping.METADATA_STORE_INSTANCE, store);
+ bkClientConf.setProperty(IsolatedBookieEnsemblePlacementPolicy.ISOLATION_BOOKIE_GROUPS, defaultGroup);
+ isolationPolicy.initialize(bkClientConf, Optional.empty(), timer, SettableFeatureProvider.DISABLE_ALL,
+ NullStatsLogger.INSTANCE, BookieSocketAddress.LEGACY_BOOKIEID_RESOLVER);
+ isolationPolicy.onClusterChanged(writableBookies, readOnlyBookies);
+
+ // --- unit-level: getIsolationGroup should parse properties, not fall back to defaults ---
+ Map props = new HashMap<>();
+ props.put(IsolatedBookieEnsemblePlacementPolicy.ISOLATION_BOOKIE_GROUPS, customGroup);
+ props.put(IsolatedBookieEnsemblePlacementPolicy.SECONDARY_ISOLATION_BOOKIE_GROUPS, "secondaryGroup");
+ EnsemblePlacementPolicyConfig zkConfig = new EnsemblePlacementPolicyConfig(
+ ZkIsolatedBookieEnsemblePlacementPolicy.class, props);
+
+ Pair, Set> groups = isolationPolicy.getIsolationGroup(zkConfig);
+ assertEquals(groups.getLeft(), Sets.newHashSet(customGroup),
+ "primary group must be read from ZkIsolated config properties");
+ assertEquals(groups.getRight(), Sets.newHashSet("secondaryGroup"),
+ "secondary group must be read from ZkIsolated config properties");
+
+ // --- integration-level: newEnsemble must select bookies from the ZkIsolated config group ---
+ Map placementPolicyProperties = new HashMap<>();
+ placementPolicyProperties.put(IsolatedBookieEnsemblePlacementPolicy.ISOLATION_BOOKIE_GROUPS, customGroup);
+ placementPolicyProperties.put(IsolatedBookieEnsemblePlacementPolicy.SECONDARY_ISOLATION_BOOKIE_GROUPS, "");
+ EnsemblePlacementPolicyConfig policyConfig = new EnsemblePlacementPolicyConfig(
+ ZkIsolatedBookieEnsemblePlacementPolicy.class, placementPolicyProperties);
+ Map customMetadata = new HashMap<>();
+ customMetadata.put(EnsemblePlacementPolicyConfig.ENSEMBLE_PLACEMENT_POLICY_CONFIG, policyConfig.encode());
+
+ Set bookieIdGroup2 = new HashSet<>();
+ bookieIdGroup2.add(new BookieSocketAddress(BOOKIE3).toBookieId());
+ bookieIdGroup2.add(new BookieSocketAddress(BOOKIE4).toBookieId());
+
+ List ensemble = isolationPolicy
+ .newEnsemble(2, 2, 2, customMetadata, new HashSet<>()).getResult();
+ assertTrue(bookieIdGroup2.containsAll(ensemble),
+ "ensemble should come from " + customGroup + " (ZkIsolated config), got " + ensemble);
+
+ // Sanity-check: without custom metadata the default group1 bookies are chosen.
+ Set bookieIdGroup1 = new HashSet<>();
+ bookieIdGroup1.add(new BookieSocketAddress(BOOKIE1).toBookieId());
+ bookieIdGroup1.add(new BookieSocketAddress(BOOKIE2).toBookieId());
+ List defaultEnsemble = isolationPolicy
+ .newEnsemble(2, 2, 2, Collections.emptyMap(), new HashSet<>()).getResult();
+ assertTrue(bookieIdGroup1.containsAll(defaultEnsemble),
+ "default ensemble should come from " + defaultGroup + ", got " + defaultEnsemble);
+ }
+
// The policy gets the bookie info asynchronously before each query or update, when putting the bookie info into
// the metadata store, the cache needs some time to receive the notification and update accordingly.
private void updateBookieInfo(IsolatedBookieEnsemblePlacementPolicy isolationPolicy, byte[] bookieInfo) {
diff --git a/pulsar-broker-common/src/test/java/org/apache/pulsar/broker/resources/MetadataStoreCacheLoaderTest.java b/pulsar-broker-common/src/test/java/org/apache/pulsar/broker/resources/MetadataStoreCacheLoaderTest.java
new file mode 100644
index 0000000000000..e8d4d28153c94
--- /dev/null
+++ b/pulsar-broker-common/src/test/java/org/apache/pulsar/broker/resources/MetadataStoreCacheLoaderTest.java
@@ -0,0 +1,76 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.broker.resources;
+
+import static org.apache.pulsar.broker.resources.MetadataStoreCacheLoader.LOADBALANCE_BROKERS_ROOT;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertTrue;
+import java.util.List;
+import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
+import lombok.Cleanup;
+import org.apache.pulsar.metadata.api.MetadataStore;
+import org.apache.pulsar.policies.data.loadbalancer.LoadManagerReport;
+import org.testng.annotations.Test;
+
+public class MetadataStoreCacheLoaderTest {
+
+ private MetadataStoreCacheLoader newCacheLoader(LoadManagerReportResources loadReportResources) throws Exception {
+ MetadataStore store = mock(MetadataStore.class);
+ PulsarResources pulsarResources = mock(PulsarResources.class);
+ when(pulsarResources.getLoadReportResources()).thenReturn(loadReportResources);
+ when(loadReportResources.getStore()).thenReturn(store);
+ return new MetadataStoreCacheLoader(pulsarResources, 5000);
+ }
+
+ @Test
+ public void testGetAvailableBrokersServesCacheWithoutBlocking() throws Exception {
+ LoadManagerReportResources loadReportResources = mock(LoadManagerReportResources.class);
+ LoadManagerReport report = mock(LoadManagerReport.class);
+ when(loadReportResources.getChildrenAsync(LOADBALANCE_BROKERS_ROOT))
+ .thenReturn(CompletableFuture.completedFuture(List.of("broker-1")));
+ when(loadReportResources.getAsync(LOADBALANCE_BROKERS_ROOT + "/broker-1"))
+ .thenReturn(CompletableFuture.completedFuture(Optional.of(report)));
+
+ @Cleanup
+ MetadataStoreCacheLoader loader = newCacheLoader(loadReportResources);
+
+ assertEquals(loader.getAvailableBrokers(), List.of(report));
+ // The blocking, synchronous getChildren(...) must never be used: it could stall a Netty IO thread.
+ verify(loadReportResources, never()).getChildren(anyString());
+ }
+
+ @Test
+ public void testGetAvailableBrokersDoesNotBlockOnEmptyCache() throws Exception {
+ LoadManagerReportResources loadReportResources = mock(LoadManagerReportResources.class);
+ when(loadReportResources.getChildrenAsync(LOADBALANCE_BROKERS_ROOT))
+ .thenReturn(CompletableFuture.completedFuture(List.of()));
+
+ @Cleanup
+ MetadataStoreCacheLoader loader = newCacheLoader(loadReportResources);
+
+ assertTrue(loader.getAvailableBrokers().isEmpty());
+ verify(loadReportResources, never()).getChildren(anyString());
+ }
+}
diff --git a/pulsar-broker/pom.xml b/pulsar-broker/pom.xml
index ec4e10e53fc98..17997528278d6 100644
--- a/pulsar-broker/pom.xml
+++ b/pulsar-broker/pom.xml
@@ -25,7 +25,7 @@
org.apache.pulsar
pulsar
- 4.2.0-SNAPSHOT
+ 4.2.4-SNAPSHOT
pulsar-broker
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/PulsarStandaloneStarter.java b/pulsar-broker/src/main/java/org/apache/pulsar/PulsarStandaloneStarter.java
index 411449d1042cb..29feac8cb46eb 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/PulsarStandaloneStarter.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/PulsarStandaloneStarter.java
@@ -29,7 +29,6 @@
import org.apache.commons.lang3.StringUtils;
import org.apache.logging.log4j.LogManager;
import org.apache.pulsar.broker.ServiceConfiguration;
-import org.apache.pulsar.broker.ServiceConfigurationUtils;
import org.apache.pulsar.common.configuration.PulsarConfigurationLoader;
import org.apache.pulsar.docs.tools.CmdGenerateDocs;
import picocli.CommandLine;
@@ -92,12 +91,8 @@ public PulsarStandaloneStarter(String[] args) throws Exception {
// Use advertised address from command line
config.setAdvertisedAddress(this.getAdvertisedAddress());
} else if (isBlank(config.getAdvertisedAddress()) && isBlank(config.getAdvertisedListeners())) {
- try {
- config.setAdvertisedAddress(ServiceConfigurationUtils.unsafeLocalhostResolve());
- } catch (Exception e) {
- log.warn("Failed to resolve FQDN, using 'localhost' as advertised address", e);
- config.setAdvertisedAddress("localhost");
- }
+ // Use advertised address as local hostname
+ config.setAdvertisedAddress("localhost");
} else {
// Use advertised or advertisedListeners address from config file
}
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/PulsarService.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/PulsarService.java
index 5670579372590..bf1f0e24848b1 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/PulsarService.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/PulsarService.java
@@ -106,6 +106,7 @@
import org.apache.pulsar.broker.rest.Topics;
import org.apache.pulsar.broker.service.BrokerService;
import org.apache.pulsar.broker.service.HealthChecker;
+import org.apache.pulsar.broker.service.LegacyAwareTopicPoliciesService;
import org.apache.pulsar.broker.service.PulsarMetadataEventSynchronizer;
import org.apache.pulsar.broker.service.SystemTopicBasedTopicPoliciesService;
import org.apache.pulsar.broker.service.Topic;
@@ -756,8 +757,13 @@ public CompletableFuture closeAsync(boolean waitForWebServiceToStop) {
} else {
LOG.warn("Closed with errors", t);
}
- state = State.Closed;
- isClosedCondition.signalAll();
+ mutex.lock();
+ try {
+ state = State.Closed;
+ isClosedCondition.signalAll();
+ } finally {
+ mutex.unlock();
+ }
return null;
});
return closeFuture;
@@ -2251,8 +2257,15 @@ private TopicPoliciesService initTopicPoliciesService() throws Exception {
return TopicPoliciesService.DISABLED;
}
}
- return (TopicPoliciesService) Reflections.createInstance(className,
+ final var configuredService = (TopicPoliciesService) Reflections.createInstance(className,
Thread.currentThread().getContextClassLoader());
+ if (!config.isSystemTopicEnabled()) {
+ LOG.info("[{}] System topic is disabled, using configured topic policies service without legacy routing",
+ className);
+ return configuredService;
+ }
+ return new LegacyAwareTopicPoliciesService(this, new SystemTopicBasedTopicPoliciesService(this),
+ configuredService);
}
/**
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/AdminResource.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/AdminResource.java
index 190ed01114329..5ac94210c965e 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/AdminResource.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/AdminResource.java
@@ -18,7 +18,6 @@
*/
package org.apache.pulsar.broker.admin;
-import static org.apache.commons.lang3.StringUtils.isBlank;
import com.fasterxml.jackson.databind.ObjectReader;
import com.fasterxml.jackson.databind.ObjectWriter;
import com.google.errorprone.annotations.CanIgnoreReturnValue;
@@ -820,12 +819,6 @@ protected void checkNotNull(Object o, String errorMessage) {
}
}
- protected void checkNotBlank(String str, String errorMessage) {
- if (isBlank(str)) {
- throw new RestException(Status.PRECONDITION_FAILED, errorMessage);
- }
- }
-
protected boolean isManagedLedgerNotFoundException(Throwable cause) {
return cause instanceof ManagedLedgerException.MetadataNotFoundException
|| cause instanceof MetadataStoreException.NotFoundException;
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/BrokersBase.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/BrokersBase.java
index 47b78bda80695..d165335719c00 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/BrokersBase.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/BrokersBase.java
@@ -128,8 +128,11 @@ public void getActiveBrokers(@Suspended final AsyncResponse asyncResponse) throw
public void getLeaderBroker(@Suspended final AsyncResponse asyncResponse) {
validateBothSuperuserAndBrokerOperation(pulsar().getConfig().getClusterName(),
pulsar().getBrokerId(), BrokerOperation.GET_LEADER_BROKER)
- .thenAccept(__ -> {
- LeaderBroker leaderBroker = pulsar().getLeaderElectionService().getCurrentLeader()
+ // The authoritative read: waits for an in-progress leader election to settle
+ // instead of returning 404 while a re-election is still in flight.
+ .thenCompose(__ -> pulsar().getLeaderElectionService().readCurrentLeader())
+ .thenAccept(leader -> {
+ LeaderBroker leaderBroker = leader
.orElseThrow(() -> new RestException(Status.NOT_FOUND, "Couldn't find leader broker"));
BrokerInfo brokerInfo = BrokerInfo.builder()
.serviceUrl(leaderBroker.getServiceUrl())
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/NamespacesBase.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/NamespacesBase.java
index 0f626eb738581..044dbdaeb68e1 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/NamespacesBase.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/NamespacesBase.java
@@ -58,7 +58,6 @@
import org.apache.pulsar.broker.PulsarServerException;
import org.apache.pulsar.broker.ServiceConfiguration;
import org.apache.pulsar.broker.admin.AdminResource;
-import org.apache.pulsar.broker.loadbalance.LeaderBroker;
import org.apache.pulsar.broker.loadbalance.extensions.ExtensibleLoadManagerImpl;
import org.apache.pulsar.broker.service.BrokerServiceException;
import org.apache.pulsar.broker.service.BrokerServiceException.SubscriptionBusyException;
@@ -79,7 +78,6 @@
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.NamespaceBundles;
import org.apache.pulsar.common.naming.NamespaceName;
import org.apache.pulsar.common.naming.SystemTopicNames;
import org.apache.pulsar.common.naming.TopicName;
@@ -481,7 +479,7 @@ private CompletableFuture precheckWhenDeleteNamespace(NamespaceName ns
// There are still more than one clusters configured for the global namespace
throw new RestException(Status.PRECONDITION_FAILED,
"Cannot delete the global namespace " + nsName + ". There are still more than "
- + "one replication clusters configured.");
+ + "one replication clusters configured or replication clusters is empty.");
}
if (!cluster.equals(config().getClusterName())) {
// the only replication cluster is other cluster, redirect
@@ -597,8 +595,7 @@ protected CompletableFuture internalDeleteNamespaceBundleAsync(String bund
}
return future
.thenCompose(__ ->
- validateNamespaceBundleOwnershipAsync(namespaceName, policies.bundles,
- bundleRange,
+ validateNamespaceBundleOwnershipAsync(namespaceName, bundleRange,
authoritative, true))
.thenCompose(bundle -> {
return pulsar().getNamespaceService().getListOfPersistentTopics(namespaceName)
@@ -1354,48 +1351,51 @@ private CompletableFuture validateLeaderBrokerAsync() {
if (this.isLeaderBroker()) {
return CompletableFuture.completedFuture(null);
}
- Optional currentLeaderOpt = pulsar().getLeaderElectionService().getCurrentLeader();
- if (currentLeaderOpt.isEmpty()) {
- String errorStr = "The current leader is empty.";
- log.error(errorStr);
- return FutureUtil.failedFuture(new RestException(Response.Status.PRECONDITION_FAILED, errorStr));
- }
- LeaderBroker leaderBroker = pulsar().getLeaderElectionService().getCurrentLeader().get();
- String leaderBrokerId = leaderBroker.getBrokerId();
- return pulsar().getNamespaceService()
- .createLookupResult(leaderBrokerId, false, null)
- .thenCompose(lookupResult -> {
- String redirectUrl = isRequestHttps() ? lookupResult.getLookupData().getHttpUrlTls()
- : lookupResult.getLookupData().getHttpUrl();
- if (redirectUrl == null) {
- log.error("Redirected broker's service url is not configured");
- return FutureUtil.failedFuture(new RestException(Response.Status.PRECONDITION_FAILED,
- "Redirected broker's service url is not configured."));
- }
+ // The authoritative read: waits for an in-progress leader election to settle instead of
+ // failing the request while a re-election is still in flight.
+ return pulsar().getLeaderElectionService().readCurrentLeader().thenCompose(currentLeaderOpt -> {
+ if (currentLeaderOpt.isEmpty()) {
+ String errorStr = "The current leader is empty.";
+ log.error(errorStr);
+ return FutureUtil.failedFuture(new RestException(Response.Status.PRECONDITION_FAILED, errorStr));
+ }
+ String leaderBrokerId = currentLeaderOpt.get().getBrokerId();
+ return pulsar().getNamespaceService()
+ .createLookupResult(leaderBrokerId, false, null)
+ .thenCompose(lookupResult -> {
+ String redirectUrl = isRequestHttps() ? lookupResult.getLookupData().getHttpUrlTls()
+ : lookupResult.getLookupData().getHttpUrl();
+ if (redirectUrl == null) {
+ log.error("Redirected broker's service url is not configured");
+ return FutureUtil.failedFuture(new RestException(Response.Status.PRECONDITION_FAILED,
+ "Redirected broker's service url is not configured."));
+ }
- try {
- URL url = new URL(redirectUrl);
- URI redirect = UriBuilder.fromUri(uri.getRequestUri()).host(url.getHost())
- .port(url.getPort())
- .replaceQueryParam("authoritative",
- false).build();
- // Redirect
- if (log.isDebugEnabled()) {
- log.debug("Redirecting the request call to leader - {}", redirect);
+ try {
+ URL url = new URL(redirectUrl);
+ URI redirect = UriBuilder.fromUri(uri.getRequestUri()).host(url.getHost())
+ .port(url.getPort())
+ .replaceQueryParam("authoritative",
+ false).build();
+ // Redirect
+ if (log.isDebugEnabled()) {
+ log.debug("Redirecting the request call to leader - {}", redirect);
+ }
+ return FutureUtil.failedFuture((
+ new WebApplicationException(Response.temporaryRedirect(redirect).build())));
+ } catch (MalformedURLException exception) {
+ log.error("The redirect url is malformed - {}", redirectUrl);
+ return FutureUtil.failedFuture(new RestException(exception));
}
- return FutureUtil.failedFuture((
- new WebApplicationException(Response.temporaryRedirect(redirect).build())));
- } catch (MalformedURLException exception) {
- log.error("The redirect url is malformed - {}", redirectUrl);
- return FutureUtil.failedFuture(new RestException(exception));
- }
- });
+ });
+ });
}
public CompletableFuture setNamespaceBundleAffinityAsync(String bundleRange, String destinationBroker) {
if (StringUtils.isBlank(destinationBroker)) {
return CompletableFuture.completedFuture(null);
}
+
return pulsar().getLoadManager().get().getAvailableBrokersAsync()
.thenCompose(brokers -> {
if (!brokers.contains(destinationBroker)) {
@@ -1416,8 +1416,11 @@ public CompletableFuture setNamespaceBundleAffinityAsync(String bundleRang
if (ExtensibleLoadManagerImpl.isLoadManagerExtensionEnabled(pulsar())) {
return;
}
+ final String bundleName = pulsar().getNamespaceService().getNamespaceBundleFactory()
+ .getBundle(namespaceName.toString(), bundleRange)
+ .toString();
// For ExtensibleLoadManager, this operation will be ignored.
- pulsar().getLoadManager().get().setNamespaceBundleAffinity(bundleRange, destinationBroker);
+ pulsar().getLoadManager().get().setNamespaceBundleAffinity(bundleName, destinationBroker);
});
}
@@ -1456,9 +1459,8 @@ public CompletableFuture internalUnloadNamespaceBundleAsync(String bundleR
}
})
.thenCompose(__ -> validatePoliciesReadOnlyAccessAsync())
- .thenCompose(__ -> getNamespacePoliciesAsync(namespaceName))
- .thenCompose(policies ->
- isBundleOwnedByAnyBroker(namespaceName, policies.bundles, bundleRange)
+ .thenCompose(__ ->
+ isBundleOwnedByAnyBroker(namespaceName, bundleRange)
.thenCompose(flag -> {
if (!flag) {
log.info("[{}] Namespace bundle is not owned by any broker {}/{}", clientAppId(),
@@ -1466,7 +1468,7 @@ public CompletableFuture internalUnloadNamespaceBundleAsync(String bundleR
return CompletableFuture.completedFuture(null);
}
Optional destinationBrokerOpt = Optional.ofNullable(destinationBroker);
- return validateNamespaceBundleOwnershipAsync(namespaceName, policies.bundles, bundleRange,
+ return validateNamespaceBundleOwnershipAsync(namespaceName, bundleRange,
authoritative, true)
.thenCompose(nsBundle -> pulsar().getNamespaceService()
.unloadNamespaceBundle(nsBundle, destinationBrokerOpt));
@@ -1506,10 +1508,8 @@ protected CompletableFuture internalSplitNamespaceBundleAsync(String bundl
.thenCompose(__ -> validatePoliciesReadOnlyAccessAsync())
.thenCompose(__ -> getBundleRangeAsync(bundleName))
.thenCompose(bundleRange -> {
- return getNamespacePoliciesAsync(namespaceName)
- .thenCompose(policies ->
- validateNamespaceBundleOwnershipAsync(namespaceName, policies.bundles, bundleRange,
- authoritative, false))
+ return validateNamespaceBundleOwnershipAsync(namespaceName, bundleRange,
+ authoritative, false)
.thenCompose(nsBundle -> pulsar().getNamespaceService().splitAndOwnBundle(nsBundle, unload,
pulsar().getNamespaceService()
.getNamespaceBundleSplitAlgorithmByName(splitAlgorithmName),
@@ -1523,9 +1523,8 @@ protected CompletableFuture internalGetTopicHashPositionsAsy
log.debug("[{}] Getting hash position for topic list {}, bundle {}", clientAppId(), topics, bundleRange);
}
return validateNamespacePolicyOperationAsync(namespaceName, PolicyName.PERSISTENCE, PolicyOperation.READ)
- .thenCompose(__ -> getNamespacePoliciesAsync(namespaceName))
- .thenCompose(policies -> {
- return validateNamespaceBundleOwnershipAsync(namespaceName, policies.bundles, bundleRange,
+ .thenCompose(__ -> {
+ return validateNamespaceBundleOwnershipAsync(namespaceName, bundleRange,
false, true)
.thenCompose(nsBundle ->
pulsar().getNamespaceService().getOwnedTopicListForNamespaceBundle(nsBundle))
@@ -1824,129 +1823,106 @@ private CompletableFuture doUpdatePersistenceAsync(PersistencePolicies per
);
}
- protected void internalClearNamespaceBacklog(AsyncResponse asyncResponse, boolean authoritative) {
- validateNamespaceOperation(namespaceName, NamespaceOperation.CLEAR_BACKLOG);
-
- final List> futures = new ArrayList<>();
- try {
- NamespaceBundles bundles = pulsar().getNamespaceService().getNamespaceBundleFactory()
- .getBundles(namespaceName);
- for (NamespaceBundle nsBundle : bundles.getBundles()) {
- // check if the bundle is owned by any broker, if not then there is no backlog on this bundle to clear
- if (pulsar().getNamespaceService().checkOwnershipPresent(nsBundle)) {
- futures.add(pulsar().getAdminClient().namespaces()
- .clearNamespaceBundleBacklogAsync(namespaceName.toString(), nsBundle.getBundleRange()));
- }
- }
- } catch (WebApplicationException wae) {
- asyncResponse.resume(wae);
- return;
- } catch (Exception e) {
- asyncResponse.resume(new RestException(e));
- return;
- }
-
- FutureUtil.waitForAll(futures).handle((result, exception) -> {
- if (exception != null) {
- log.warn("[{}] Failed to clear backlog on the bundles for namespace {}: {}", clientAppId(),
- namespaceName, exception.getCause().getMessage());
- if (exception.getCause() instanceof PulsarAdminException) {
- asyncResponse.resume(new RestException((PulsarAdminException) exception.getCause()));
- return null;
- } else {
- asyncResponse.resume(new RestException(exception.getCause()));
- return null;
- }
- }
- log.info("[{}] Successfully cleared backlog on all the bundles for namespace {}", clientAppId(),
- namespaceName);
- asyncResponse.resume(Response.noContent().build());
- return null;
- });
+ protected CompletableFuture internalClearNamespaceBacklogAsync(boolean authoritative) {
+ return validateNamespaceOperationAsync(namespaceName, NamespaceOperation.CLEAR_BACKLOG)
+ .thenCompose(__ -> pulsar().getNamespaceService().getNamespaceBundleFactory()
+ .getBundlesAsync(namespaceName))
+ .thenCompose(bundles -> {
+ final List> futures = new ArrayList<>();
+ for (NamespaceBundle nsBundle : bundles.getBundles()) {
+ try {
+ futures.add(pulsar().getAdminClient().namespaces()
+ .clearNamespaceBundleBacklogAsync(namespaceName.toString(),
+ nsBundle.getBundleRange()));
+ } catch (PulsarServerException e) {
+ return CompletableFuture.failedFuture(e);
+ }
+ }
+ return FutureUtil.waitForAll(futures);
+ }).thenRun(() -> log.info("[{}] Successfully cleared backlog on all the bundles for namespace {}",
+ clientAppId(), namespaceName));
}
@SuppressWarnings("deprecation")
- protected void internalClearNamespaceBundleBacklog(String bundleRange, boolean authoritative) {
- validateNamespaceOperation(namespaceName, NamespaceOperation.CLEAR_BACKLOG);
- checkNotNull(bundleRange, "BundleRange should not be null");
-
- Policies policies = getNamespacePolicies(namespaceName);
-
- // check cluster ownership for a given global namespace: redirect if peer-cluster owns it
- validateGlobalNamespaceOwnership(namespaceName);
-
- validateNamespaceBundleOwnership(namespaceName, policies.bundles, bundleRange, authoritative, true);
+ protected CompletableFuture internalClearNamespaceBundleBacklogAsync(String bundleRange,
+ boolean authoritative) {
+ if (bundleRange == null) {
+ return FutureUtil.failedFuture(new RestException(Status.BAD_REQUEST, "BundleRange should not be null"));
+ }
- clearBacklog(namespaceName, bundleRange, null);
- log.info("[{}] Successfully cleared backlog on namespace bundle {}/{}", clientAppId(), namespaceName,
- bundleRange);
+ return validateNamespaceOperationAsync(namespaceName, NamespaceOperation.CLEAR_BACKLOG)
+ .thenCompose(__ -> {
+ // check cluster ownership for a given global namespace: redirect if peer-cluster owns it
+ return validateGlobalNamespaceOwnershipAsync(namespaceName);
+ })
+ .thenCompose(__ ->
+ // Allow acquiring ownership for an unassigned bundle so backlog can be cleared
+ // even if not loaded.
+ validateNamespaceBundleOwnershipAsync(namespaceName, bundleRange,
+ authoritative, false))
+ .thenCompose(bundle -> clearBacklogAsync(bundle, null))
+ .thenRun(() -> log.info("[{}] Successfully cleared backlog on namespace bundle {}/{}", clientAppId(),
+ namespaceName, bundleRange));
}
- protected void internalClearNamespaceBacklogForSubscription(AsyncResponse asyncResponse, String subscription,
- boolean authoritative) {
- validateNamespaceOperation(namespaceName, NamespaceOperation.CLEAR_BACKLOG);
- checkNotNull(subscription, "Subscription should not be null");
-
- final List> futures = new ArrayList<>();
- try {
- NamespaceBundles bundles = pulsar().getNamespaceService().getNamespaceBundleFactory()
- .getBundles(namespaceName);
- for (NamespaceBundle nsBundle : bundles.getBundles()) {
- // check if the bundle is owned by any broker, if not then there is no backlog on this bundle to clear
- if (pulsar().getNamespaceService().checkOwnershipPresent(nsBundle)) {
- futures.add(pulsar().getAdminClient().namespaces().clearNamespaceBundleBacklogForSubscriptionAsync(
- namespaceName.toString(), nsBundle.getBundleRange(), subscription));
- }
- }
- } catch (WebApplicationException wae) {
- asyncResponse.resume(wae);
- return;
- } catch (Exception e) {
- asyncResponse.resume(new RestException(e));
- return;
+ protected CompletableFuture internalClearNamespaceBacklogForSubscriptionAsync(String subscription,
+ boolean authoritative) {
+ if (subscription == null) {
+ return FutureUtil.failedFuture(new RestException(Status.BAD_REQUEST, "Subscription should not be null"));
}
- FutureUtil.waitForAll(futures).handle((result, exception) -> {
- if (exception != null) {
- log.warn("[{}] Failed to clear backlog for subscription {} on the bundles for namespace {}: {}",
- clientAppId(), subscription, namespaceName, exception.getCause().getMessage());
- if (exception.getCause() instanceof PulsarAdminException) {
- asyncResponse.resume(new RestException((PulsarAdminException) exception.getCause()));
- return null;
- } else {
- asyncResponse.resume(new RestException(exception.getCause()));
- return null;
- }
- }
- log.info("[{}] Successfully cleared backlog for subscription {} on all the bundles for namespace {}",
- clientAppId(), subscription, namespaceName);
- asyncResponse.resume(Response.noContent().build());
- return null;
- });
+ return validateNamespaceOperationAsync(namespaceName, NamespaceOperation.CLEAR_BACKLOG)
+ .thenCompose(__ -> pulsar().getNamespaceService().getNamespaceBundleFactory()
+ .getBundlesAsync(namespaceName))
+ .thenCompose(bundles -> {
+ final List> futures = new ArrayList<>();
+ for (NamespaceBundle nsBundle : bundles.getBundles()) {
+ try {
+ futures.add(pulsar().getAdminClient().namespaces()
+ .clearNamespaceBundleBacklogForSubscriptionAsync(
+ namespaceName.toString(), nsBundle.getBundleRange(), subscription));
+ } catch (PulsarServerException e) {
+ return CompletableFuture.failedFuture(e);
+ }
+ }
+ return FutureUtil.waitForAll(futures);
+ }).thenRun(() -> log.info(
+ "[{}] Successfully cleared backlog for subscription {} on all the bundles for namespace {}",
+ clientAppId(), subscription, namespaceName));
}
@SuppressWarnings("deprecation")
- protected void internalClearNamespaceBundleBacklogForSubscription(String subscription, String bundleRange,
- boolean authoritative) {
- validateNamespaceOperation(namespaceName, NamespaceOperation.CLEAR_BACKLOG);
- checkNotNull(subscription, "Subscription should not be null");
- checkNotNull(bundleRange, "BundleRange should not be null");
-
- Policies policies = getNamespacePolicies(namespaceName);
-
- // check cluster ownership for a given global namespace: redirect if peer-cluster owns it
- validateGlobalNamespaceOwnership(namespaceName);
-
- validateNamespaceBundleOwnership(namespaceName, policies.bundles, bundleRange, authoritative, true);
+ protected CompletableFuture internalClearNamespaceBundleBacklogForSubscriptionAsync(String subscription,
+ String bundleRange,
+ boolean authoritative) {
+ if (subscription == null) {
+ return FutureUtil.failedFuture(new RestException(Status.BAD_REQUEST, "Subscription should not be null"));
+ }
+ if (bundleRange == null) {
+ return FutureUtil.failedFuture(new RestException(Status.BAD_REQUEST, "BundleRange should not be null"));
+ }
- clearBacklog(namespaceName, bundleRange, subscription);
- log.info("[{}] Successfully cleared backlog for subscription {} on namespace bundle {}/{}", clientAppId(),
- subscription, namespaceName, bundleRange);
+ return validateNamespaceOperationAsync(namespaceName, NamespaceOperation.CLEAR_BACKLOG)
+ .thenCompose(__ -> {
+ // check cluster ownership for a given global namespace: redirect if peer-cluster owns it
+ return validateGlobalNamespaceOwnershipAsync(namespaceName);
+ })
+ .thenCompose(__ ->
+ // Allow acquiring ownership for an unassigned bundle so backlog can be cleared
+ // even if not loaded.
+ validateNamespaceBundleOwnershipAsync(namespaceName, bundleRange,
+ authoritative, false))
+ .thenCompose(bundle -> clearBacklogAsync(bundle, subscription))
+ .thenRun(() -> log.info(
+ "[{}] Successfully cleared backlog for subscription {} on namespace bundle {}/{}",
+ clientAppId(), subscription, namespaceName, bundleRange));
}
protected CompletableFuture internalUnsubscribeNamespaceAsync(String subscription,
boolean authoritative) {
- checkNotNull(subscription, "Subscription should not be null");
+ if (subscription == null) {
+ return FutureUtil.failedFuture(new RestException(Status.BAD_REQUEST, "Subscription should not be null"));
+ }
return validateNamespaceOperationAsync(namespaceName, NamespaceOperation.UNSUBSCRIBE)
.thenCompose(__ -> pulsar().getNamespaceService().getNamespaceBundleFactory()
@@ -1969,14 +1945,17 @@ protected CompletableFuture internalUnsubscribeNamespaceAsync(String subsc
@SuppressWarnings("deprecation")
protected CompletableFuture internalUnsubscribeNamespaceBundleAsync(String subscription, String bundleRange,
boolean authoritative) {
- checkNotNull(subscription, "Subscription should not be null");
- checkNotNull(bundleRange, "BundleRange should not be null");
+ if (subscription == null) {
+ return FutureUtil.failedFuture(new RestException(Status.BAD_REQUEST, "Subscription should not be null"));
+ }
+ if (bundleRange == null) {
+ return FutureUtil.failedFuture(new RestException(Status.BAD_REQUEST, "BundleRange should not be null"));
+ }
return validateNamespaceOperationAsync(namespaceName, NamespaceOperation.UNSUBSCRIBE)
.thenCompose(__ -> validateGlobalNamespaceOwnershipAsync(namespaceName))
- .thenCompose(__ -> getNamespacePoliciesAsync(namespaceName))
- .thenCompose(policies ->
- validateNamespaceBundleOwnershipAsync(namespaceName, policies.bundles, bundleRange,
+ .thenCompose(__ ->
+ validateNamespaceBundleOwnershipAsync(namespaceName, bundleRange,
authoritative, false))
.thenCompose(bundle -> unsubscribeAsync(bundle, subscription))
.thenRun(() -> log.info("[{}] Successfully unsubscribed {} on namespace bundle {}/{}",
@@ -2062,8 +2041,14 @@ protected void internalSetDelayedDelivery(DelayedDeliveryPolicies delayedDeliver
}
protected CompletableFuture internalSetNamespaceAntiAffinityGroupAsync(String antiAffinityGroup) {
- checkNotNull(antiAffinityGroup, "Anti-affinity group should not be null");
- checkNotBlank(antiAffinityGroup, "Anti-affinity group can't be empty");
+ if (antiAffinityGroup == null) {
+ return FutureUtil.failedFuture(
+ new RestException(Status.BAD_REQUEST, "Anti-affinity group should not be null"));
+ }
+ if (StringUtils.isBlank(antiAffinityGroup)) {
+ return FutureUtil.failedFuture(
+ new RestException(Status.PRECONDITION_FAILED, "Anti-affinity group can't be empty"));
+ }
return validateNamespacePolicyOperationAsync(namespaceName, PolicyName.ANTI_AFFINITY, PolicyOperation.WRITE)
.thenCompose(__ -> validatePoliciesReadOnlyAccessAsync()).thenCompose(
__ -> getDefaultBundleDataAsync().thenCompose(
@@ -2101,10 +2086,20 @@ protected CompletableFuture internalRemoveNamespaceAntiAffinityGroupAsync(
protected CompletableFuture> internalGetAntiAffinityNamespacesAsync(String cluster,
String antiAffinityGroup,
String tenant) {
- checkNotNull(cluster, "Cluster should not be null");
- checkNotNull(antiAffinityGroup, "Anti-affinity group should not be null");
- checkNotNull(tenant, "Tenant should not be null");
- checkNotBlank(antiAffinityGroup, "Anti-affinity group can't be empty");
+ if (cluster == null) {
+ return FutureUtil.failedFuture(new RestException(Status.BAD_REQUEST, "Cluster should not be null"));
+ }
+ if (antiAffinityGroup == null) {
+ return FutureUtil.failedFuture(
+ new RestException(Status.BAD_REQUEST, "Anti-affinity group should not be null"));
+ }
+ if (tenant == null) {
+ return FutureUtil.failedFuture(new RestException(Status.BAD_REQUEST, "Tenant should not be null"));
+ }
+ if (StringUtils.isBlank(antiAffinityGroup)) {
+ return FutureUtil.failedFuture(
+ new RestException(Status.PRECONDITION_FAILED, "Anti-affinity group can't be empty"));
+ }
return validateNamespacePolicyOperationAsync(namespaceName, PolicyName.ANTI_AFFINITY, PolicyOperation.READ)
.thenCompose(__ -> validateClusterExistsAsync(cluster))
@@ -2135,39 +2130,48 @@ private boolean checkQuotas(Policies policies, RetentionPolicies retention) {
return checkBacklogQuota(quota, retention);
}
- private void clearBacklog(NamespaceName nsName, String bundleRange, String subscription) {
- try {
- List topicList = pulsar().getBrokerService().getAllTopicsFromNamespaceBundle(nsName.toString(),
- nsName.toString() + "/" + bundleRange);
-
- List> futures = new ArrayList<>();
- if (subscription != null) {
- if (subscription.startsWith(pulsar().getConfiguration().getReplicatorPrefix())) {
- subscription = PersistentReplicator.getRemoteCluster(subscription);
- }
- for (Topic topic : topicList) {
- if (topic instanceof PersistentTopic
- && !pulsar().getBrokerService().isSystemTopic(TopicName.get(topic.getName()))) {
- futures.add(((PersistentTopic) topic).clearBacklog(subscription));
+ private CompletableFuture clearBacklogAsync(NamespaceBundle bundle, String subscription) {
+ return pulsar().getNamespaceService().getOwnedPersistentTopicListForNamespaceBundle(bundle)
+ .thenCompose(topicsInBundle -> {
+ List> futures = new ArrayList<>();
+ String effectiveSubscription = subscription;
+ if (effectiveSubscription != null
+ && effectiveSubscription.startsWith(pulsar().getConfiguration().getReplicatorPrefix())) {
+ effectiveSubscription = PersistentReplicator.getRemoteCluster(effectiveSubscription);
}
- }
- } else {
- for (Topic topic : topicList) {
- if (topic instanceof PersistentTopic
- && !pulsar().getBrokerService().isSystemTopic(TopicName.get(topic.getName()))) {
- futures.add(((PersistentTopic) topic).clearBacklog());
+ final String finalSubscription = effectiveSubscription;
+
+ for (String topic : topicsInBundle) {
+ TopicName topicName = TopicName.get(topic);
+ if (pulsar().getBrokerService().isSystemTopic(topicName)) {
+ continue;
+ }
+ futures.add(pulsar().getBrokerService().getTopic(topicName.toString(), false)
+ .thenCompose(optTopic -> {
+ if (optTopic.isEmpty()) {
+ return CompletableFuture.completedFuture(null);
+ }
+ Topic loaded = optTopic.get();
+ if (!(loaded instanceof PersistentTopic persistentTopic)) {
+ return CompletableFuture.completedFuture(null);
+ }
+ return finalSubscription != null
+ ? persistentTopic.clearBacklog(finalSubscription)
+ : persistentTopic.clearBacklog();
+ }));
}
- }
- }
- FutureUtil.waitForAll(futures).get();
- } catch (Exception e) {
- log.error("[{}] Failed to clear backlog for namespace {}/{}, subscription: {}", clientAppId(),
- nsName.toString(), bundleRange, subscription, e);
- throw new RestException(e);
- }
+ return FutureUtil.waitForAll(futures);
+ }).exceptionally(ex -> {
+ Throwable cause = FutureUtil.unwrapCompletionException(ex);
+ if (cause instanceof RestException) {
+ throw (RestException) cause;
+ }
+ throw new RestException(cause);
+ });
}
+
private CompletableFuture unsubscribeAsync(NamespaceBundle bundle, String subscription) {
if (subscription.startsWith(pulsar().getConfiguration().getReplicatorPrefix())) {
return CompletableFuture.failedFuture(
@@ -2242,7 +2246,7 @@ private CompletableFuture validatePoliciesAsync(NamespaceName ns, Policies
log.info(msg);
return FutureUtil.failedFuture(new RestException(Status.BAD_REQUEST, msg));
}
- pulsar().getBrokerService().setCurrentClusterAllowedIfNoClusterIsAllowed(ns, policies);
+ pulsar().getBrokerService().setCurrentClusterAllowedWhenCreating(ns, policies);
// Validate cluster names and permissions
return Stream.concat(policies.replication_clusters.stream(), policies.allowed_clusters.stream())
@@ -2926,8 +2930,12 @@ protected void internalScanOffloadedLedgers(OffloaderObjectsScannerUtils.Scanner
validateNamespacePolicyOperation(namespaceName, PolicyName.OFFLOAD, PolicyOperation.READ);
Policies policies = getNamespacePolicies(namespaceName);
+ OffloadPoliciesImpl nsLevelOffloadPolicies = (OffloadPoliciesImpl) policies.offload_policies;
+ OffloadPoliciesImpl offloadPolicies = OffloadPoliciesImpl.mergeConfiguration(null,
+ OffloadPoliciesImpl.oldPoliciesCompatible(nsLevelOffloadPolicies, policies),
+ pulsar().getConfig().getProperties());
LedgerOffloader managedLedgerOffloader = pulsar()
- .getManagedLedgerOffloader(namespaceName, (OffloadPoliciesImpl) policies.offload_policies);
+ .getManagedLedgerOffloader(namespaceName, offloadPolicies);
String localClusterName = pulsar().getConfiguration().getClusterName();
@@ -3081,22 +3089,15 @@ protected void internalRemoveBacklogQuota(AsyncResponse asyncResponse, BacklogQu
});
}
- protected void internalEnableMigration(boolean migrated) {
- validateSuperUserAccess();
- try {
- getLocalPolicies().setLocalPoliciesWithCreate(namespaceName, oldPolicies -> oldPolicies.map(
- policies -> new LocalPolicies(policies.bundles,
- policies.bookieAffinityGroup,
- policies.namespaceAntiAffinityGroup,
- migrated))
- .orElseGet(() -> new LocalPolicies(getDefaultBundleData(), null, null, migrated)));
- log.info("Successfully updated migration on namespace {}", namespaceName);
- } catch (RestException re) {
- throw re;
- } catch (Exception e) {
- log.error("Failed to update migration on namespace {}", namespaceName, e);
- throw new RestException(e);
- }
+ protected CompletableFuture internalEnableMigrationAsync(boolean migrated) {
+ return validateSuperUserAccessAsync().thenCompose(__ -> getDefaultBundleDataAsync().thenCompose(
+ defaultBundleData -> getLocalPolicies().setLocalPoliciesWithCreateAsync(namespaceName,
+ oldPolicies -> oldPolicies.map(
+ policies -> new LocalPolicies(policies.bundles,
+ policies.bookieAffinityGroup,
+ policies.namespaceAntiAffinityGroup, migrated))
+ .orElseGet(() -> new LocalPolicies(defaultBundleData, null, null, migrated)))))
+ .thenAccept(__ -> log.info("Successfully updated migration on namespace {}", namespaceName));
}
protected Policies getDefaultPolicesIfNull(Policies policies) {
@@ -3183,19 +3184,6 @@ protected CompletableFuture> internalGetNamespaceAllowedClustersAsyn
.thenApply(policies -> policies.allowed_clusters);
}
- // TODO remove this sync method after async refactor
- @Deprecated
- private BundlesData getDefaultBundleData() {
- try {
- return getDefaultBundleDataAsync().get(config().getMetadataStoreOperationTimeoutSeconds(),
- TimeUnit.SECONDS);
- } catch (Exception e) {
- log.error("[{}] Failed to get namespace-policy configuration for namespace {}", clientAppId(),
- namespaceName, e);
- throw new RestException(e);
- }
- }
-
private CompletableFuture getDefaultBundleDataAsync() {
return namespaceResources().getPoliciesAsync(namespaceName).thenApply(
optionalPolicies -> optionalPolicies.isPresent() ? optionalPolicies.get().bundles :
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/ResourceQuotasBase.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/ResourceQuotasBase.java
index a1a3ce6558392..8e1956e6d0d71 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/ResourceQuotasBase.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/ResourceQuotasBase.java
@@ -71,7 +71,6 @@ private CompletableFuture getNamespaceBundleRangeAsync(String b
}
});
return ret
- .thenCompose(__ -> getNamespacePoliciesAsync(namespaceName))
- .thenApply(policies -> validateNamespaceBundleRange(namespaceName, policies.bundles, bundleRange));
+ .thenCompose(__ -> validateNamespaceBundleRangeAsync(namespaceName, bundleRange));
}
}
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v2/Namespaces.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v2/Namespaces.java
index b770c078a2876..ad01dd6f39c12 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v2/Namespaces.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v2/Namespaces.java
@@ -1516,14 +1516,14 @@ public void getPersistence(
public void clearNamespaceBacklog(@Suspended final AsyncResponse asyncResponse, @PathParam("tenant") String tenant,
@PathParam("namespace") String namespace,
@QueryParam("authoritative") @DefaultValue("false") boolean authoritative) {
- try {
- validateNamespaceName(tenant, namespace);
- internalClearNamespaceBacklog(asyncResponse, authoritative);
- } catch (WebApplicationException wae) {
- asyncResponse.resume(wae);
- } catch (Exception e) {
- asyncResponse.resume(new RestException(e));
- }
+ validateNamespaceName(tenant, namespace);
+ internalClearNamespaceBacklogAsync(authoritative)
+ .thenAccept(__ -> asyncResponse.resume(Response.noContent().build()))
+ .exceptionally(ex -> {
+ log.error("[{}] Failed to clear backlog on namespace {}", clientAppId(), namespaceName, ex);
+ resumeAsyncResponseExceptionally(asyncResponse, ex);
+ return null;
+ });
}
@POST
@@ -1534,11 +1534,19 @@ public void clearNamespaceBacklog(@Suspended final AsyncResponse asyncResponse,
@ApiResponse(code = 307, message = "Current broker doesn't serve the namespace"),
@ApiResponse(code = 403, message = "Don't have admin or operate permission on the namespace"),
@ApiResponse(code = 404, message = "Namespace does not exist") })
- public void clearNamespaceBundleBacklog(@PathParam("tenant") String tenant,
+ public void clearNamespaceBundleBacklog(@Suspended final AsyncResponse asyncResponse,
+ @PathParam("tenant") String tenant,
@PathParam("namespace") String namespace, @PathParam("bundle") String bundleRange,
@QueryParam("authoritative") @DefaultValue("false") boolean authoritative) {
validateNamespaceName(tenant, namespace);
- internalClearNamespaceBundleBacklog(bundleRange, authoritative);
+ internalClearNamespaceBundleBacklogAsync(bundleRange, authoritative)
+ .thenAccept(__ -> asyncResponse.resume(Response.noContent().build()))
+ .exceptionally(ex -> {
+ log.error("[{}] Failed to clear backlog on namespace bundle {}/{}", clientAppId(),
+ namespaceName, bundleRange, ex);
+ resumeAsyncResponseExceptionally(asyncResponse, ex);
+ return null;
+ });
}
@POST
@@ -1552,14 +1560,15 @@ public void clearNamespaceBacklogForSubscription(@Suspended final AsyncResponse
@PathParam("tenant") String tenant, @PathParam("namespace") String namespace,
@PathParam("subscription") String subscription,
@QueryParam("authoritative") @DefaultValue("false") boolean authoritative) {
- try {
- validateNamespaceName(tenant, namespace);
- internalClearNamespaceBacklogForSubscription(asyncResponse, subscription, authoritative);
- } catch (WebApplicationException wae) {
- asyncResponse.resume(wae);
- } catch (Exception e) {
- asyncResponse.resume(new RestException(e));
- }
+ validateNamespaceName(tenant, namespace);
+ internalClearNamespaceBacklogForSubscriptionAsync(subscription, authoritative)
+ .thenAccept(__ -> asyncResponse.resume(Response.noContent().build()))
+ .exceptionally(ex -> {
+ log.error("[{}] Failed to clear backlog for subscription {} on namespace {}", clientAppId(),
+ subscription, namespaceName, ex);
+ resumeAsyncResponseExceptionally(asyncResponse, ex);
+ return null;
+ });
}
@POST
@@ -1570,12 +1579,20 @@ public void clearNamespaceBacklogForSubscription(@Suspended final AsyncResponse
@ApiResponse(code = 307, message = "Current broker doesn't serve the namespace"),
@ApiResponse(code = 403, message = "Don't have admin or operate permission on the namespace"),
@ApiResponse(code = 404, message = "Namespace does not exist") })
- public void clearNamespaceBundleBacklogForSubscription(@PathParam("tenant") String tenant,
+ public void clearNamespaceBundleBacklogForSubscription(@Suspended final AsyncResponse asyncResponse,
+ @PathParam("tenant") String tenant,
@PathParam("namespace") String namespace, @PathParam("subscription") String subscription,
@PathParam("bundle") String bundleRange,
@QueryParam("authoritative") @DefaultValue("false") boolean authoritative) {
validateNamespaceName(tenant, namespace);
- internalClearNamespaceBundleBacklogForSubscription(subscription, bundleRange, authoritative);
+ internalClearNamespaceBundleBacklogForSubscriptionAsync(subscription, bundleRange, authoritative)
+ .thenAccept(__ -> asyncResponse.resume(Response.noContent().build()))
+ .exceptionally(ex -> {
+ log.error("[{}] Failed to clear backlog for subscription {} on namespace bundle {}/{}",
+ clientAppId(), subscription, namespaceName, bundleRange, ex);
+ resumeAsyncResponseExceptionally(asyncResponse, ex);
+ return null;
+ });
}
@POST
@@ -3158,11 +3175,19 @@ public void removeNamespaceEntryFilters(@Suspended AsyncResponse asyncResponse,
@ApiResponse(code = 200, message = "Operation successful"),
@ApiResponse(code = 403, message = "Don't have admin permission"),
@ApiResponse(code = 404, message = "Property or cluster or namespace doesn't exist") })
- public void enableMigration(@PathParam("tenant") String tenant,
+ public void enableMigration(@Suspended AsyncResponse asyncResponse,
+ @PathParam("tenant") String tenant,
@PathParam("namespace") String namespace,
boolean migrated) {
validateNamespaceName(tenant, namespace);
- internalEnableMigration(migrated);
+ internalEnableMigrationAsync(migrated)
+ .thenAccept(__ -> asyncResponse.resume(Response.noContent().build()))
+ .exceptionally(ex -> {
+ log.error("[{}] Failed to update migration, tenant: {}, namespace: {}",
+ clientAppId(), tenant, namespace, ex);
+ resumeAsyncResponseExceptionally(asyncResponse, ex);
+ return null;
+ });
}
@POST
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v2/NonPersistentTopics.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v2/NonPersistentTopics.java
index 16cbc7104d4d8..cf06b84c365c4 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v2/NonPersistentTopics.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v2/NonPersistentTopics.java
@@ -463,18 +463,17 @@ public void getListFromBundle(
}
validateNamespaceOperation(namespaceName, NamespaceOperation.GET_BUNDLE);
- Policies policies = getNamespacePolicies(namespaceName);
// check cluster ownership for a given global namespace: redirect if peer-cluster owns it
validateGlobalNamespaceOwnership(namespaceName);
- isBundleOwnedByAnyBroker(namespaceName, policies.bundles, bundleRange).thenAccept(flag -> {
+ isBundleOwnedByAnyBroker(namespaceName, bundleRange).thenAccept(flag -> {
if (!flag) {
log.info("[{}] Namespace bundle is not owned by any broker {}/{}", clientAppId(), namespaceName,
bundleRange);
asyncResponse.resume(Response.noContent().build());
} else {
- validateNamespaceBundleOwnershipAsync(namespaceName, policies.bundles, bundleRange, true, true)
+ validateNamespaceBundleOwnershipAsync(namespaceName, bundleRange, true, true)
.thenAccept(nsBundle -> {
final var bundleTopics = pulsar().getBrokerService().getMultiLayerTopicsMap()
.get(namespaceName.toString());
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/AbstractDelayedDeliveryTracker.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/AbstractDelayedDeliveryTracker.java
index bec5134c4f79a..de843d035068b 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/AbstractDelayedDeliveryTracker.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/AbstractDelayedDeliveryTracker.java
@@ -34,20 +34,25 @@ public abstract class AbstractDelayedDeliveryTracker implements DelayedDeliveryT
// Reference to the shared (per-broker) timer for delayed delivery
protected final Timer timer;
- // Current timeout or null if not set
- protected Timeout timeout;
+ // Current timeout or null if not set. Guarded by timeoutLock.
+ private Timeout timeout;
- // Timestamp at which the timeout is currently set
+ // Timestamp at which the timeout is currently set. Guarded by timeoutLock.
private long currentTimeoutTarget;
- // Last time the TimerTask was triggered for this class
+ // Last time the TimerTask was triggered for this class. Guarded by timeoutLock.
private long lastTickRun;
- protected long tickTimeMillis;
+ // Updated through resetTickTime() from dispatcher threads and read on the timer thread.
+ protected volatile long tickTimeMillis;
protected final Clock clock;
private final boolean isDelayedDeliveryDeliverAtTimeStrict;
+ // Guards the timer state (timeout, currentTimeoutTarget, lastTickRun) against concurrent access from
+ // dispatcher threads (updateTimer/rescheduleTimer/close) and the timer thread (run). It is a leaf lock:
+ // no subclass method is invoked while holding it.
+ private final Object timeoutLock = new Object();
public AbstractDelayedDeliveryTracker(AbstractPersistentDispatcherMultipleConsumers dispatcher, Timer timer,
long tickTimeMillis,
@@ -81,14 +86,34 @@ protected long getCutoffTime() {
return isDelayedDeliveryDeliverAtTimeStrict ? clock.millis() : clock.millis() + tickTimeMillis;
}
+ protected boolean isDeliverAtTimeStrict() {
+ return isDelayedDeliveryDeliverAtTimeStrict;
+ }
+
public void resetTickTime(long tickTime) {
if (this.tickTimeMillis != tickTime) {
this.tickTimeMillis = tickTime;
}
}
- protected void updateTimer() {
- if (getNumberOfDelayedMessages() == 0) {
+ /**
+ * Update the delivery timer to fire when the next message in the tracker becomes due.
+ *
+ * Callers are expected to serialize all tracker state mutations (at the dispatcher or tracker level), so the
+ * snapshot of {@link #getNumberOfDelayedMessages()} and {@link #nextDeliveryTime()} is taken before acquiring
+ * timeoutLock. This keeps timeoutLock a leaf lock that never calls into subclass methods, ruling out lock
+ * ordering deadlocks with subclasses that synchronize those methods on the tracker instance.
+ */
+ protected final void updateTimer() {
+ long numberOfDelayedMessages = getNumberOfDelayedMessages();
+ long nextDeliveryTimestamp = numberOfDelayedMessages > 0 ? nextDeliveryTime() : -1;
+ synchronized (timeoutLock) {
+ doUpdateTimer(numberOfDelayedMessages, nextDeliveryTimestamp);
+ }
+ }
+
+ private void doUpdateTimer(long numberOfDelayedMessages, long timestamp) {
+ if (numberOfDelayedMessages == 0) {
if (timeout != null) {
currentTimeoutTarget = -1;
timeout.cancel();
@@ -96,7 +121,6 @@ protected void updateTimer() {
}
return;
}
- long timestamp = nextDeliveryTime();
if (timestamp == currentTimeoutTarget) {
// The timer is already set to the correct target time
return;
@@ -104,18 +128,22 @@ protected void updateTimer() {
if (timeout != null) {
timeout.cancel();
+ timeout = null;
}
+ // Reset the tracked state so a subsequent updateTimer() call cannot short-circuit on a stale
+ // currentTimeoutTarget while no live timer remains. See #25996.
+ currentTimeoutTarget = -1;
long now = clock.millis();
long delayMillis = timestamp - now;
- if (delayMillis < 0) {
+ if (delayMillis <= 0) {
// There are messages that are already ready to be delivered. If
// the dispatcher is not getting them is because the consumer is
// either not connected or slow.
// We don't need to keep retriggering the timer. When the consumer
// catches up, the dispatcher will do the readMoreEntries() and
- // get these messages
+ // get these messages.
return;
}
@@ -126,7 +154,6 @@ protected void updateTimer() {
if (log.isDebugEnabled()) {
log.debug("[{}] Start timer in {} millis", dispatcher.getName(), calculatedDelayMillis);
}
-
// Even though we may delay longer than this timestamp because of the tick delay, we still track the
// current timeout with reference to the next message's timestamp.
currentTimeoutTarget = timestamp;
@@ -134,27 +161,53 @@ protected void updateTimer() {
}
@Override
- public void run(Timeout timeout) throws Exception {
+ public void run(Timeout triggeredTimeout) throws Exception {
if (log.isDebugEnabled()) {
log.debug("[{}] Timer triggered", dispatcher.getName());
}
- if (timeout == null || timeout.isCancelled()) {
+ if (triggeredTimeout == null || triggeredTimeout.isCancelled()) {
return;
}
- synchronized (dispatcher) {
+ synchronized (timeoutLock) {
lastTickRun = clock.millis();
- currentTimeoutTarget = -1;
- this.timeout = null;
+ // Only reset the timer state if the triggered timeout is the currently armed one. A timeout that
+ // was already superseded by updateTimer()/rescheduleTimer() may still fire if it passed its
+ // isCancelled() check before being cancelled; it must not clear the state of the newer timer.
+ if (triggeredTimeout == this.timeout) {
+ currentTimeoutTarget = -1;
+ this.timeout = null;
+ }
+ }
+
+ synchronized (dispatcher) {
dispatcher.readMoreEntriesAsync();
}
}
+ /**
+ * Cancel the current timer (if any) and schedule the timer task to run after the given delay. Used by
+ * subclasses to trigger a dispatch round from asynchronous completions instead of mutating the timer
+ * state directly.
+ */
+ protected final void rescheduleTimer(long delayMillis) {
+ synchronized (timeoutLock) {
+ if (timeout != null) {
+ timeout.cancel();
+ }
+ currentTimeoutTarget = -1;
+ timeout = timer.newTimeout(this, delayMillis, TimeUnit.MILLISECONDS);
+ }
+ }
+
@Override
public void close() {
- if (timeout != null) {
- timeout.cancel();
- timeout = null;
+ synchronized (timeoutLock) {
+ if (timeout != null) {
+ timeout.cancel();
+ timeout = null;
+ }
+ currentTimeoutTarget = -1;
}
}
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/DelayedDeliveryTracker.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/DelayedDeliveryTracker.java
index 7c954879fe845..96e7151e7ea45 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/DelayedDeliveryTracker.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/DelayedDeliveryTracker.java
@@ -86,6 +86,16 @@ public interface DelayedDeliveryTracker extends AutoCloseable {
*/
void close();
+ /**
+ * Close the subscription tracker and release all resources, completing the returned future once the
+ * tracker has finished closing. Prefer this over {@link #close()} on asynchronous paths (e.g. inside a
+ * {@link CompletableFuture} continuation) so the caller is not blocked.
+ */
+ default CompletableFuture closeAsync() {
+ close();
+ return CompletableFuture.completedFuture(null);
+ }
+
DelayedDeliveryTracker DISABLE = new DelayedDeliveryTracker() {
@Override
public boolean addMessage(long ledgerId, long entryId, long deliveryAt) {
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/InMemoryDelayedDeliveryTracker.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/InMemoryDelayedDeliveryTracker.java
index ad5ab25fbbf6b..089069968340e 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/InMemoryDelayedDeliveryTracker.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/InMemoryDelayedDeliveryTracker.java
@@ -64,6 +64,7 @@ public class InMemoryDelayedDeliveryTracker extends AbstractDelayedDeliveryTrack
// The bit count to trim to reduce memory occupation.
private final int timestampPrecisionBitCnt;
+ private final long precisionMillis;
// Count of delayed messages in the tracker.
private final AtomicLong delayedMessagesCount = new AtomicLong(0);
@@ -83,6 +84,7 @@ public InMemoryDelayedDeliveryTracker(AbstractPersistentDispatcherMultipleConsum
super(dispatcher, timer, tickTimeMillis, clock, isDelayedDeliveryDeliverAtTimeStrict);
this.fixedDelayDetectionLookahead = fixedDelayDetectionLookahead;
this.timestampPrecisionBitCnt = calculateTimestampPrecisionBitCnt(tickTimeMillis);
+ this.precisionMillis = 1L << timestampPrecisionBitCnt;
}
/**
@@ -125,11 +127,17 @@ public boolean addMessage(long ledgerId, long entryId, long deliverAt) {
deliverAt - clock.millis());
}
- long timestamp = trimLowerBit(deliverAt, timestampPrecisionBitCnt);
- delayedMessageMap.computeIfAbsent(timestamp, k -> new Long2ObjectRBTreeMap<>())
- .computeIfAbsent(ledgerId, k -> new Roaring64Bitmap())
- .add(entryId);
- delayedMessagesCount.incrementAndGet();
+ long timestamp = roundTimestamp(deliverAt);
+ Roaring64Bitmap bitmap = delayedMessageMap.computeIfAbsent(timestamp, k -> new Long2ObjectRBTreeMap<>())
+ .computeIfAbsent(ledgerId, k -> new Roaring64Bitmap());
+ // Roaring64Bitmap does not store duplicates, so track if it a new element
+ // so we can keep delayedMessagesCount in sync
+ boolean isNew = !bitmap.contains(entryId);
+
+ if (isNew) {
+ bitmap.addLong(entryId);
+ delayedMessagesCount.incrementAndGet();
+ }
updateTimer();
@@ -138,6 +146,33 @@ public boolean addMessage(long ledgerId, long entryId, long deliverAt) {
return true;
}
+ /**
+ * Round the deliverAt timestamp to the bucket boundary used as the key in {@link #delayedMessageMap}, so that
+ * all messages within the same bucket share a single map entry to reduce memory usage.
+ *
+ * In strict delivery mode the timestamp is rounded up: a bucket then becomes due only after every deliverAt
+ * time inside it has passed, so messages are delivered up to one bucket (less than tickTimeMillis) late, but
+ * never before their deliverAt time. Rounding down instead would let {@link #getScheduledMessages(int)} hand a
+ * message to the dispatcher before its deliverAt time; the dispatcher would put it back and re-trigger reads
+ * in a loop until the deliverAt time is reached (see issue #25996).
+ *
+ * In non-strict mode the timestamp is rounded down, since delivering up to tickTimeMillis early is allowed.
+ * Availability checks account for the full bucket width so that the original deliverAt is still within that
+ * allowed window.
+ */
+ private long roundTimestamp(long deliverAt) {
+ if (isDeliverAtTimeStrict()) {
+ // round up, saturating at Long.MAX_VALUE instead of overflowing for deliverAt close to Long.MAX_VALUE
+ long roundedUp = deliverAt + precisionMillis - 1;
+ return trimLowerBit(roundedUp < deliverAt ? Long.MAX_VALUE : roundedUp, timestampPrecisionBitCnt);
+ }
+ return trimLowerBit(deliverAt, timestampPrecisionBitCnt);
+ }
+
+ private long getBucketCutoffTime() {
+ return isDeliverAtTimeStrict() ? getCutoffTime() : getCutoffTime() - precisionMillis + 1;
+ }
+
/**
* Check that new delivery time comes after the current highest, or at
* least within a single tick time interval of 1 second.
@@ -156,7 +191,7 @@ private void checkAndUpdateHighest(long deliverAt) {
@Override
public boolean hasMessageAvailable() {
boolean hasMessageAvailable = !delayedMessageMap.isEmpty()
- && delayedMessageMap.firstLongKey() <= getCutoffTime();
+ && delayedMessageMap.firstLongKey() <= getBucketCutoffTime();
if (!hasMessageAvailable) {
updateTimer();
}
@@ -170,7 +205,7 @@ public boolean hasMessageAvailable() {
public NavigableSet getScheduledMessages(int maxMessages) {
int n = maxMessages;
NavigableSet positions = new TreeSet<>();
- long cutoffTime = getCutoffTime();
+ long cutoffTime = getBucketCutoffTime();
while (n > 0 && !delayedMessageMap.isEmpty()) {
long timestamp = delayedMessageMap.firstLongKey();
@@ -183,20 +218,22 @@ public NavigableSet getScheduledMessages(int maxMessages) {
for (Long2ObjectMap.Entry ledgerEntry : ledgerMap.long2ObjectEntrySet()) {
long ledgerId = ledgerEntry.getLongKey();
Roaring64Bitmap entryIds = ledgerEntry.getValue();
- int cardinality = (int) entryIds.getLongCardinality();
+ long cardinality = entryIds.getLongCardinality();
if (cardinality <= n) {
+ int cardinalityInt = (int) cardinality;
entryIds.forEach(entryId -> {
positions.add(PositionFactory.create(ledgerId, entryId));
});
- n -= cardinality;
- delayedMessagesCount.addAndGet(-cardinality);
+ n -= cardinalityInt;
+ delayedMessagesCount.addAndGet(-cardinalityInt);
ledgerIdToDelete.add(ledgerId);
} else {
- long[] entryIdsArray = entryIds.toArray();
- for (int i = 0; i < n; i++) {
- positions.add(PositionFactory.create(ledgerId, entryIdsArray[i]));
- entryIds.removeLong(entryIdsArray[i]);
- }
+ Roaring64Bitmap entryIdsToRemove = new Roaring64Bitmap();
+ entryIds.stream().limit(n).forEach(entryId -> {
+ positions.add(PositionFactory.create(ledgerId, entryId));
+ entryIdsToRemove.addLong(entryId);
+ });
+ entryIds.andNot(entryIdsToRemove);
delayedMessagesCount.addAndGet(-n);
n = 0;
}
@@ -211,7 +248,6 @@ public NavigableSet getScheduledMessages(int maxMessages) {
delayedMessageMap.remove(timestamp);
}
}
-
if (log.isDebugEnabled()) {
log.debug("[{}] Get scheduled messages - found {}", dispatcher.getName(), positions.size());
}
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/bucket/BucketDelayedDeliveryTracker.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/bucket/BucketDelayedDeliveryTracker.java
index 9ed304c73312d..4b6121fd3e4a0 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/bucket/BucketDelayedDeliveryTracker.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/bucket/BucketDelayedDeliveryTracker.java
@@ -38,6 +38,7 @@
import java.util.Optional;
import java.util.TreeSet;
import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionException;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
@@ -48,6 +49,7 @@
import lombok.Getter;
import lombok.extern.slf4j.Slf4j;
import org.apache.bookkeeper.mledger.ManagedCursor;
+import org.apache.bookkeeper.mledger.ManagedLedger;
import org.apache.bookkeeper.mledger.Position;
import org.apache.bookkeeper.mledger.PositionFactory;
import org.apache.commons.collections4.CollectionUtils;
@@ -112,6 +114,8 @@ public static record SnapshotKey(long ledgerId, long entryId) {}
private CompletableFuture pendingLoad = null;
+ private volatile CompletableFuture trimFuture;
+
public BucketDelayedDeliveryTracker(AbstractPersistentDispatcherMultipleConsumers dispatcher,
Timer timer, long tickTimeMillis,
boolean isDelayedDeliveryDeliverAtTimeStrict,
@@ -387,8 +391,15 @@ public synchronized boolean addMessage(long ledgerId, long entryId, long deliver
afterCreateImmutableBucket(immutableBucketDelayedIndexPair, createStartTime);
lastMutableBucket.resetLastMutableBucketRange();
- if (maxNumBuckets > 0 && immutableBuckets.asMapOfRanges().size() > maxNumBuckets) {
- asyncMergeBucketSnapshot();
+ if (maxNumBuckets > 0 && immutableBuckets.asMapOfRanges().size() > maxNumBuckets
+ && (trimFuture == null || trimFuture.isDone())) {
+ trimFuture = asyncTrimImmutableBuckets()
+ .thenCompose(ignore -> asyncMergeBucketSnapshot())
+ .whenComplete((ignore, t) -> {
+ if (t != null) {
+ log.warn("Failed to trim or merge bucket snapshots", t);
+ }
+ });
}
}
@@ -452,6 +463,10 @@ private synchronized List selectMergedBuckets(final List asyncMergeBucketSnapshot() {
List immutableBucketList = immutableBuckets.asMapOfRanges().values().stream().toList();
+ if (maxNumBuckets <= 0 || immutableBucketList.size() <= maxNumBuckets) {
+ return CompletableFuture.completedFuture(null);
+ }
+
List toBeMergeImmutableBuckets = selectMergedBuckets(immutableBucketList, MAX_MERGE_NUM);
if (toBeMergeImmutableBuckets.isEmpty()) {
@@ -578,6 +593,11 @@ protected synchronized long nextDeliveryTime() {
return sharedBucketPriorityQueue.peekN1();
} else if (sharedBucketPriorityQueue.isEmpty() && !lastMutableBucket.isEmpty()) {
return lastMutableBucket.nextDeliveryTime();
+ } else if (lastMutableBucket.isEmpty() && sharedBucketPriorityQueue.isEmpty()) {
+ // numberDelayedMessages can be > 0 while both queues are empty (e.g. remaining
+ // messages live in not-yet-loaded snapshot segments). Returning Long.MAX_VALUE
+ // signals "no imminent delivery" without throwing on the empty queues.
+ return Long.MAX_VALUE;
}
long timestamp = lastMutableBucket.nextDeliveryTime();
long bucketTimestamp = sharedBucketPriorityQueue.peekN1();
@@ -605,6 +625,7 @@ public synchronized NavigableSet getScheduledMessages(int maxMessages)
}
long cutoffTime = getCutoffTime();
+ Long firstLiveLedgerId = firstActiveLedgerId();
lastMutableBucket.moveScheduledMessageToSharedQueue(cutoffTime, sharedBucketPriorityQueue);
@@ -613,13 +634,19 @@ public synchronized NavigableSet getScheduledMessages(int maxMessages)
while (n > 0 && !sharedBucketPriorityQueue.isEmpty()) {
long timestamp = sharedBucketPriorityQueue.peekN1();
+ long ledgerId = sharedBucketPriorityQueue.peekN2();
+ long entryId = sharedBucketPriorityQueue.peekN3();
+ if (firstLiveLedgerId != null && ledgerId < firstLiveLedgerId) {
+ sharedBucketPriorityQueue.pop();
+ if (removeIndexBit(ledgerId, entryId)) {
+ numberDelayedMessages.decrementAndGet();
+ }
+ continue;
+ }
if (timestamp > cutoffTime) {
break;
}
- long ledgerId = sharedBucketPriorityQueue.peekN2();
- long entryId = sharedBucketPriorityQueue.peekN3();
-
SnapshotKey snapshotKey = new SnapshotKey(ledgerId, entryId);
ImmutableBucket bucket = snapshotSegmentLastIndexMap.get(snapshotKey);
@@ -682,12 +709,7 @@ public synchronized NavigableSet getScheduledMessages(int maxMessages)
stats.recordSuccessEvent(BucketDelayedMessageIndexStats.Type.load,
System.currentTimeMillis() - loadStartTime);
}
- synchronized (this) {
- if (timeout != null) {
- timeout.cancel();
- }
- timeout = timer.newTimeout(this, 0, TimeUnit.MILLISECONDS);
- }
+ rescheduleTimer(0);
});
if (!checkPendingLoadDone() || loadFuture.isCompletedExceptionally()) {
@@ -724,26 +746,49 @@ public boolean shouldPauseAllDeliveries() {
@Override
public synchronized CompletableFuture clear() {
- CompletableFuture future = cleanImmutableBuckets();
- sharedBucketPriorityQueue.clear();
- lastMutableBucket.clear();
- snapshotSegmentLastIndexMap.clear();
- numberDelayedMessages.set(0);
- return future;
+ // Wait for any in-flight trim+merge to settle, then clear.
+ // Reuse trimFuture to block new triggers until the clear chain completes.
+ CompletableFuture before = trimFuture != null && !trimFuture.isDone()
+ ? trimFuture : CompletableFuture.completedFuture(null);
+ trimFuture = before
+ .exceptionally(t -> {
+ log.warn("Trim/merge buckets failed, but still clear", t);
+ return null;
+ })
+ .thenCompose(__ -> {
+ synchronized (BucketDelayedDeliveryTracker.this) {
+ CompletableFuture future = cleanImmutableBuckets();
+ sharedBucketPriorityQueue.clear();
+ lastMutableBucket.clear();
+ snapshotSegmentLastIndexMap.clear();
+ numberDelayedMessages.set(0);
+ return future;
+ }
+ });
+ return trimFuture;
}
@Override
- public synchronized void close() {
- super.close();
- lastMutableBucket.close();
- sharedBucketPriorityQueue.close();
- try {
- List> completableFutures = immutableBuckets.asMapOfRanges().values().stream()
+ public void close() {
+ // Block for AutoCloseable / synchronous callers; asynchronous callers should use closeAsync().
+ closeAsync().join();
+ }
+
+ @Override
+ public CompletableFuture closeAsync() {
+ List> completableFutures;
+ synchronized (this) {
+ super.close();
+ lastMutableBucket.close();
+ sharedBucketPriorityQueue.close();
+ completableFutures = immutableBuckets.asMapOfRanges().values().stream()
.map(bucket -> bucket.getSnapshotCreateFuture().orElse(NULL_LONG_PROMISE)).toList();
- FutureUtil.waitForAll(completableFutures).get(AsyncOperationTimeoutSeconds, TimeUnit.SECONDS);
- } catch (Exception e) {
- log.warn("[{}] Failed wait to snapshot generate", dispatcher.getName(), e);
}
+ return FutureUtil.waitForAll(completableFutures)
+ .exceptionally(e -> {
+ log.warn("[{}] Failed wait to snapshot generate", dispatcher.getName(), e);
+ return null;
+ });
}
private CompletableFuture cleanImmutableBuckets() {
@@ -787,4 +832,56 @@ public Map genTopicMetricMap() {
stats.recordBucketSnapshotSizeBytes(totalSnapshotLength.longValue());
return stats.genTopicMetricMap();
}
+
+ /**
+ * Delete orphaned bucket snapshots whose ledger range lies entirely before the earliest
+ * surviving ledger. Buckets are deleted sequentially; the chain stops on first failure
+ * to avoid wasted work when storage is unavailable.
+ */
+ private synchronized CompletableFuture asyncTrimImmutableBuckets() {
+ Long firstLedgerId = firstActiveLedgerId();
+ if (null == firstLedgerId) {
+ return CompletableFuture.completedFuture(null);
+ }
+ ManagedLedger ledger = dispatcher.getCursor().getManagedLedger();
+
+ Map, ImmutableBucket> toBeDeletedBuckets =
+ new HashMap<>(immutableBuckets.subRangeMap(Range.lessThan(firstLedgerId)).asMapOfRanges());
+
+ if (toBeDeletedBuckets.isEmpty()) {
+ return CompletableFuture.completedFuture(null);
+ }
+
+ String ledgerName = ledger.getName();
+ CompletableFuture chain = CompletableFuture.completedFuture(null);
+ for (Map.Entry, ImmutableBucket> entry : toBeDeletedBuckets.entrySet()) {
+ chain = chain.thenCompose(__ ->
+ deleteBucketSnapshot(ledgerName, entry.getKey(), entry.getValue()));
+ }
+ return chain;
+ }
+
+ private CompletableFuture deleteBucketSnapshot(String ledgerName,
+ Range range, ImmutableBucket bucket) {
+ return bucket.asyncDeleteBucketSnapshot(stats)
+ .handle((__, t) -> {
+ if (t != null) {
+ log.warn("Failed to delete bucket snapshot, LedgerName: {}, BucketKey: {}",
+ ledgerName, bucket.bucketKey());
+ throw new CompletionException(t);
+ }
+ synchronized (this) {
+ snapshotSegmentLastIndexMap.entrySet().removeIf(entry -> entry.getValue() == bucket);
+ immutableBuckets.remove(range);
+ numberDelayedMessages.addAndGet(-bucket.getNumberBucketDelayedMessages());
+ }
+ return null;
+ });
+ }
+
+ private Long firstActiveLedgerId() {
+ ManagedCursor cursor = dispatcher.getCursor();
+ Position mdp = cursor.getMarkDeletedPosition();
+ return mdp == null ? null : mdp.getLedgerId();
+ }
}
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/LeaderElectionService.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/LeaderElectionService.java
index 2e53b54e98f61..21f67bd6b3613 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/LeaderElectionService.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/LeaderElectionService.java
@@ -56,10 +56,20 @@ public void close() throws Exception {
leaderElection.close();
}
+ /**
+ * Authoritative read of the current leader: if a leader election is in progress, the returned
+ * future completes once it settles (bounded by the default metadata operation timeout). Use
+ * this whenever a decision is made based on who the leader is.
+ */
public CompletableFuture> readCurrentLeader() {
return leaderElection.getLeaderValue();
}
+ /**
+ * Non-blocking snapshot of the current leader; empty while a re-election is settling even
+ * though a leader may technically exist. Only suitable for best-effort uses such as logging —
+ * decision-making callers must use {@link #readCurrentLeader()}.
+ */
public Optional getCurrentLeader() {
return leaderElection.getLeaderValueIfPresent();
}
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/LoadManager.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/LoadManager.java
index 9399fcb3aad7a..c9aa673d6b53d 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/LoadManager.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/LoadManager.java
@@ -151,35 +151,31 @@ default void writeLoadReportOnZookeeper(boolean force) throws Exception {
void initialize(PulsarService pulsar);
static LoadManager create(final PulsarService pulsar) {
- try {
- final ServiceConfiguration conf = pulsar.getConfiguration();
-
- String loadManagerClassName = conf.getLoadManagerClassName();
- if (StringUtils.isBlank(loadManagerClassName)) {
- loadManagerClassName = SimpleLoadManagerImpl.class.getName();
- }
-
- // Assume there is a constructor with one argument of PulsarService.
- final Object loadManagerInstance = Reflections.createInstance(loadManagerClassName,
- Thread.currentThread().getContextClassLoader());
- if (loadManagerInstance instanceof LoadManager casted) {
- casted.initialize(pulsar);
- return casted;
- } else if (loadManagerInstance instanceof ModularLoadManager modularLoadManager) {
- final LoadManager casted = new ModularLoadManagerWrapper(modularLoadManager);
- casted.initialize(pulsar);
- return casted;
- } else if (loadManagerInstance instanceof ExtensibleLoadManager) {
- final LoadManager casted =
- new ExtensibleLoadManagerWrapper((ExtensibleLoadManagerImpl) loadManagerInstance);
- casted.initialize(pulsar);
- return casted;
- }
- } catch (Exception e) {
- LOG.warn("Error when trying to create load manager: ", e);
+ final ServiceConfiguration conf = pulsar.getConfiguration();
+
+ String loadManagerClassName = conf.getLoadManagerClassName();
+ if (StringUtils.isBlank(loadManagerClassName)) {
+ loadManagerClassName = SimpleLoadManagerImpl.class.getName();
+ }
+
+ // Assume there is a constructor with one argument of PulsarService.
+ final Object loadManagerInstance = Reflections.createInstance(loadManagerClassName,
+ Thread.currentThread().getContextClassLoader());
+ if (loadManagerInstance instanceof LoadManager casted) {
+ casted.initialize(pulsar);
+ return casted;
+ } else if (loadManagerInstance instanceof ModularLoadManager modularLoadManager) {
+ final LoadManager casted = new ModularLoadManagerWrapper(modularLoadManager);
+ casted.initialize(pulsar);
+ return casted;
+ } else if (loadManagerInstance instanceof ExtensibleLoadManager) {
+ final LoadManager casted =
+ new ExtensibleLoadManagerWrapper((ExtensibleLoadManagerImpl) loadManagerInstance);
+ casted.initialize(pulsar);
+ return casted;
+ } else {
+ throw new IllegalArgumentException(loadManagerClassName + " is not a supported LoadManager");
}
- // If we failed to create a load manager, default to SimpleLoadManagerImpl.
- return new SimpleLoadManagerImpl(pulsar);
}
}
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/ModularLoadManager.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/ModularLoadManager.java
index d608bd6784f29..93059d722e734 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/ModularLoadManager.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/ModularLoadManager.java
@@ -138,8 +138,22 @@ default void writeBrokerDataOnZooKeeper(boolean force) {
*
* @param bundle
* @return bundle data
+ * @deprecated use {@link #getBundleDataOrDefaultAsync(String)} instead; this method blocks on
+ * metadata-store reads and must not be called from async or event-loop threads.
*/
+ @Deprecated
BundleData getBundleDataOrDefault(String bundle);
+ /**
+ * Asynchronously fetch bundle's load report data.
+ *
+ * @param bundle
+ * @return future of the bundle data
+ */
+ @SuppressWarnings("deprecation")
+ default CompletableFuture getBundleDataOrDefaultAsync(String bundle) {
+ return CompletableFuture.completedFuture(getBundleDataOrDefault(bundle));
+ }
+
String setNamespaceBundleAffinity(String bundle, String broker);
}
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/channel/ServiceUnitStateTableViewSyncer.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/channel/ServiceUnitStateTableViewSyncer.java
index 45ff0dcb2674e..999c7bea93ad2 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/channel/ServiceUnitStateTableViewSyncer.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/channel/ServiceUnitStateTableViewSyncer.java
@@ -23,20 +23,26 @@
import static org.apache.pulsar.broker.loadbalance.extensions.ExtensibleLoadManagerImpl.COMPACTION_THRESHOLD;
import static org.apache.pulsar.broker.loadbalance.extensions.ExtensibleLoadManagerImpl.configureSystemTopics;
import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.ObjectWriter;
+import com.google.common.annotations.VisibleForTesting;
import java.io.Closeable;
import java.io.IOException;
import java.util.ArrayList;
+import java.util.HashMap;
import java.util.List;
import java.util.Optional;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
+import java.util.function.BiConsumer;
import lombok.Cleanup;
import lombok.extern.slf4j.Slf4j;
import org.apache.pulsar.broker.PulsarService;
+import org.apache.pulsar.client.api.PulsarClientException;
import org.apache.pulsar.common.util.FutureUtil;
import org.apache.pulsar.common.util.ObjectMapperFactory;
+import org.jspecify.annotations.NonNull;
/**
* Defines ServiceUnitTableViewSyncer.
@@ -47,10 +53,15 @@
public class ServiceUnitStateTableViewSyncer implements Closeable {
private static final int MAX_CONCURRENT_SYNC_COUNT = 100;
private static final int SYNC_WAIT_TIME_IN_SECS = 300;
+ private static final long RECONCILE_INTERVAL_IN_MILLIS = 5_000;
+ private static final BiConsumer NOOP_CONSUMER = (__, ___) -> {
+ };
+ private volatile int syncWaitTimeInSecs = SYNC_WAIT_TIME_IN_SECS;
private PulsarService pulsar;
private volatile ServiceUnitStateTableView systemTopicTableView;
private volatile ServiceUnitStateTableView metadataStoreTableView;
private volatile boolean isActive = false;
+ private final ObjectWriter jsonWriter = ObjectMapperFactory.getMapper().writer();
public void start(PulsarService pulsar)
@@ -82,53 +93,77 @@ public void start(PulsarService pulsar)
}
private CompletableFuture syncToSystemTopic(String key, ServiceUnitStateData data) {
- return systemTopicTableView.put(key, data);
+ return logIfFailed(sync(systemTopicTableView, key, data), key, data, "systemTopic");
}
private CompletableFuture syncToMetadataStore(String key, ServiceUnitStateData data) {
- return metadataStoreTableView.put(key, data);
+ return logIfFailed(sync(metadataStoreTableView, key, data), key, data, "metadataStore");
}
- private void dummy(String key, ServiceUnitStateData data) {
+ private CompletableFuture sync(ServiceUnitStateTableView dst, String key, ServiceUnitStateData data) {
+ // A null tail item is a tombstone: the source view removed the key. Route it to
+ // delete() rather than put(): the metadata-store view's put() rejects null
+ // (@NonNull) and the system-topic view's delete() is itself a null-valued put(),
+ // so a uniform delete keeps both sync directions symmetric and prevents a missed
+ // deletion from leaving the two views with different sizes (which would make
+ // waitUntilSynced spin until the timeout budget).
+ return data == null ? dst.delete(key) : dst.put(key, data);
+ }
+
+ private CompletableFuture logIfFailed(CompletableFuture future, String key,
+ ServiceUnitStateData data, String dst) {
+ return future.whenComplete((__, e) -> {
+ if (e != null && !(e instanceof PulsarClientException.AlreadyClosedException)) {
+ log.warn("Failed to sync tableview item; sizes may diverge until the next update;"
+ + " dst={} serviceUnit={} data={}", dst, key, data, e);
+ }
+ });
}
private void syncExistingItems()
throws IOException, ExecutionException, InterruptedException, TimeoutException {
long started = System.currentTimeMillis();
+
@Cleanup
ServiceUnitStateTableView metadataStoreTableView = new ServiceUnitStateMetadataStoreTableViewImpl();
metadataStoreTableView.start(
pulsar,
- this::dummy,
- this::dummy,
- this::dummy
+ NOOP_CONSUMER,
+ NOOP_CONSUMER,
+ NOOP_CONSUMER
);
@Cleanup
ServiceUnitStateTableView systemTopicTableView = new ServiceUnitStateTableViewImpl();
systemTopicTableView.start(
pulsar,
- this::dummy,
- this::dummy,
- this::dummy
+ NOOP_CONSUMER,
+ NOOP_CONSUMER,
+ NOOP_CONSUMER
);
var syncer = pulsar.getConfiguration().getLoadBalancerServiceUnitTableViewSyncer();
+ ServiceUnitStateTableView src;
+ ServiceUnitStateTableView dst;
if (syncer == SystemTopicToMetadataStoreSyncer) {
clean(metadataStoreTableView);
syncExistingItemsToMetadataStore(systemTopicTableView);
+ src = systemTopicTableView;
+ dst = metadataStoreTableView;
} else {
clean(systemTopicTableView);
syncExistingItemsToSystemTopic(metadataStoreTableView, systemTopicTableView);
+ src = metadataStoreTableView;
+ dst = systemTopicTableView;
}
- if (!waitUntilSynced(metadataStoreTableView, systemTopicTableView, started)) {
+ if (!waitUntilSynced(src, dst, started)) {
throw new TimeoutException(
syncer + " failed to sync existing items in tableviews. MetadataStoreTableView.size: "
+ metadataStoreTableView.entrySet().size()
+ ", SystemTopicTableView.size: " + systemTopicTableView.entrySet().size() + " in "
- + SYNC_WAIT_TIME_IN_SECS + " secs");
+ + syncWaitTimeInSecs + " secs");
}
log.info("Synced existing items MetadataStoreTableView.size:{} , "
@@ -154,8 +189,8 @@ private void syncTailItems() throws InterruptedException, IOException, TimeoutEx
this.metadataStoreTableView.start(
pulsar,
this::syncToSystemTopic,
- this::dummy,
- this::dummy
+ NOOP_CONSUMER,
+ NOOP_CONSUMER
);
log.info("Started MetadataStoreTableView");
@@ -163,18 +198,20 @@ private void syncTailItems() throws InterruptedException, IOException, TimeoutEx
this.systemTopicTableView.start(
pulsar,
this::syncToMetadataStore,
- this::dummy,
- this::dummy
+ NOOP_CONSUMER,
+ NOOP_CONSUMER
);
log.info("Started SystemTopicTableView");
var syncer = pulsar.getConfiguration().getLoadBalancerServiceUnitTableViewSyncer();
- if (!waitUntilSynced(metadataStoreTableView, systemTopicTableView, started)) {
+ var src = syncer == SystemTopicToMetadataStoreSyncer ? systemTopicTableView : metadataStoreTableView;
+ var dst = syncer == SystemTopicToMetadataStoreSyncer ? metadataStoreTableView : systemTopicTableView;
+ if (!waitUntilSynced(src, dst, started)) {
throw new TimeoutException(
syncer + " failed to sync tableviews. MetadataStoreTableView.size: "
+ metadataStoreTableView.entrySet().size()
+ ", SystemTopicTableView.size: " + systemTopicTableView.entrySet().size() + " in "
- + SYNC_WAIT_TIME_IN_SECS + " secs");
+ + syncWaitTimeInSecs + " secs");
}
@@ -187,62 +224,134 @@ private void syncTailItems() throws InterruptedException, IOException, TimeoutEx
private void syncExistingItemsToMetadataStore(ServiceUnitStateTableView src)
throws JsonProcessingException, ExecutionException, InterruptedException, TimeoutException {
// Directly use store to sync existing items to metadataStoreTableView(otherwise, they are conflicted out)
- var store = pulsar.getLocalMetadataStore();
- var writer = ObjectMapperFactory.getMapper().writer();
- var opTimeout = pulsar.getConfiguration().getMetadataStoreOperationTimeoutSeconds();
List> futures = new ArrayList<>();
var srcIter = src.entrySet().iterator();
while (srcIter.hasNext()) {
var e = srcIter.next();
- futures.add(store.put(ServiceUnitStateMetadataStoreTableViewImpl.PATH_PREFIX + "/" + e.getKey(),
- writer.writeValueAsBytes(e.getValue()), Optional.empty()).thenApply(__ -> null));
- if (futures.size() == MAX_CONCURRENT_SYNC_COUNT || !srcIter.hasNext()) {
- FutureUtil.waitForAll(futures).get(opTimeout, TimeUnit.SECONDS);
- }
+ futures.add(writeToMetadataStore(e.getKey(), e.getValue()));
+ maybeWaitCompletion(futures, !srcIter.hasNext());
+ }
+ }
+
+ private void maybeWaitCompletion(List> futures, boolean forceWait)
+ throws InterruptedException, ExecutionException, TimeoutException {
+ if (!futures.isEmpty() && (futures.size() == MAX_CONCURRENT_SYNC_COUNT || forceWait)) {
+ FutureUtil.waitForAll(futures)
+ .get(pulsar.getConfiguration().getMetadataStoreOperationTimeoutSeconds(), TimeUnit.SECONDS);
+ futures.clear();
}
}
+ private @NonNull CompletableFuture writeToMetadataStore(String key, ServiceUnitStateData value)
+ throws JsonProcessingException {
+ return pulsar.getLocalMetadataStore().put(ServiceUnitStateMetadataStoreTableViewImpl.PATH_PREFIX + "/" + key,
+ jsonWriter.writeValueAsBytes(value), Optional.empty()).thenApply(__ -> null);
+ }
+
private void syncExistingItemsToSystemTopic(ServiceUnitStateTableView src,
ServiceUnitStateTableView dst)
throws ExecutionException, InterruptedException, TimeoutException {
- var opTimeout = pulsar.getConfiguration().getMetadataStoreOperationTimeoutSeconds();
List> futures = new ArrayList<>();
var srcIter = src.entrySet().iterator();
while (srcIter.hasNext()) {
var e = srcIter.next();
futures.add(dst.put(e.getKey(), e.getValue()));
- if (futures.size() == MAX_CONCURRENT_SYNC_COUNT || !srcIter.hasNext()) {
- FutureUtil.waitForAll(futures).get(opTimeout, TimeUnit.SECONDS);
- }
+ maybeWaitCompletion(futures, !srcIter.hasNext());
}
}
private void clean(ServiceUnitStateTableView dst)
throws ExecutionException, InterruptedException, TimeoutException {
- var opTimeout = pulsar.getConfiguration().getMetadataStoreOperationTimeoutSeconds();
var dstIter = dst.entrySet().iterator();
List> futures = new ArrayList<>();
while (dstIter.hasNext()) {
var e = dstIter.next();
futures.add(dst.delete(e.getKey()));
- if (futures.size() == MAX_CONCURRENT_SYNC_COUNT || !dstIter.hasNext()) {
- FutureUtil.waitForAll(futures).get(opTimeout, TimeUnit.SECONDS);
- }
+ maybeWaitCompletion(futures, !dstIter.hasNext());
}
}
- private boolean waitUntilSynced(ServiceUnitStateTableView srt, ServiceUnitStateTableView dst, long started)
+ private boolean waitUntilSynced(ServiceUnitStateTableView src, ServiceUnitStateTableView dst, long started)
throws InterruptedException {
- while (srt.entrySet().size() != dst.entrySet().size()) {
- if (TimeUnit.MILLISECONDS.toSeconds(System.currentTimeMillis() - started)
- > SYNC_WAIT_TIME_IN_SECS) {
+ long lastReconciled = started;
+ while (src.entrySet().size() != dst.entrySet().size()) {
+ long now = System.currentTimeMillis();
+ if (TimeUnit.MILLISECONDS.toSeconds(now - started) > syncWaitTimeInSecs) {
return false;
}
+ // Give in-flight syncs a grace period to settle on their own, then reconcile
+ // periodically: updates that raced with the table views' (re)start were replayed
+ // to the fresh views as existing items — which are deliberately not wired to
+ // sync — so without reconciliation the views would never converge.
+ if (now - lastReconciled >= RECONCILE_INTERVAL_IN_MILLIS) {
+ if (log.isDebugEnabled()) {
+ log.debug("Tableviews not synced yet; reconciling; srcSize={} dstSize={} elapsedSecs={}",
+ src.entrySet().size(), dst.entrySet().size(),
+ TimeUnit.MILLISECONDS.toSeconds(now - started));
+ }
+ reconcile(src, dst, started);
+ lastReconciled = now;
+ }
Thread.sleep(100);
}
return true;
}
+ /**
+ * Copies items the destination table view is missing and removes stale items that no longer
+ * exist in the source. Channel updates that land between the existing-items copy and the
+ * registration of the tail listeners are only visible as existing items of the freshly
+ * started views, so the tail listeners never see them. Writes flow to the migration source
+ * while the syncer starts, making the source view authoritative; destination-only items are
+ * removed only when they predate this sync phase and are still absent from the source, so a
+ * concurrent fresh write to the destination is never discarded. Failures are logged and left
+ * for the next reconcile pass. Runs on the caller's (load manager) thread with each batch
+ * bounded by the metadata store operation timeout.
+ */
+ private void reconcile(ServiceUnitStateTableView src, ServiceUnitStateTableView dst, long started)
+ throws InterruptedException {
+ // Snapshot the destination entries before iterating the source so that a key arriving
+ // in the destination through a concurrent tail sync cannot be misclassified as stale.
+ var staleDstItems = new HashMap();
+ for (var e : dst.entrySet()) {
+ staleDstItems.put(e.getKey(), e.getValue());
+ }
+ try {
+ List> futures = new ArrayList<>();
+ for (var e : src.entrySet()) {
+ if (staleDstItems.remove(e.getKey()) == null) {
+ log.info("Reconciling item missing from the destination tableview; serviceUnit={}",
+ e.getKey());
+ if (dst.isMetadataStoreBased()) {
+ // Write directly to the store like syncExistingItemsToMetadataStore
+ // does; the view's put() would conflict the item out.
+ futures.add(writeToMetadataStore(e.getKey(), e.getValue()));
+ } else {
+ futures.add(dst.put(e.getKey(), e.getValue()));
+ }
+ maybeWaitCompletion(futures, false);
+ }
+ }
+ for (var e : staleDstItems.entrySet()) {
+ // Only remove items written before this sync phase began and re-confirmed absent
+ // from the source: a fresh destination write (e.g. from a broker already switched
+ // to the destination implementation) is propagated to the source by the tail
+ // listener instead of being deleted here.
+ if (e.getValue().timestamp() < started && src.get(e.getKey()) == null) {
+ log.info("Reconciling stale item in the destination tableview; serviceUnit={}",
+ e.getKey());
+ futures.add(dst.delete(e.getKey()));
+ maybeWaitCompletion(futures, false);
+ }
+ }
+ maybeWaitCompletion(futures, true);
+ } catch (IOException | ExecutionException | TimeoutException e) {
+ // Transient write failures leave a size divergence behind; the next reconcile pass
+ // (or the sync-wait timeout) handles it.
+ log.warn("Failed to reconcile tableview items", e);
+ }
+ }
+
@Override
public void close() throws IOException {
if (!isActive) {
@@ -282,4 +391,9 @@ public void close() throws IOException {
public boolean isActive() {
return isActive;
}
+
+ @VisibleForTesting
+ public void setSyncWaitTimeInSecs(int syncWaitTimeInSecs) {
+ this.syncWaitTimeInSecs = syncWaitTimeInSecs;
+ }
}
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/ModularLoadManagerImpl.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/ModularLoadManagerImpl.java
index a9d7ddd78e07d..214d12e7ad2c8 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/ModularLoadManagerImpl.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/ModularLoadManagerImpl.java
@@ -381,46 +381,55 @@ private boolean checkBundleDataExistInNamespaceBundles(NamespaceBundles namespac
// Attempt to local the data for the given bundle in metadata store
// If it cannot be found, return the default bundle data.
+ /**
+ * @deprecated use {@link #getBundleDataOrDefaultAsync(String)} instead.
+ */
+ @Deprecated
@Override
public BundleData getBundleDataOrDefault(final String bundle) {
- BundleData bundleData = null;
- try {
- Optional optBundleData =
- pulsarResources.getLoadBalanceResources().getBundleDataResources().getBundleData(bundle).join();
- if (optBundleData.isPresent()) {
- return optBundleData.get();
- }
+ return getBundleDataOrDefaultAsync(bundle).join();
+ }
- Optional optQuota = pulsarResources.getLoadBalanceResources().getQuotaResources()
- .getQuota(bundle).join();
- if (optQuota.isPresent()) {
- ResourceQuota quota = optQuota.get();
- bundleData = new BundleData(NUM_SHORT_SAMPLES, NUM_LONG_SAMPLES);
- // Initialize from existing resource quotas if new API ZNodes do not exist.
- final TimeAverageMessageData shortTermData = bundleData.getShortTermData();
- final TimeAverageMessageData longTermData = bundleData.getLongTermData();
-
- shortTermData.setMsgRateIn(quota.getMsgRateIn());
- shortTermData.setMsgRateOut(quota.getMsgRateOut());
- shortTermData.setMsgThroughputIn(quota.getBandwidthIn());
- shortTermData.setMsgThroughputOut(quota.getBandwidthOut());
-
- longTermData.setMsgRateIn(quota.getMsgRateIn());
- longTermData.setMsgRateOut(quota.getMsgRateOut());
- longTermData.setMsgThroughputIn(quota.getBandwidthIn());
- longTermData.setMsgThroughputOut(quota.getBandwidthOut());
-
- // Assume ample history.
- shortTermData.setNumSamples(NUM_SHORT_SAMPLES);
- longTermData.setNumSamples(NUM_LONG_SAMPLES);
- }
- } catch (Exception e) {
- log.warn("Error when trying to find bundle {} on metadata store: {}", bundle, e);
- }
- if (bundleData == null) {
- bundleData = new BundleData(NUM_SHORT_SAMPLES, NUM_LONG_SAMPLES, defaultStats);
- }
- return bundleData;
+ @Override
+ public CompletableFuture getBundleDataOrDefaultAsync(final String bundle) {
+ return pulsarResources.getLoadBalanceResources().getBundleDataResources().getBundleData(bundle)
+ .thenCompose(optBundleData -> {
+ if (optBundleData.isPresent()) {
+ return CompletableFuture.completedFuture(optBundleData.get());
+ }
+ return pulsarResources.getLoadBalanceResources().getQuotaResources().getQuota(bundle)
+ .thenApply(optQuota -> {
+ if (optQuota.isEmpty()) {
+ return null;
+ }
+ ResourceQuota quota = optQuota.get();
+ BundleData bundleData = new BundleData(NUM_SHORT_SAMPLES, NUM_LONG_SAMPLES);
+ // Initialize from existing resource quotas if new API ZNodes do not exist.
+ final TimeAverageMessageData shortTermData = bundleData.getShortTermData();
+ final TimeAverageMessageData longTermData = bundleData.getLongTermData();
+
+ shortTermData.setMsgRateIn(quota.getMsgRateIn());
+ shortTermData.setMsgRateOut(quota.getMsgRateOut());
+ shortTermData.setMsgThroughputIn(quota.getBandwidthIn());
+ shortTermData.setMsgThroughputOut(quota.getBandwidthOut());
+
+ longTermData.setMsgRateIn(quota.getMsgRateIn());
+ longTermData.setMsgRateOut(quota.getMsgRateOut());
+ longTermData.setMsgThroughputIn(quota.getBandwidthIn());
+ longTermData.setMsgThroughputOut(quota.getBandwidthOut());
+
+ // Assume ample history.
+ shortTermData.setNumSamples(NUM_SHORT_SAMPLES);
+ longTermData.setNumSamples(NUM_LONG_SAMPLES);
+ return bundleData;
+ });
+ })
+ .exceptionally(e -> {
+ log.warn("Error when trying to find bundle {} on metadata store: {}", bundle, e);
+ return null;
+ })
+ .thenApply(bundleData -> bundleData != null ? bundleData
+ : new BundleData(NUM_SHORT_SAMPLES, NUM_LONG_SAMPLES, defaultStats));
}
// Use the Pulsar client to acquire the namespace bundle stats.
@@ -561,7 +570,7 @@ private void updateBundleData() {
} else {
// Otherwise, attempt to find the bundle data on metadata store.
// If it cannot be found, use the latest stats as the first sample.
- BundleData currentBundleData = getBundleDataOrDefault(bundle);
+ BundleData currentBundleData = getBundleDataOrDefaultAsync(bundle).join();
currentBundleData.update(stats);
bundleData.put(bundle, currentBundleData);
}
@@ -879,7 +888,7 @@ public Optional selectBrokerForAssignment(final ServiceUnitId serviceUni
private void preallocateBundle(String bundle, String broker) {
final BundleData data = loadData.getBundleData().computeIfAbsent(bundle,
- key -> getBundleDataOrDefault(bundle));
+ key -> getBundleDataOrDefaultAsync(bundle).join());
loadData.getBrokerData().get(broker).getPreallocatedBundleData().put(bundle, data);
preallocatedBundleToBroker.put(bundle, broker);
@@ -893,7 +902,7 @@ Optional selectBroker(final ServiceUnitId serviceUnit) {
synchronized (brokerCandidateCache) {
final String bundle = serviceUnit.toString();
final BundleData data = loadData.getBundleData().computeIfAbsent(bundle,
- key -> getBundleDataOrDefault(bundle));
+ key -> getBundleDataOrDefaultAsync(bundle).join());
brokerCandidateCache.clear();
LoadManagerShared.applyNamespacePolicies(serviceUnit, policies, brokerCandidateCache,
getAvailableBrokers(),
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/ModularLoadManagerWrapper.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/ModularLoadManagerWrapper.java
index c8d81bda1bc13..b5bb4d52717f9 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/ModularLoadManagerWrapper.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/ModularLoadManagerWrapper.java
@@ -64,8 +64,7 @@ public LoadManagerReport generateLoadReport() {
@Override
public Optional getLeastLoaded(final ServiceUnitId serviceUnit) {
- String bundleRange = LoadManagerShared.getBundleRangeFromBundleName(serviceUnit.toString());
- String affinityBroker = loadManager.setNamespaceBundleAffinity(bundleRange, null);
+ String affinityBroker = loadManager.setNamespaceBundleAffinity(serviceUnit.toString(), null);
if (!StringUtils.isBlank(affinityBroker)) {
return Optional.of(buildBrokerResourceUnit(affinityBroker));
}
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 c16c9338bf994..8a14b16f37339 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
@@ -61,7 +61,6 @@
import org.apache.pulsar.broker.PulsarServerException;
import org.apache.pulsar.broker.PulsarService;
import org.apache.pulsar.broker.ServiceConfiguration;
-import org.apache.pulsar.broker.loadbalance.LeaderBroker;
import org.apache.pulsar.broker.loadbalance.LeaderElectionService;
import org.apache.pulsar.broker.loadbalance.LoadManager;
import org.apache.pulsar.broker.loadbalance.ResourceUnit;
@@ -561,7 +560,6 @@ public CompletableFuture getHeartbeatOrSLAMonitorBrokerId(
private void searchForCandidateBroker(NamespaceBundle bundle,
CompletableFuture> lookupFuture,
LookupOptions options) {
- String candidateBroker;
LeaderElectionService les = pulsar.getLeaderElectionService();
if (les == null) {
LOG.warn("The leader election has not yet been completed! NamespaceBundle[{}]", bundle);
@@ -570,67 +568,98 @@ private void searchForCandidateBroker(NamespaceBundle bundle,
return;
}
- boolean authoritativeRedirect = les.isLeader();
+ selectCandidateBroker(bundle, options, les)
+ .thenAcceptAsync(selection -> {
+ if (selection.isEmpty()) {
+ LOG.warn("Load manager didn't return any available broker. "
+ + "Returning empty result to lookup. NamespaceBundle[{}]",
+ bundle);
+ lookupFuture.complete(Optional.empty());
+ return;
+ }
+ acquireOwnershipOrRedirect(bundle, options, selection.get(), lookupFuture);
+ }, pulsar.getExecutor())
+ .exceptionally(e -> {
+ LOG.warn("Error when searching for candidate broker to acquire {}: {}", bundle, e.getMessage(), e);
+ lookupFuture.completeExceptionally(FutureUtil.unwrapCompletionException(e));
+ return null;
+ });
+ }
- try {
- // check if this is Heartbeat or SLAMonitor namespace
- candidateBroker = getHeartbeatOrSLAMonitorBrokerId(bundle, cb ->
- CompletableFuture.completedFuture(isBrokerActive(cb)))
- .get(config.getMetadataStoreOperationTimeoutSeconds(), SECONDS);
+ /** The broker selected for a bundle assignment, and whether the redirect to it is authoritative. */
+ private record CandidateBrokerSelection(String candidateBroker, boolean authoritativeRedirect) { }
- if (candidateBroker == null) {
- Optional currentLeader = pulsar.getLeaderElectionService().getCurrentLeader();
+ private CompletableFuture> selectCandidateBroker(
+ NamespaceBundle bundle, LookupOptions options, LeaderElectionService les) {
+ boolean authoritativeRedirect = les.isLeader();
- if (options.isAuthoritative()) {
- // leader broker already assigned the current broker as owner
- candidateBroker = pulsar.getBrokerId();
- } else {
+ // check if this is Heartbeat or SLAMonitor namespace
+ return getHeartbeatOrSLAMonitorBrokerId(bundle, cb ->
+ CompletableFuture.completedFuture(isBrokerActive(cb)))
+ .thenComposeAsync(heartbeatOrSlaBroker -> {
+ if (heartbeatOrSlaBroker != null) {
+ return completedSelection(heartbeatOrSlaBroker, authoritativeRedirect);
+ }
+ if (options.isAuthoritative()) {
+ // leader broker already assigned the current broker as owner
+ return completedSelection(pulsar.getBrokerId(), authoritativeRedirect);
+ }
LoadManager loadManager = this.loadManager.get();
- boolean makeLoadManagerDecisionOnThisBroker = !loadManager.isCentralized() || les.isLeader();
- if (!makeLoadManagerDecisionOnThisBroker) {
- // If leader is not active, fallback to pick the least loaded from current broker loadmanager
+ if (!loadManager.isCentralized() || les.isLeader()) {
+ return selectLeastLoadedBroker(bundle);
+ }
+ // The load manager decision belongs to the leader: read the leader
+ // authoritatively (waits for an in-progress election to settle) instead of
+ // acting on a possibly-empty snapshot during a leadership handoff.
+ return les.readCurrentLeader().thenComposeAsync(currentLeader -> {
boolean leaderBrokerActive = currentLeader.isPresent()
&& isBrokerActive(currentLeader.get().getBrokerId());
- if (!leaderBrokerActive) {
- makeLoadManagerDecisionOnThisBroker = true;
- if (currentLeader.isEmpty()) {
- LOG.warn(
- "The information about the current leader broker wasn't available. "
- + "Handling load manager decisions in a decentralized way. "
- + "NamespaceBundle[{}]",
- bundle);
- } else {
- LOG.warn(
- "The current leader broker {} isn't active. "
- + "Handling load manager decisions in a decentralized way. "
- + "NamespaceBundle[{}]",
- currentLeader.get(), bundle);
- }
+ if (leaderBrokerActive) {
+ // forward to leader broker to make assignment
+ return completedSelection(currentLeader.get().getBrokerId(), authoritativeRedirect);
}
- }
- if (makeLoadManagerDecisionOnThisBroker) {
- Optional availableBroker = getLeastLoadedFromLoadManager(bundle);
- if (availableBroker.isEmpty()) {
- LOG.warn("Load manager didn't return any available broker. "
- + "Returning empty result to lookup. NamespaceBundle[{}]",
+ // If leader is not active, fallback to pick the least loaded from current broker loadmanager
+ if (currentLeader.isEmpty()) {
+ LOG.warn(
+ "The information about the current leader broker wasn't available. "
+ + "Handling load manager decisions in a decentralized way. "
+ + "NamespaceBundle[{}]",
bundle);
- lookupFuture.complete(Optional.empty());
- return;
+ } else {
+ LOG.warn(
+ "The current leader broker {} isn't active. "
+ + "Handling load manager decisions in a decentralized way. "
+ + "NamespaceBundle[{}]",
+ currentLeader.get(), bundle);
}
- candidateBroker = availableBroker.get();
- authoritativeRedirect = true;
- } else {
- // forward to leader broker to make assignment
- candidateBroker = currentLeader.get().getBrokerId();
- }
- }
- }
+ return selectLeastLoadedBroker(bundle);
+ }, pulsar.getExecutor());
+ }, pulsar.getExecutor());
+ }
+
+ private static CompletableFuture> completedSelection(
+ String candidateBroker, boolean authoritativeRedirect) {
+ return CompletableFuture.completedFuture(
+ Optional.of(new CandidateBrokerSelection(candidateBroker, authoritativeRedirect)));
+ }
+
+ // The decentralized decision is authoritative: this broker picked the owner itself.
+ private CompletableFuture> selectLeastLoadedBroker(NamespaceBundle bundle) {
+ Optional availableBroker;
+ try {
+ availableBroker = getLeastLoadedFromLoadManager(bundle);
} catch (Exception e) {
- LOG.warn("Error when searching for candidate broker to acquire {}: {}", bundle, e.getMessage(), e);
- lookupFuture.completeExceptionally(e);
- return;
+ return CompletableFuture.failedFuture(e);
}
+ return CompletableFuture.completedFuture(
+ availableBroker.map(broker -> new CandidateBrokerSelection(broker, true)));
+ }
+ private void acquireOwnershipOrRedirect(NamespaceBundle bundle, LookupOptions options,
+ CandidateBrokerSelection selection,
+ CompletableFuture> lookupFuture) {
+ final String candidateBroker = selection.candidateBroker();
+ final boolean authoritativeRedirect = selection.authoritativeRedirect();
try {
Objects.requireNonNull(candidateBroker);
@@ -1378,6 +1407,25 @@ public CompletableFuture> getOwnedTopicListForNamespaceBundle(Names
});
}
+ public CompletableFuture> getOwnedPersistentTopicListForNamespaceBundle(NamespaceBundle bundle) {
+ return getListOfPersistentTopics(bundle.getNamespaceObject()).thenCompose(topics ->
+ CompletableFuture.completedFuture(
+ topics.stream()
+ .filter(topic -> bundle.includes(TopicName.get(topic)))
+ .collect(Collectors.toList())))
+ .thenCombine(getPartitions(bundle.getNamespaceObject(), TopicDomain.persistent).thenCompose(topics ->
+ CompletableFuture.completedFuture(
+ topics.stream().filter(topic -> bundle.includes(TopicName.get(topic)))
+ .collect(Collectors.toList()))), (left, right) -> {
+ for (String topic : right) {
+ if (!left.contains(topic)) {
+ left.add(topic);
+ }
+ }
+ return left;
+ });
+ }
+
/***
* Checks whether the topic exists( partitioned or non-partitioned ).
*/
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractBaseDispatcher.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractBaseDispatcher.java
index c5e001692f2f3..c4d99f1f6039d 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractBaseDispatcher.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractBaseDispatcher.java
@@ -481,7 +481,8 @@ protected void checkAndApplyReachedEndOfTopicOrTopicMigration(List con
public static void checkAndApplyReachedEndOfTopicOrTopicMigration(PersistentTopic topic, List consumers) {
if (topic.isMigrated()) {
- consumers.forEach(c -> c.topicMigrated(topic.getMigratedClusterUrl()));
+ topic.getMigratedClusterUrlAsync()
+ .thenAccept(clusterUrl -> consumers.forEach(c -> c.topicMigrated(clusterUrl)));
} else {
consumers.forEach(Consumer::reachedEndOfTopic);
}
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractReplicator.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractReplicator.java
index bb1d1e08f12a5..0d41d7353ade8 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractReplicator.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractReplicator.java
@@ -85,6 +85,11 @@ public abstract class AbstractReplicator implements Replicator {
private static final AtomicReferenceFieldUpdater ATTRIBUTES_UPDATER =
AtomicReferenceFieldUpdater.newUpdater(AbstractReplicator.class, Attributes.class, "attributes");
+ protected volatile long latestPublishTime = System.currentTimeMillis();
+
+ // The estimated time when the producer connection is successful, "0" means it will be connected immediately.
+ protected volatile long estimatedTimeStampProducerConnected = 0;
+
public enum State {
/**
* This enum has two mean meanings：
@@ -169,7 +174,7 @@ protected CompletableFuture prepareCreateProducer() {
return CompletableFuture.completedFuture(null);
}
- public void startProducer() {
+ protected void startProducer() {
// Guarantee only one task call "producerBuilder.createAsync()".
Pair setStartingRes = compareSetAndGetState(State.Disconnected, State.Starting);
if (!setStartingRes.getLeft()) {
@@ -203,12 +208,14 @@ public void startProducer() {
builderImpl.getConf().setNonPartitionedTopicExpected(true);
builderImpl.getConf().setReplProducer(true);
return producerBuilder.createAsync().thenAccept(producer -> {
+ estimatedTimeStampProducerConnected = 0;
setProducerAndTriggerReadEntries(producer);
});
}).exceptionally(ex -> {
Pair setDisconnectedRes = compareSetAndGetState(State.Starting, State.Disconnected);
if (setDisconnectedRes.getLeft()) {
long waitTimeMs = backOff.next().toMillis();
+ estimatedTimeStampProducerConnected = System.currentTimeMillis() + waitTimeMs;
log.warn("[{}] Failed to create remote producer ({}), retrying in {} s",
replicatorId, ex.getMessage(), waitTimeMs / 1000.0);
// BackOff before retrying
@@ -311,8 +318,7 @@ protected CompletableFuture isLocalTopicActive() {
/**
* This method only be used by {@link PersistentTopic#checkGC} now.
*/
- @Override
- public CompletableFuture disconnect() {
+ protected CompletableFuture disconnect() {
long backlog = getNumberOfEntriesInBacklog();
if (backlog > 0) {
CompletableFuture disconnectFuture = new CompletableFuture<>();
@@ -390,8 +396,6 @@ protected CompletableFuture closeProducerAsync(boolean closeTheStartingPro
Pair setDisconnectedRes = compareSetAndGetState(State.Disconnecting, State.Disconnected);
if (setDisconnectedRes.getLeft()) {
this.producer = null;
- // deactivate further read
- disableReplicatorRead();
return;
}
if (setDisconnectedRes.getRight() == State.Terminating
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractTopic.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractTopic.java
index 8be1002b3a8db..c7ea8bf8ec969 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractTopic.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractTopic.java
@@ -43,11 +43,13 @@
import java.util.concurrent.CompletionException;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedQueue;
+import java.util.concurrent.ExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
import java.util.concurrent.atomic.AtomicLongFieldUpdater;
import java.util.concurrent.atomic.LongAdder;
import java.util.concurrent.locks.ReentrantReadWriteLock;
+import java.util.function.Function;
import java.util.function.ToLongFunction;
import lombok.Getter;
import lombok.Setter;
@@ -58,6 +60,7 @@
import org.apache.commons.lang3.tuple.Pair;
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.resourcegroup.ResourceGroup;
import org.apache.pulsar.broker.resourcegroup.ResourceGroupPublishLimiter;
import org.apache.pulsar.broker.resources.NamespaceResources;
@@ -66,6 +69,7 @@
import org.apache.pulsar.broker.service.BrokerServiceException.ProducerFencedException;
import org.apache.pulsar.broker.service.BrokerServiceException.TopicMigratedException;
import org.apache.pulsar.broker.service.BrokerServiceException.TopicTerminatedException;
+import org.apache.pulsar.broker.service.persistent.PersistentReplicator;
import org.apache.pulsar.broker.service.persistent.PersistentTopic;
import org.apache.pulsar.broker.service.plugin.EntryFilter;
import org.apache.pulsar.broker.service.schema.SchemaRegistryService;
@@ -115,6 +119,10 @@ public abstract class AbstractTopic implements Topic, TopicPolicyListener {
protected final BrokerService brokerService;
+ // Wraps this topic as a TopicPolicyListener so topic-policy updates received while the initial policy is still
+ // loading are buffered and applied in order once initTopicPolicy() completes initialization.
+ protected final TopicPolicyListenerWrapper topicPolicyListener = new TopicPolicyListenerWrapper(this);
+
// Prefix for replication cursors
protected final String replicatorPrefix;
@@ -174,7 +182,17 @@ public abstract class AbstractTopic implements Topic, TopicPolicyListener {
AtomicLongFieldUpdater.newUpdater(AbstractTopic.class, "usageCount");
private volatile long usageCount = 0;
- private Map subscriptionPolicies = Collections.emptyMap();
+ // Effective per-subscription policies, merged from the local and global topic policies with local precedence.
+ // Unlike the PolicyHierarchyValue-backed fields, subscriptionPolicies is a plain map. It is kept as the merge of
+ // the two scopes below (rather than assigned directly) because the local-before-global initialization order
+ // (see TopicPolicyListenerWrapper) would otherwise let the global map -- empty by default in TopicPolicies --
+ // overwrite and clear the local per-subscription policies that were applied just before it.
+ // subscriptionPolicies is volatile because it is read on dispatch threads (getSubscriptionDispatchRate) while it
+ // is updated on the policy-update thread. localSubscriptionPolicies/globalSubscriptionPolicies are only ever read
+ // and written on the (single) policy-update thread, so they don't need to be volatile.
+ private volatile Map subscriptionPolicies = Collections.emptyMap();
+ private Map localSubscriptionPolicies = Collections.emptyMap();
+ private Map globalSubscriptionPolicies = Collections.emptyMap();
protected final LongAdder msgOutFromRemovedSubscriptions = new LongAdder();
protected final LongAdder bytesOutFromRemovedSubscriptions = new LongAdder();
@@ -185,10 +203,16 @@ public abstract class AbstractTopic implements Topic, TopicPolicyListener {
protected final Clock clock;
protected Set additionalSystemCursorNames = new TreeSet<>();
+ private final ExecutorService topicPoliciesNotifyThread;
public AbstractTopic(String topic, BrokerService brokerService) {
this.topic = topic;
this.namespace = TopicName.get(topic).getNamespaceObject();
+ // Pin the per-topic policies-notify thread once. BrokerService#getTopicPoliciesNotifyThread centralizes
+ // the topic-to-thread mapping so it stays consistent with SystemTopicBasedTopicPoliciesService. In unit
+ // tests that construct topics with a mock BrokerService this returns null (the thread is unused there).
+ this.topicPoliciesNotifyThread =
+ brokerService.getTopicPoliciesNotifyThread(TopicName.getPartitionedTopicName(topic));
this.clock = brokerService.getClock();
this.brokerService = brokerService;
this.producers = new ConcurrentHashMap<>();
@@ -307,11 +331,36 @@ protected void updateTopicPolicy(TopicPolicies data) {
topicPolicies.getEntryFilters().updateTopicValue(data.getEntryFilters(), isGlobalPolicies);
topicPolicies.getDispatcherPauseOnAckStatePersistentEnabled()
.updateTopicValue(data.getDispatcherPauseOnAckStatePersistentEnabled(), isGlobalPolicies);
- this.subscriptionPolicies = data.getSubscriptionPolicies();
+
+ // Merge instead of assigning directly: keep the local and global per-subscription policies separately and
+ // recompute the effective map with local precedence, so applying the (default-empty) global map does not
+ // clear the local per-subscription policies during the local-before-global initialization.
+ if (isGlobalPolicies) {
+ globalSubscriptionPolicies = data.getSubscriptionPolicies();
+ } else {
+ localSubscriptionPolicies = data.getSubscriptionPolicies();
+ }
+ subscriptionPolicies = mergeSubscriptionPolicies(globalSubscriptionPolicies, localSubscriptionPolicies);
updateEntryFilters();
}
+ // Merges the global and local per-subscription policies with local precedence: a subscription present in the
+ // local policies keeps its local value; otherwise the global value (if any) is used.
+ private static Map mergeSubscriptionPolicies(
+ Map globalSubscriptionPolicies,
+ Map localSubscriptionPolicies) {
+ if (globalSubscriptionPolicies.isEmpty()) {
+ return localSubscriptionPolicies;
+ }
+ if (localSubscriptionPolicies.isEmpty()) {
+ return globalSubscriptionPolicies;
+ }
+ Map merged = new HashMap<>(globalSubscriptionPolicies);
+ merged.putAll(localSubscriptionPolicies);
+ return merged;
+ }
+
protected void updateTopicPolicyByNamespacePolicy(Policies namespacePolicies) {
if (log.isDebugEnabled()) {
log.debug("[{}]updateTopicPolicyByNamespacePolicy,data={}", topic, namespacePolicies);
@@ -550,14 +599,67 @@ protected boolean isProducersExceeded(boolean isRemote) {
&& maxProducers <= USER_CREATED_PRODUCER_COUNTER_UPDATER.get(this);
}
- protected void registerTopicPolicyListener() {
- brokerService.getPulsar().getTopicPoliciesService()
- .registerListener(TopicName.getPartitionedTopicName(topic), this);
+ protected TopicPolicyListener getTopicPolicyListener() {
+ return topicPolicyListener;
}
protected void unregisterTopicPolicyListener() {
brokerService.getPulsar().getTopicPoliciesService()
- .unregisterListener(TopicName.getPartitionedTopicName(topic), this);
+ .unregisterListener(TopicName.getPartitionedTopicName(topic), getTopicPolicyListener());
+ }
+
+ /**
+ * Registers the topic-policy listener and applies the topic's initial policies (global and local) to this topic.
+ * Shared by {@link org.apache.pulsar.broker.service.persistent.PersistentTopic} and
+ * {@link org.apache.pulsar.broker.service.nonpersistent.NonPersistentTopic} so both load their own policies on
+ * topic load, which removes the need to broadcast every topic's policy when a namespace's policy cache finishes
+ * loading (see {@code topicPolicyListenerReplayEnabled}).
+ *
+ * Each call re-initializes the listener wrapper and, whatever the outcome, always completes its initialization
+ * afterwards, so the wrapper never stays in the buffering phase (dropping updates) even if policy loading fails.
+ * This makes the method safe to run again (e.g. a future retry); runs are expected to be serialized.
+ */
+ protected CompletableFuture initTopicPolicy() {
+ final var topicPoliciesService = brokerService.getPulsar().getTopicPoliciesService();
+ final var partitionedTopicName = TopicName.getPartitionedTopicName(topic);
+
+ // Begin a fresh initialization phase: updates are buffered until initialization completes below. This resets
+ // any previous phase so the method can be run again.
+ topicPolicyListener.startInitialization();
+ CompletableFuture initTopicPolicyFuture =
+ topicPoliciesService.registerListenerAsync(partitionedTopicName, topicPolicyListener)
+ .thenCompose(registered -> {
+ if (!registered) {
+ return CompletableFuture.completedFuture(null);
+ }
+ if (ExtensibleLoadManagerImpl.isInternalTopic(topic)) {
+ // Internal topics don't load topic-level policies
+ return CompletableFuture.completedFuture(null);
+ }
+ // future for fetching global topic policies
+ CompletableFuture> globalPoliciesFuture =
+ topicPoliciesService.getTopicPoliciesAsync(partitionedTopicName,
+ TopicPoliciesService.GetType.GLOBAL_ONLY);
+ // future for fetching local topic policies
+ CompletableFuture> localPoliciesFuture =
+ topicPoliciesService.getTopicPoliciesAsync(partitionedTopicName,
+ TopicPoliciesService.GetType.LOCAL_ONLY);
+ return globalPoliciesFuture.thenCombine(localPoliciesFuture, (global, local) -> {
+ // finally update the topic policies with the latest value or loaded value
+ return CompletableFuture.runAsync(() ->
+ topicPolicyListener.completeInitialization(global.orElse(null),
+ local.orElse(null)),
+ getPoliciesNotifyThread());
+ }).thenCompose(Function.identity());
+ });
+ // Whatever the outcome -- success, failure, or the listener not being registered -- make sure the wrapper
+ // leaves the initialization (buffering) phase, so it forwards any buffered value plus all future live updates
+ // instead of dropping them. This is a no-op when the loaded policies were already applied above. Return the
+ // whenComplete stage (not initTopicPolicyFuture) so the returned future completes only after this has run, and
+ // whenComplete's pass-through semantics carry the original success or failure to the caller's initialize().
+ return initTopicPolicyFuture.whenCompleteAsync((v, ex) -> {
+ topicPolicyListener.completeInitializationUnlessAlreadyCompleted();
+ }, getPoliciesNotifyThread());
}
protected boolean isSameAddressProducersExceeded(Producer producer) {
@@ -658,6 +760,19 @@ protected Consumer getActiveConsumer(Subscription subscription) {
return null;
}
+ protected boolean hasProducersActive() {
+ return !producers.isEmpty();
+ }
+
+ protected boolean hasActiveReplicators() {
+ for (Replicator replicator : getReplicators().values()) {
+ if (replicator.isConnected()) {
+ return true;
+ }
+ }
+ return false;
+ }
+
protected boolean hasLocalProducers() {
if (producers.isEmpty()) {
return false;
@@ -670,6 +785,19 @@ protected boolean hasLocalProducers() {
return false;
}
+ public void disconnectReplicatorsIfNoTrafficAndBacklog() {
+ for (Replicator replicator : getReplicators().values()) {
+ if (replicator instanceof PersistentReplicator persistentReplicator) {
+ persistentReplicator.disconnectIfNoTrafficAndBacklog();
+ }
+ }
+ for (Replicator replicator : getShadowReplicators().values()) {
+ if (replicator instanceof PersistentReplicator persistentReplicator) {
+ persistentReplicator.disconnectIfNoTrafficAndBacklog();
+ }
+ }
+ }
+
@Override
public String toString() {
return MoreObjects.toStringHelper(this).add("topic", topic).toString();
@@ -1338,8 +1466,8 @@ public void updateBrokerSubscribeRate() {
subscribeRateInBroker(brokerService.pulsar().getConfiguration()));
}
- public Optional getMigratedClusterUrl() {
- return getMigratedClusterUrl(brokerService.getPulsar(), topic);
+ public CompletableFuture> getMigratedClusterUrlAsync() {
+ return getMigratedClusterUrlAsync(brokerService.getPulsar(), topic);
}
public static CompletableFuture isClusterMigrationEnabled(PulsarService pulsar,
@@ -1379,16 +1507,6 @@ private static CompletableFuture isNamespaceMigrationEnabledAsync(Pulsa
.thenApply(policies -> policies.isPresent() && policies.get().migrated);
}
- public static Optional getMigratedClusterUrl(PulsarService pulsar, String topic) {
- try {
- return getMigratedClusterUrlAsync(pulsar, topic)
- .get(pulsar.getPulsarResources().getClusterResources().getOperationTimeoutSec(), TimeUnit.SECONDS);
- } catch (Exception e) {
- log.warn("[{}] Failed to get migration cluster URL", topic, e);
- }
- return Optional.empty();
- }
-
public boolean isSystemCursor(String sub) {
return COMPACTION_SUBSCRIPTION.equals(sub)
|| (additionalSystemCursorNames != null && additionalSystemCursorNames.contains(sub));
@@ -1451,4 +1569,8 @@ private static Map getNonPartitionedPropertiesAsync(PulsarServic
return Collections.emptyMap();
}
}
+
+ protected ExecutorService getPoliciesNotifyThread() {
+ return topicPoliciesNotifyThread;
+ }
}
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BacklogQuotaManager.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BacklogQuotaManager.java
index f04570c5441cd..b5a0d5877fed3 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BacklogQuotaManager.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BacklogQuotaManager.java
@@ -19,6 +19,7 @@
package org.apache.pulsar.broker.service;
import static java.util.concurrent.TimeUnit.SECONDS;
+import com.google.common.annotations.VisibleForTesting;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
@@ -33,12 +34,14 @@
import org.apache.bookkeeper.mledger.proto.ManagedLedgerInfo;
import org.apache.pulsar.broker.PulsarService;
import org.apache.pulsar.broker.resources.NamespaceResources;
+import org.apache.pulsar.broker.service.persistent.PersistentSubscription;
import org.apache.pulsar.broker.service.persistent.PersistentTopic;
import org.apache.pulsar.broker.service.persistent.PersistentTopicMetrics.BacklogQuotaMetrics;
import org.apache.pulsar.common.naming.NamespaceName;
import org.apache.pulsar.common.policies.data.BacklogQuota;
import org.apache.pulsar.common.policies.data.BacklogQuota.BacklogQuotaType;
import org.apache.pulsar.common.policies.data.impl.BacklogQuotaImpl;
+import org.apache.pulsar.common.util.Codec;
import org.apache.pulsar.common.util.FutureUtil;
import org.apache.pulsar.metadata.api.MetadataStoreException;
@@ -85,6 +88,17 @@ public BacklogQuotaImpl getBacklogQuota(NamespaceName namespace, BacklogQuotaTyp
*/
public void handleExceededBacklogQuota(PersistentTopic persistentTopic, BacklogQuotaType backlogQuotaType,
boolean preciseTimeBasedBacklogQuotaCheck) {
+ if (persistentTopic.isFenced() || persistentTopic.isClosingOrDeleting()) {
+ // Skip eviction work on a topic that is being torn down or transiently fenced.
+ // Mutating cursors here (skipEntries / markDeletePosition) contends with the
+ // delete path and can keep namespace force-delete from completing in time;
+ // the entries are about to be discarded anyway.
+ if (log.isDebugEnabled()) {
+ log.debug("Skipping backlog-quota eviction on fenced/closing topic {}, backlog quota type {}",
+ persistentTopic.getName(), backlogQuotaType);
+ }
+ return;
+ }
BacklogQuota quota = persistentTopic.getBacklogQuota(backlogQuotaType);
BacklogQuotaMetrics topicBacklogQuotaMetrics =
persistentTopic.getPersistentTopicMetrics().getBacklogQuotaMetrics();
@@ -126,9 +140,7 @@ public void handleExceededBacklogQuota(PersistentTopic persistentTopic, BacklogQ
* Backlog quota set for the topic
*/
private void dropBacklogForSizeLimit(PersistentTopic persistentTopic, BacklogQuota quota) {
- // Set the reduction factor to 90%. The aim is to drop down the backlog to 90% of the quota limit.
- double reductionFactor = 0.9;
- double targetSize = reductionFactor * quota.getLimitSize();
+ long targetSize = computeEvictionTarget(quota.getLimitSize());
// Get estimated unconsumed size for the managed ledger associated with this topic. Estimated size is more
// useful than the actual storage size. Actual storage size gets updated only when managed ledger is trimmed.
@@ -137,7 +149,7 @@ private void dropBacklogForSizeLimit(PersistentTopic persistentTopic, BacklogQuo
if (log.isDebugEnabled()) {
log.debug("[{}] target size is [{}] for quota limit [{}], backlog size is [{}]", persistentTopic.getName(),
- targetSize, targetSize / reductionFactor, backlogSize);
+ targetSize, quota.getLimitSize(), backlogSize);
}
ManagedCursor previousSlowestConsumer = null;
while (backlogSize > targetSize) {
@@ -154,13 +166,17 @@ private void dropBacklogForSizeLimit(PersistentTopic persistentTopic, BacklogQuo
if (slowestConsumer == previousSlowestConsumer) {
log.info("[{}] Cursors not progressing, target size is [{}] for quota limit [{}], backlog size is [{}]",
- persistentTopic.getName(), targetSize, targetSize / reductionFactor, backlogSize);
+ persistentTopic.getName(), targetSize, quota.getLimitSize(), backlogSize);
break;
}
// Calculate number of messages to be skipped using the current backlog and the skip factor.
long entriesInBacklog = slowestConsumer.getNumberOfEntriesInBacklog(false);
- int messagesToSkip = (int) (messageSkipFactor * entriesInBacklog);
+
+ int messagesToSkip = computeEntriesToEvict(
+ backlogSize,
+ quota.getLimitSize(),
+ entriesInBacklog);
try {
// If there are no messages to skip, break out of the loop
if (messagesToSkip == 0) {
@@ -175,6 +191,7 @@ private void dropBacklogForSizeLimit(PersistentTopic persistentTopic, BacklogQuo
persistentTopic.getName(), messagesToSkip, slowestConsumer.getName(), entriesInBacklog);
}
slowestConsumer.skipEntries(messagesToSkip, IndividualDeletedEntries.Include);
+ markDeletePositionMoveForward(persistentTopic, slowestConsumer);
} catch (Exception e) {
log.error("[{}] Error skipping [{}] messages from slowest consumer [{}]", persistentTopic.getName(),
messagesToSkip, slowestConsumer.getName(), e);
@@ -202,9 +219,7 @@ private void dropBacklogForTimeLimit(PersistentTopic persistentTopic, BacklogQuo
boolean preciseTimeBasedBacklogQuotaCheck) {
// If enabled precise time based backlog quota check, will expire message based on the timeBaseQuota
if (preciseTimeBasedBacklogQuotaCheck) {
- // Set the reduction factor to 90%. The aim is to drop down the backlog to 90% of the quota limit.
- double reductionFactor = 0.9;
- int target = (int) (reductionFactor * quota.getLimitTime());
+ int target = (int) computeEvictionTarget(quota.getLimitTime());
if (log.isDebugEnabled()) {
log.debug("[{}] target backlog expire time is [{}]", persistentTopic.getName(), target);
}
@@ -233,6 +248,7 @@ private void dropBacklogForTimeLimit(PersistentTopic persistentTopic, BacklogQuo
long ledgerId = mLedger.getLedgersInfo().ceilingKey(oldestPosition.getLedgerId() + 1);
Position nextPosition = PositionFactory.create(ledgerId, -1);
slowestConsumer.markDelete(nextPosition);
+ markDeletePositionMoveForward(persistentTopic, slowestConsumer);
continue;
}
// Timestamp only > 0 if ledger has been closed
@@ -243,6 +259,7 @@ private void dropBacklogForTimeLimit(PersistentTopic persistentTopic, BacklogQuo
Position nextPosition = PositionFactory.create(ledgerId, -1);
if (!nextPosition.equals(oldestPosition)) {
slowestConsumer.markDelete(nextPosition);
+ markDeletePositionMoveForward(persistentTopic, slowestConsumer);
continue;
}
}
@@ -303,4 +320,48 @@ private boolean advanceSlowestSystemCursor(PersistentTopic persistentTopic) {
// We may need to check other system cursors here : replicator, compaction
return false;
}
+
+ /**
+ * Invoke {@link Dispatcher#markDeletePositionMoveForward()} for the subscription that owns the given cursor.
+ * This ensures pending acks and redelivery state are cleaned up when the cursor is advanced by
+ * backlog quota eviction (bypassing the subscription-level wrappers that normally fire this hook).
+ *
+ * @param persistentTopic the topic
+ * @param cursor the cursor that was advanced
+ */
+ private void markDeletePositionMoveForward(PersistentTopic persistentTopic, ManagedCursor cursor) {
+ PersistentSubscription subscription =
+ persistentTopic.getSubscriptions().get(Codec.decode(cursor.getName()));
+ if (subscription != null && subscription.getDispatcher() != null) {
+ subscription.getDispatcher().markDeletePositionMoveForward();
+ }
+ }
+
+
+ /**
+ * Compute the target value after backlog eviction.
+ *
+ * @param quotaLimit configured quota limit
+ * @return target value after eviction
+ */
+ private static long computeEvictionTarget(long quotaLimit) {
+ double factor = 0.9;
+ return (long) (factor * quotaLimit);
+ }
+
+ /**
+ * Compute the number of entries to evict in a single eviction iteration.
+ *
+ * @param currentValue current backlog value
+ * @param quotaLimit configured quota limit
+ * @param totalEntries total entries in backlog
+ * @return entries to evict
+ */
+ @VisibleForTesting
+ static int computeEntriesToEvict(
+ long currentValue, long quotaLimit, long totalEntries) {
+ long evictionTarget = computeEvictionTarget(quotaLimit);
+ return (int) ((currentValue - evictionTarget)
+ * (double) totalEntries / currentValue);
+ }
}
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java
index d6226254f3c45..73b8aef9d2873 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java
@@ -67,6 +67,7 @@
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.Semaphore;
@@ -284,6 +285,7 @@ public class BrokerService implements Closeable {
.register();
private final SingleThreadNonConcurrentFixedRateScheduler inactivityMonitor;
+ @Getter
private final SingleThreadNonConcurrentFixedRateScheduler messageExpiryMonitor;
private final SingleThreadNonConcurrentFixedRateScheduler compactionMonitor;
private final SingleThreadNonConcurrentFixedRateScheduler consumedLedgersMonitor;
@@ -723,6 +725,10 @@ protected void startInactivityMonitor() {
int interval = pulsar().getConfiguration().getBrokerDeleteInactiveTopicsFrequencySeconds();
inactivityMonitor.scheduleAtFixedRateNonConcurrently(() -> checkGC(), interval, interval,
TimeUnit.SECONDS);
+ if (pulsar().getConfig().getBrokerReplicationInactiveThresholdSeconds() > 0) {
+ inactivityMonitor.scheduleAtFixedRateNonConcurrently(() -> checkInactiveReplication(), interval,
+ interval, TimeUnit.SECONDS);
+ }
}
// Deduplication info checker
@@ -2328,6 +2334,14 @@ public void checkGC() {
forEachTopic(Topic::checkGC);
}
+ public void checkInactiveReplication() {
+ forEachTopic(topic -> {
+ if (topic instanceof AbstractTopic abstractTopic) {
+ abstractTopic.disconnectReplicatorsIfNoTrafficAndBacklog();
+ }
+ });
+ }
+
public void checkClusterMigration() {
forEachTopic(Topic::checkClusterMigration);
}
@@ -2666,9 +2680,10 @@ private void handleLocalPoliciesUpdates(NamespaceName namespace) {
log.debug("Notifying topic that local policies have changed: {}", name);
}
topic.ifPresent(t -> {
- if (t instanceof PersistentTopic) {
- PersistentTopic topic1 = (PersistentTopic) t;
- topic1.onLocalPoliciesUpdate();
+ if (t instanceof PersistentTopic persistentTopic) {
+ runOnTopicPoliciesNotifyThread(t, () -> {
+ persistentTopic.onLocalPoliciesUpdate();
+ });
}
});
});
@@ -2688,7 +2703,8 @@ private void handlePoliciesUpdates(NamespaceName namespace) {
log.info("[{}] updating with {}", namespace, policies);
topics.forEach((name, topicFuture) -> {
- if (namespace.includes(TopicName.get(name))) {
+ TopicName topicName = TopicName.get(name);
+ if (namespace.includes(topicName)) {
// If the topic is already created, immediately apply the updated policies, otherwise
// once the topic is created it'll apply the policies update
topicFuture.thenAccept(topic -> {
@@ -2696,7 +2712,11 @@ private void handlePoliciesUpdates(NamespaceName namespace) {
log.debug("Notifying topic that policies have changed: {}", name);
}
- topic.ifPresent(t -> t.onPoliciesUpdate(policies));
+ topic.ifPresent(t -> {
+ runOnTopicPoliciesNotifyThread(t, () -> {
+ t.onPoliciesUpdate(policies);
+ });
+ });
});
}
});
@@ -2707,6 +2727,16 @@ private void handlePoliciesUpdates(NamespaceName namespace) {
}, pulsar.getExecutor());
}
+ private void runOnTopicPoliciesNotifyThread(Topic t, Runnable runnable) {
+ ExecutorService policiesNotifyThread;
+ if (t instanceof AbstractTopic abstractTopic) {
+ policiesNotifyThread = abstractTopic.getPoliciesNotifyThread();
+ } else {
+ policiesNotifyThread = getTopicPoliciesNotifyThread(TopicName.getPartitionedTopicName(t.getName()));
+ }
+ policiesNotifyThread.execute(runnable);
+ }
+
private void handleDynamicConfigurationUpdates() {
DynamicConfigurationResources dynamicConfigResources = null;
try {
@@ -3569,6 +3599,24 @@ public OrderedExecutor getTopicOrderedExecutor() {
return topicOrderedExecutor;
}
+ /**
+ * Returns the single executor thread used to apply topic-policy updates for the given topic. All
+ * topic-policy notifications and policy application for a topic must run on this deterministically-chosen
+ * thread so that they are serialized and never run concurrently. Centralizing the topic-to-thread mapping
+ * here keeps it consistent between {@link AbstractTopic} and
+ * {@link SystemTopicBasedTopicPoliciesService} so the two cannot accidentally diverge.
+ */
+ public ExecutorService getTopicPoliciesNotifyThread(TopicName topicName) {
+ TopicName baseTopicName;
+ if (topicName.isPartitioned()) {
+ // for partitioned topics, we need to use the base topic name
+ baseTopicName = TopicName.get(topicName.getPartitionedTopicName());
+ } else {
+ baseTopicName = topicName;
+ }
+ return topicOrderedExecutor.chooseThread(baseTopicName);
+ }
+
/**
* If per-broker unacked message reached to limit then it blocks dispatcher if its unacked message limit has been
* reached to {@link #maxUnackedMsgsPerDispatcher}.
@@ -4029,14 +4077,14 @@ public boolean isCurrentClusterAllowed(NamespaceName nsName, Policies nsPolicies
return nsPolicies.replication_clusters.contains(pulsar.getConfig().getClusterName());
}
- public void setCurrentClusterAllowedIfNoClusterIsAllowed(NamespaceName nsName, Policies nsPolicies) {
- if (nsPolicies.replication_clusters.contains(pulsar.getConfig().getClusterName())
- || nsPolicies.allowed_clusters.contains(pulsar.getConfig().getClusterName())) {
- return;
- }
+ public void setCurrentClusterAllowedWhenCreating(NamespaceName nsName, Policies nsPolicies) {
if (nsPolicies.replication_clusters.isEmpty()) {
nsPolicies.replication_clusters.add(pulsar.getConfig().getClusterName());
- } else {
+ }
+ if (nsPolicies.allowed_clusters.isEmpty()) {
+ return;
+ }
+ if (!nsPolicies.allowed_clusters.contains(pulsar.getConfig().getClusterName())) {
nsPolicies.allowed_clusters.add(pulsar.getConfig().getClusterName());
}
}
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Consumer.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Consumer.java
index 3edf6e2d68140..f7fca52065d9c 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Consumer.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Consumer.java
@@ -253,6 +253,7 @@ public Consumer(Subscription subscription, SubType subType, String topicName, lo
OPEN_TELEMETRY_ATTRIBUTES_FIELD_UPDATER.set(this, null);
}
+
@VisibleForTesting
Consumer(String consumerName, int availablePermits) {
this.subscription = null;
@@ -378,9 +379,18 @@ public Future sendMessages(final List entries,
} else {
stickyKeyHash = stickyKeyHashes.get(i);
}
- boolean sendingAllowed =
- pendingAcks.addPendingAckIfAllowed(entry.getLedgerId(), entry.getEntryId(), batchSize,
- stickyKeyHash);
+ boolean sendingAllowed;
+ long[] ackSet = batchIndexesAcks == null ? null : batchIndexesAcks.getAckSet(i);
+ int remainingUnacked;
+ if (ackSet != null) {
+ remainingUnacked = BitSet.valueOf(ackSet).cardinality();
+ unackedMessages -= (batchSize - remainingUnacked);
+ } else {
+ remainingUnacked = batchSize;
+ }
+ sendingAllowed =
+ pendingAcks.addPendingAckIfAllowed(entry.getLedgerId(), entry.getEntryId(),
+ remainingUnacked, stickyKeyHash);
if (!sendingAllowed) {
// sending isn't allowed when pending acks doesn't accept adding the entry
// this happens when Key_Shared draining hashes contains the stickyKeyHash
@@ -395,10 +405,6 @@ public Future sendMessages(final List entries,
consumerId);
}
} else {
- long[] ackSet = batchIndexesAcks == null ? null : batchIndexesAcks.getAckSet(i);
- if (ackSet != null) {
- unackedMessages -= (batchSize - BitSet.valueOf(ackSet).cardinality());
- }
if (log.isDebugEnabled()) {
log.debug("[{}-{}] Added {}:{} ledger entry with batchSize of {} to pendingAcks in"
+ " broker.service.Consumer for consumerId: {}",
@@ -578,7 +584,7 @@ private CompletableFuture individualAckNormal(CommandAck ack, Map ackOwnerConsumerAndBatchSize =
getAckOwnerConsumerAndBatchSize(msgId.getLedgerId(), msgId.getEntryId());
Consumer ackOwnerConsumer = ackOwnerConsumerAndBatchSize.left();
- long ackedCount;
+ long ackedCount = 0;
int batchSize = ackOwnerConsumerAndBatchSize.rightInt();
if (msgId.getAckSetsCount() > 0) {
long[] ackSets = new long[msgId.getAckSetsCount()];
@@ -594,11 +600,19 @@ private CompletableFuture individualAckNormal(CommandAck ack, Map 0) {
+ boolean updated = ackOwnerConsumer.updateRemainingUnacked(
+ position.getLedgerId(), position.getEntryId(), (int) ackedCount);
+ if (updated) {
+ addAndGetUnAckedMsgs(ackOwnerConsumer, -(int) ackedCount);
+ }
+ }
} else {
position = PositionFactory.create(msgId.getLedgerId(), msgId.getEntryId());
- ackedCount = getAckedCountForMsgIdNoAckSets(batchSize, position, ackOwnerConsumer);
- if (checkCanRemovePendingAcksAndHandle(ackOwnerConsumer, position, msgId)) {
+ IntIntPair removed = ackOwnerConsumer.removePendingAckAndGet(
+ position.getLedgerId(), position.getEntryId());
+ if (removed != null) {
+ ackedCount = removed.leftInt();
addAndGetUnAckedMsgs(ackOwnerConsumer, -(int) ackedCount);
updateBlockedConsumerOnUnackedMsgs(ackOwnerConsumer);
}
@@ -676,12 +690,22 @@ private CompletableFuture individualAckWithTransaction(CommandAck ack) {
}
AckSetStateUtil.getAckSetState(position).setAckSet(ackSets);
ackedCount = getAckedCountForTransactionAck(batchSize, ackSets);
+ if (ackedCount > 0) {
+ boolean updated = ackOwnerConsumer.updateRemainingUnacked(
+ position.getLedgerId(), position.getEntryId(), (int) ackedCount);
+ if (updated) {
+ addAndGetUnAckedMsgs(ackOwnerConsumer, -(int) ackedCount);
+ }
+ }
+ } else {
+ IntIntPair removed = ackOwnerConsumer.removePendingAckAndGet(
+ position.getLedgerId(), position.getEntryId());
+ if (removed != null) {
+ addAndGetUnAckedMsgs(ackOwnerConsumer, -removed.leftInt());
+ updateBlockedConsumerOnUnackedMsgs(ackOwnerConsumer);
+ }
}
- addAndGetUnAckedMsgs(ackOwnerConsumer, -(int) ackedCount);
-
- checkCanRemovePendingAcksAndHandle(ackOwnerConsumer, position, msgId);
-
checkAckValidationError(ack, position);
totalAckCount.add(ackedCount);
@@ -705,16 +729,6 @@ private CompletableFuture individualAckWithTransaction(CommandAck ack) {
return completableFuture.thenApply(__ -> totalAckCount.sum());
}
- private long getAckedCountForMsgIdNoAckSets(int batchSize, Position position, Consumer consumer) {
- if (isAcknowledgmentAtBatchIndexLevelEnabled && Subscription.isIndividualAckMode(subType)) {
- long[] cursorAckSet = getCursorAckSet(position);
- if (cursorAckSet != null) {
- return getAckedCountForBatchIndexLevelEnabled(position, batchSize, EMPTY_ACK_SET, consumer);
- }
- }
- return batchSize;
- }
-
private long getAckedCountForBatchIndexLevelEnabled(Position position, int batchSize, long[] ackSets,
Consumer consumer) {
long ackedCount = 0;
@@ -744,19 +758,6 @@ private long getAckedCountForTransactionAck(int batchSize, long[] ackSets) {
return ackedCount;
}
- private long getUnAckedCountForBatchIndexLevelEnabled(Position position, int batchSize) {
- long unAckedCount = batchSize;
- if (isAcknowledgmentAtBatchIndexLevelEnabled) {
- long[] cursorAckSet = getCursorAckSet(position);
- if (cursorAckSet != null) {
- BitSetRecyclable cursorBitSet = BitSetRecyclable.create().resetWords(cursorAckSet);
- unAckedCount = cursorBitSet.cardinality();
- cursorBitSet.recycle();
- }
- }
- return unAckedCount;
- }
-
private void checkAckValidationError(CommandAck ack, Position position) {
if (ack.hasValidationError()) {
log.warn("[{}] [{}] Received ack for corrupted message at {} - Reason: {}", subscription,
@@ -764,14 +765,6 @@ private void checkAckValidationError(CommandAck ack, Position position) {
}
}
- private boolean checkCanRemovePendingAcksAndHandle(Consumer ackOwnedConsumer,
- Position position, MessageIdData msgId) {
- if (Subscription.isIndividualAckMode(subType) && msgId.getAckSetsCount() == 0) {
- return removePendingAcks(ackOwnedConsumer, position);
- }
- return false;
- }
-
/**
* Retrieves the acknowledgment owner consumer and batch size for the specified ledgerId and entryId.
*
@@ -921,20 +914,16 @@ public void topicMigrated(Optional clusterUrl) {
}
}
- public boolean checkAndApplyTopicMigration() {
- if (subscription.isSubscriptionMigrated()) {
- Optional clusterUrl = AbstractTopic.getMigratedClusterUrl(cnx.getBrokerService().getPulsar(),
- topicName);
- if (clusterUrl.isPresent()) {
- ClusterUrl url = clusterUrl.get();
- cnx.getCommandSender().sendTopicMigrated(ResourceType.Consumer, consumerId, url.getBrokerServiceUrl(),
- url.getBrokerServiceUrlTls());
- // disconnect consumer after sending migrated cluster url
- disconnect();
- return true;
- }
+ public CompletableFuture