This is an automated email from the ASF dual-hosted git repository.
yubiao pushed a commit to branch branch-3.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
The following commit(s) were added to refs/heads/branch-3.0 by this push:
new 87c29701330 [fix] [broker] Update topic policies as much as possible
when some ex was thrown (#21810)
87c29701330 is described below
commit 87c29701330d96602eb49fa860a5c56e0650be87
Author: fengyubiao <[email protected]>
AuthorDate: Wed Jan 3 21:51:26 2024 +0800
[fix] [broker] Update topic policies as much as possible when some ex was
thrown (#21810)
After the topic policies update, there are many components will be updated
one by one, even if the config of components has not been modified. There are
the 11 components that need update:
- `7` rate limiters(`publish`, `dispatch topic-level`, `dispatch
subscription-level`, `dispatch resourceGroup-level`, `subscribe API`,
`replication`, `shadow topic replication`)
- update ManagedLedger configs(`retention`, `offloader`)
- start/stop replication
- start/stop compaction
- start/stop deduplication
Once a component update fails, the following update will be skipped. It
would cause a confusing thing: you want to set a retention policy, but it will
be skipped due to the `update subscribe rate limiter` failure (you did not edit
the `subscribe rate limitation policy`)
Since none of the components in the above list have any additional
dependencies for individual updates, ensuring success as much as possible is
appropriate.
- Update topic policies as much as possible even if some component updates
fail, all component updates are still in the same thread, and they still update
one by one, just throw the error later.
- Rename `updatePublishDispatcher` to `updatePublishRateLimiter`
(cherry picked from commit ed599673c7e60ab5bb02e1fb0615a7ff8e5d6430)
---
.../pulsar/broker/service/AbstractTopic.java | 2 +-
.../pulsar/broker/service/BrokerService.java | 2 +-
.../service/nonpersistent/NonPersistentTopic.java | 2 +-
.../broker/service/persistent/PersistentTopic.java | 117 +++++++++++----------
.../pulsar/broker/admin/TopicPoliciesTest.java | 51 +++++++++
.../org/apache/pulsar/common/util/FutureUtil.java | 6 ++
6 files changed, 123 insertions(+), 57 deletions(-)
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 9a5771e18ce..507a5af5270 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
@@ -1286,7 +1286,7 @@ public abstract class AbstractTopic implements Topic,
TopicPolicyListener<TopicP
/**
* update topic publish dispatcher for this topic.
*/
- public void updatePublishDispatcher() {
+ public void updatePublishRateLimiter() {
synchronized (topicPublishRateLimiterLock) {
PublishRate publishRate = topicPolicies.getPublishRate().get();
if (publishRate.publishThrottlingRateInByte > 0 ||
publishRate.publishThrottlingRateInMsg > 0) {
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 c87402f9253..08283488b66 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
@@ -2796,7 +2796,7 @@ public class BrokerService implements Closeable {
forEachTopic(topic -> {
if (topic instanceof AbstractTopic) {
((AbstractTopic) topic).updateBrokerPublishRate();
- ((AbstractTopic) topic).updatePublishDispatcher();
+ ((AbstractTopic) topic).updatePublishRateLimiter();
}
}));
}
diff --git
a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentTopic.java
b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentTopic.java
index cd09f187368..17b6bd8f350 100644
---
a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentTopic.java
+++
b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentTopic.java
@@ -179,7 +179,7 @@ public class NonPersistentTopic extends AbstractTopic
implements Topic, TopicPol
isEncryptionRequired = policies.encryption_required;
isAllowAutoUpdateSchema =
policies.is_allow_auto_update_schema;
}
- updatePublishDispatcher();
+ updatePublishRateLimiter();
updateResourceGroupLimiter(policies);
return updateClusterMigrated();
});
diff --git
a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java
b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java
index 7ca57ac8657..d521f6609a8 100644
---
a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java
+++
b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java
@@ -357,7 +357,7 @@ public class PersistentTopic extends AbstractTopic
implements Topic, AddEntryCal
.thenAcceptAsync(optPolicies -> {
if (!optPolicies.isPresent()) {
isEncryptionRequired = false;
- updatePublishDispatcher();
+ updatePublishRateLimiter();
updateResourceGroupLimiter(new Policies());
initializeDispatchRateLimiterIfNeeded();
updateSubscribeRateLimiter();
@@ -372,7 +372,7 @@ public class PersistentTopic extends AbstractTopic
implements Topic, AddEntryCal
updateSubscribeRateLimiter();
- updatePublishDispatcher();
+ updatePublishRateLimiter();
updateResourceGroupLimiter(policies);
@@ -2818,39 +2818,60 @@ public class PersistentTopic extends AbstractTopic
implements Topic, AddEntryCal
return CompletableFuture.completedFuture(null);
}
+ // Update props.
+ // The component "EntryFilters" is update in the method
"updateTopicPolicyByNamespacePolicy(data)".
+ // see more detail: https://github.com/apache/pulsar/pull/19364.
updateTopicPolicyByNamespacePolicy(data);
checkReplicatedSubscriptionControllerState();
isEncryptionRequired = data.encryption_required;
-
isAllowAutoUpdateSchema = data.is_allow_auto_update_schema;
- updateDispatchRateLimiter();
-
- updateSubscribeRateLimiter();
+ // Apply policies for components.
+ List<CompletableFuture<Void>> applyPolicyTasks =
applyUpdatedTopicPolicies();
+ applyPolicyTasks.add(applyUpdatedNamespacePolicies(data));
+ return FutureUtil.waitForAll(applyPolicyTasks)
+ .thenAccept(__ -> log.info("[{}] namespace-level policies updated
successfully", topic))
+ .exceptionally(ex -> {
+ log.error("[{}] update namespace polices : {} error",
this.getName(), data, ex);
+ throw FutureUtil.wrapToCompletionException(ex);
+ });
+ }
- updatePublishDispatcher();
+ private CompletableFuture<Void> applyUpdatedNamespacePolicies(Policies
namespaceLevelPolicies) {
+ return FutureUtil.runWithCurrentThread(() ->
updateResourceGroupLimiter(namespaceLevelPolicies));
+ }
- updateResourceGroupLimiter(data);
+ private List<CompletableFuture<Void>> applyUpdatedTopicPolicies() {
+ List<CompletableFuture<Void>> applyPoliciesFutureList = new
ArrayList<>();
- List<CompletableFuture<Void>> producerCheckFutures = new
ArrayList<>(producers.size());
- producers.values().forEach(producer -> producerCheckFutures.add(
+ // Client permission check.
+ subscriptions.forEach((subName, sub) -> {
+ sub.getConsumers().forEach(consumer ->
applyPoliciesFutureList.add(consumer.checkPermissionsAsync()));
+ });
+ producers.values().forEach(producer -> applyPoliciesFutureList.add(
producer.checkPermissionsAsync().thenRun(producer::checkEncryption)));
+ // Check message expiry.
+ applyPoliciesFutureList.add(FutureUtil.runWithCurrentThread(() ->
checkMessageExpiry()));
- return FutureUtil.waitForAll(producerCheckFutures).thenCompose((__) ->
{
- return
updateSubscriptionsDispatcherRateLimiter().thenCompose((___) -> {
- replicators.forEach((name, replicator) ->
replicator.updateRateLimiter());
- shadowReplicators.forEach((name, replicator) ->
replicator.updateRateLimiter());
- checkMessageExpiry();
- CompletableFuture<Void> replicationFuture =
checkReplicationAndRetryOnFailure();
- CompletableFuture<Void> dedupFuture =
checkDeduplicationStatus();
- CompletableFuture<Void> persistentPoliciesFuture =
checkPersistencePolicies();
- return CompletableFuture.allOf(replicationFuture, dedupFuture,
persistentPoliciesFuture,
- preCreateSubscriptionForCompactionIfNeeded());
- });
- }).exceptionally(ex -> {
- log.error("[{}] update namespace polices : {} error",
this.getName(), data, ex);
- throw FutureUtil.wrapToCompletionException(ex);
- });
+ // Update rate limiters.
+ applyPoliciesFutureList.add(FutureUtil.runWithCurrentThread(() ->
updateDispatchRateLimiter()));
+ applyPoliciesFutureList.add(FutureUtil.runWithCurrentThread(() ->
updateSubscribeRateLimiter()));
+ applyPoliciesFutureList.add(FutureUtil.runWithCurrentThread(() ->
updatePublishRateLimiter()));
+
+ applyPoliciesFutureList.add(FutureUtil.runWithCurrentThread(() ->
updateSubscriptionsDispatcherRateLimiter()));
+ applyPoliciesFutureList.add(FutureUtil.runWithCurrentThread(
+ () -> replicators.forEach((name, replicator) ->
replicator.updateRateLimiter())));
+ applyPoliciesFutureList.add(FutureUtil.runWithCurrentThread(
+ () -> shadowReplicators.forEach((name, replicator) ->
replicator.updateRateLimiter())));
+
+ // Other components.
+ applyPoliciesFutureList.add(FutureUtil.runWithCurrentThread(() ->
checkReplicationAndRetryOnFailure()));
+ applyPoliciesFutureList.add(FutureUtil.runWithCurrentThread(() ->
checkDeduplicationStatus()));
+ applyPoliciesFutureList.add(FutureUtil.runWithCurrentThread(() ->
checkPersistencePolicies()));
+ applyPoliciesFutureList.add(FutureUtil.runWithCurrentThread(
+ () -> preCreateSubscriptionForCompactionIfNeeded()));
+
+ return applyPoliciesFutureList;
}
/**
@@ -3493,42 +3514,30 @@ public class PersistentTopic extends AbstractTopic
implements Topic, AddEntryCal
if (policies == null) {
return;
}
+ // Update props.
+ // The component "EntryFilters" is update in the method
"updateTopicPolicy(data)".
+ // see more detail: https://github.com/apache/pulsar/pull/19364.
updateTopicPolicy(policies);
shadowTopics = policies.getShadowTopics();
- updateDispatchRateLimiter();
checkReplicatedSubscriptionControllerState();
- updateSubscriptionsDispatcherRateLimiter().thenRun(() -> {
- updatePublishDispatcher();
- updateSubscribeRateLimiter();
- replicators.forEach((name, replicator) ->
replicator.updateRateLimiter());
- shadowReplicators.forEach((name, replicator) ->
replicator.updateRateLimiter());
- checkMessageExpiry();
- })
- .thenCompose(__ -> checkReplicationAndRetryOnFailure())
- .thenCompose(__ -> checkDeduplicationStatus())
- .thenCompose(__ -> preCreateSubscriptionForCompactionIfNeeded())
- .thenCompose(__ -> checkPersistencePolicies())
- .thenAccept(__ -> log.info("[{}] Policies updated successfully",
topic))
- .exceptionally(e -> {
- Throwable t = FutureUtil.unwrapCompletionException(e);
- log.error("[{}] update topic policy error: {}", topic,
t.getMessage(), t);
- return null;
- });
+
+ // Apply policies for components(not contains the specified policies
which only defined in namespace policies).
+ FutureUtil.waitForAll(applyUpdatedTopicPolicies())
+ .thenAccept(__ -> log.info("[{}] topic-level policies updated
successfully", topic))
+ .exceptionally(e -> {
+ Throwable t = FutureUtil.unwrapCompletionException(e);
+ log.error("[{}] update topic-level policy error: {}", topic,
t.getMessage(), t);
+ return null;
+ });
}
- private CompletableFuture<Void> updateSubscriptionsDispatcherRateLimiter()
{
- List<CompletableFuture<Void>> subscriptionCheckFutures = new
ArrayList<>((int) subscriptions.size());
+ private void updateSubscriptionsDispatcherRateLimiter() {
subscriptions.forEach((subName, sub) -> {
- List<CompletableFuture<Void>> consumerCheckFutures = new
ArrayList<>(sub.getConsumers().size());
- sub.getConsumers().forEach(consumer ->
consumerCheckFutures.add(consumer.checkPermissionsAsync()));
-
subscriptionCheckFutures.add(FutureUtil.waitForAll(consumerCheckFutures).thenRun(()
-> {
- Dispatcher dispatcher = sub.getDispatcher();
- if (dispatcher != null) {
- dispatcher.updateRateLimiter();
- }
- }));
+ Dispatcher dispatcher = sub.getDispatcher();
+ if (dispatcher != null) {
+ dispatcher.updateRateLimiter();
+ }
});
- return FutureUtil.waitForAll(subscriptionCheckFutures);
}
protected CompletableFuture<Void> initTopicPolicy() {
diff --git
a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/TopicPoliciesTest.java
b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/TopicPoliciesTest.java
index dc11de96bce..fe0901bf2d5 100644
---
a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/TopicPoliciesTest.java
+++
b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/TopicPoliciesTest.java
@@ -42,6 +42,8 @@ import java.util.concurrent.TimeUnit;
import lombok.Cleanup;
import lombok.extern.slf4j.Slf4j;
import org.apache.bookkeeper.mledger.ManagedLedgerConfig;
+import org.apache.bookkeeper.mledger.impl.ManagedLedgerImpl;
+import org.apache.pulsar.broker.BrokerTestUtil;
import org.apache.pulsar.broker.ConfigHelper;
import org.apache.pulsar.broker.auth.MockedPulsarServiceBaseTest;
import org.apache.pulsar.broker.namespace.NamespaceService;
@@ -50,6 +52,7 @@ import
org.apache.pulsar.broker.service.PublishRateLimiterImpl;
import org.apache.pulsar.broker.service.SystemTopicBasedTopicPoliciesService;
import org.apache.pulsar.broker.service.Topic;
import org.apache.pulsar.broker.service.persistent.DispatchRateLimiter;
+import org.apache.pulsar.broker.service.persistent.PersistentSubscription;
import org.apache.pulsar.broker.service.persistent.PersistentTopic;
import org.apache.pulsar.broker.service.persistent.SubscribeRateLimiter;
import org.apache.pulsar.client.admin.PulsarAdminException;
@@ -83,8 +86,11 @@ import org.apache.pulsar.common.policies.data.SubscribeRate;
import org.apache.pulsar.common.policies.data.TenantInfoImpl;
import org.apache.pulsar.common.policies.data.TopicPolicies;
import org.apache.pulsar.common.policies.data.TopicStats;
+import org.apache.pulsar.common.util.collections.ConcurrentOpenHashMap;
import org.assertj.core.api.Assertions;
import org.awaitility.Awaitility;
+import org.awaitility.reflect.WhiteboxImpl;
+import org.mockito.Mockito;
import org.testng.Assert;
import org.testng.annotations.AfterMethod;
import org.testng.annotations.BeforeMethod;
@@ -3161,4 +3167,49 @@ public class TopicPoliciesTest extends
MockedPulsarServiceBaseTest {
});
}
+ @Test
+ public void testUpdateRetentionWithPartialFailure() throws Exception {
+ String tpName = BrokerTestUtil.newUniqueName("persistent://" +
myNamespace + "/tp");
+ admin.topics().createNonPartitionedTopic(tpName);
+
+ // Load topic up.
+ admin.topics().getInternalStats(tpName);
+
+ // Inject an error that makes dispatch rate update fail.
+ PersistentTopic persistentTopic =
+ (PersistentTopic) pulsar.getBrokerService().getTopic(tpName,
false).join().get();
+ ConcurrentOpenHashMap<String, PersistentSubscription> subscriptions =
+ WhiteboxImpl.getInternalState(persistentTopic,
"subscriptions");
+ PersistentSubscription mockedSubscription =
Mockito.mock(PersistentSubscription.class);
+ Mockito.when(mockedSubscription.getDispatcher()).thenThrow(new
RuntimeException("Mocked error: getDispatcher"));
+ subscriptions.put("mockedSubscription", mockedSubscription);
+
+ // Update namespace-level retention policies.
+ RetentionPolicies retentionPolicies1 = new RetentionPolicies(1, 1);
+ admin.namespaces().setRetentionAsync(myNamespace, retentionPolicies1);
+
+ // Verify: update retention will be success even if other component
update throws exception.
+ Awaitility.await().untilAsserted(() -> {
+ ManagedLedgerImpl ML = (ManagedLedgerImpl)
persistentTopic.getManagedLedger();
+ assertEquals(ML.getConfig().getRetentionSizeInMB(), 1);
+ assertEquals(ML.getConfig().getRetentionTimeMillis(), 1 * 60 *
1000);
+ });
+
+ // Update topic-level retention policies.
+ RetentionPolicies retentionPolicies2 = new RetentionPolicies(2, 2);
+ admin.topics().setRetentionAsync(tpName, retentionPolicies2);
+
+ // Verify: update retention will be success even if other component
update throws exception.
+ Awaitility.await().untilAsserted(() -> {
+ ManagedLedgerImpl ML = (ManagedLedgerImpl)
persistentTopic.getManagedLedger();
+ assertEquals(ML.getConfig().getRetentionSizeInMB(), 2);
+ assertEquals(ML.getConfig().getRetentionTimeMillis(), 2 * 60 *
1000);
+ });
+
+ // Cleanup.
+ subscriptions.clear();
+ admin.namespaces().removeRetention(myNamespace);
+ admin.topics().delete(tpName, false);
+ }
+
}
diff --git
a/pulsar-common/src/main/java/org/apache/pulsar/common/util/FutureUtil.java
b/pulsar-common/src/main/java/org/apache/pulsar/common/util/FutureUtil.java
index 2b082b4a789..6f625898535 100644
--- a/pulsar-common/src/main/java/org/apache/pulsar/common/util/FutureUtil.java
+++ b/pulsar-common/src/main/java/org/apache/pulsar/common/util/FutureUtil.java
@@ -18,6 +18,7 @@
*/
package org.apache.pulsar.common.util;
+import com.google.common.util.concurrent.MoreExecutors;
import java.time.Duration;
import java.util.ArrayList;
import java.util.Collection;
@@ -55,6 +56,11 @@ public class FutureUtil {
return CompletableFuture.allOf(futures.toArray(new
CompletableFuture[0]));
}
+ public static CompletableFuture<Void> runWithCurrentThread(Runnable
runnable) {
+ return CompletableFuture.runAsync(
+ () -> runnable.run(), MoreExecutors.directExecutor());
+ }
+
public static <T> CompletableFuture<List<T>>
waitForAll(Stream<CompletableFuture<List<T>>> futures) {
return futures.reduce(CompletableFuture.completedFuture(new
ArrayList<>()),
(pre, curr) -> pre.thenCompose(preV -> curr.thenApply(currV ->
{
