From 15542f700d7f08a5d0cea7424c67f3dbe168e571 Mon Sep 17 00:00:00 2001 From: fengyubiao Date: Wed, 3 Jan 2024 21:51:26 +0800 Subject: [PATCH] [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 +- .../nonpersistent/NonPersistentTopic.java | 2 +- .../service/persistent/PersistentTopic.java | 122 +++++++++--------- .../broker/admin/TopicPoliciesTest.java | 51 ++++++++ .../apache/pulsar/common/util/FutureUtil.java | 6 + 6 files changed, 118 insertions(+), 67 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 472a972c0d7b4..7f5c30a729fe7 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 @@ -1222,7 +1222,7 @@ protected boolean isExceedMaximumMessageSize(int size, PublishContext publishCon /** * 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 0ab8cffeb362c..6860ab57a688e 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 @@ -2449,7 +2449,7 @@ private void updateMaxPublishRatePerTopicInMessages() { 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 a021e7ce94def..e3ecd39b71898 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 @@ -169,7 +169,7 @@ public CompletableFuture initialize() { isAllowAutoUpdateSchema = policies.is_allow_auto_update_schema; schemaValidationEnforced = policies.schema_validation_enforced; } - updatePublishDispatcher(); + updatePublishRateLimiter(); updateResourceGroupLimiter(policies); }); } 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 7563386058355..cf6b82294dc02 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 @@ -318,7 +318,7 @@ public CompletableFuture initialize() { .thenAccept(optPolicies -> { if (!optPolicies.isPresent()) { isEncryptionRequired = false; - updatePublishDispatcher(); + updatePublishRateLimiter(); updateResourceGroupLimiter(new Policies()); initializeDispatchRateLimiterIfNeeded(); updateSubscribeRateLimiter(); @@ -333,7 +333,7 @@ public CompletableFuture initialize() { updateSubscribeRateLimiter(); - updatePublishDispatcher(); + updatePublishRateLimiter(); updateResourceGroupLimiter(policies); @@ -2447,44 +2447,58 @@ public CompletableFuture onPoliciesUpdate(@Nonnull Policies data) { 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; - schemaValidationEnforced = data.schema_validation_enforced; + // Apply policies for components. + List> 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); + }); + } - updateDispatchRateLimiter(); + private CompletableFuture applyUpdatedNamespacePolicies(Policies namespaceLevelPolicies) { + return FutureUtil.runWithCurrentThread(() -> updateResourceGroupLimiter(namespaceLevelPolicies)); + } - updateSubscribeRateLimiter(); + private List> applyUpdatedTopicPolicies() { + List> applyPoliciesFutureList = new ArrayList<>(); - updatePublishDispatcher(); + // 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())); - updateResourceGroupLimiter(data); + // Update rate limiters. + applyPoliciesFutureList.add(FutureUtil.runWithCurrentThread(() -> updateDispatchRateLimiter())); + applyPoliciesFutureList.add(FutureUtil.runWithCurrentThread(() -> updateSubscribeRateLimiter())); + applyPoliciesFutureList.add(FutureUtil.runWithCurrentThread(() -> updatePublishRateLimiter())); - List> producerCheckFutures = new ArrayList<>(producers.size()); - producers.values().forEach(producer -> producerCheckFutures.add( - producer.checkPermissionsAsync().thenRun(producer::checkEncryption))); + applyPoliciesFutureList.add(FutureUtil.runWithCurrentThread(() -> updateSubscriptionsDispatcherRateLimiter())); + applyPoliciesFutureList.add(FutureUtil.runWithCurrentThread( + () -> replicators.forEach((name, replicator) -> replicator.updateRateLimiter()))); - return FutureUtil.waitForAll(producerCheckFutures).thenCompose((__) -> { - return updateSubscriptionsDispatcherRateLimiter().thenCompose((___) -> { - replicators.forEach((name, replicator) -> replicator.updateRateLimiter()); - checkMessageExpiry(); - CompletableFuture replicationFuture = checkReplicationAndRetryOnFailure(); - CompletableFuture dedupFuture = checkDeduplicationStatus(); - CompletableFuture persistentPoliciesFuture = checkPersistencePolicies(); - if (this.subscribeRateLimiter.isPresent()) { - subscribeRateLimiter.get().onSubscribeRateUpdate(getSubscribeRate()); - } + // Other components. + applyPoliciesFutureList.add(FutureUtil.runWithCurrentThread(() -> checkReplicationAndRetryOnFailure())); + applyPoliciesFutureList.add(FutureUtil.runWithCurrentThread(() -> checkDeduplicationStatus())); + applyPoliciesFutureList.add(FutureUtil.runWithCurrentThread(() -> checkPersistencePolicies())); + applyPoliciesFutureList.add(FutureUtil.runWithCurrentThread( + () -> preCreateSubscriptionForCompactionIfNeeded())); - return CompletableFuture.allOf(replicationFuture, dedupFuture, persistentPoliciesFuture, - preCreateSubscriptionForCompactionIfNeeded()); - }); - }).exceptionally(ex -> { - log.error("[{}] update namespace polices : {} error", this.getName(), data, ex); - throw FutureUtil.wrapToCompletionException(ex); - }); + return applyPoliciesFutureList; } /** @@ -3099,49 +3113,29 @@ public void onUpdate(TopicPolicies policies) { 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); - - updateDispatchRateLimiter(); checkReplicatedSubscriptionControllerState(); - updateSubscriptionsDispatcherRateLimiter().thenRun(() -> { - updatePublishDispatcher(); - initializeTopicSubscribeRateLimiterIfNeeded(Optional.ofNullable(policies)); - if (this.subscribeRateLimiter.isPresent()) { - subscribeRateLimiter.ifPresent(subscribeRateLimiter -> - subscribeRateLimiter.onSubscribeRateUpdate(getSubscribeRate())); - } - replicators.forEach((name, replicator) -> replicator.updateRateLimiter()); - checkMessageExpiry(); - if (policies.getReplicationClusters() != null) { - checkReplicationAndRetryOnFailure(); - } - checkDeduplicationStatus(); - - preCreateSubscriptionForCompactionIfNeeded(); - - // update managed ledger config - checkPersistencePolicies(); - }).exceptionally(e -> { - Throwable t = e instanceof CompletionException ? e.getCause() : 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 updateSubscriptionsDispatcherRateLimiter() { - List> subscriptionCheckFutures = new ArrayList<>((int) subscriptions.size()); + private void updateSubscriptionsDispatcherRateLimiter() { subscriptions.forEach((subName, sub) -> { - List> 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); } private void initializeTopicSubscribeRateLimiterIfNeeded(Optional policies) { 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 adfa95e8f4de3..376391e6db50f 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 @@ -43,6 +43,8 @@ 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; @@ -51,6 +53,7 @@ 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; @@ -84,8 +87,11 @@ 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; @@ -3060,4 +3066,49 @@ public void testProduceChangesWithEncryptionRequired() throws Exception { }); } + @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 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 79ad0100a9687..352a3909854f4 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.Collection; import java.util.List; @@ -54,6 +55,11 @@ public static CompletableFuture waitForAll(List runWithCurrentThread(Runnable runnable) { + return CompletableFuture.runAsync( + () -> runnable.run(), MoreExecutors.directExecutor()); + } + /** * Return a future that represents the completion of the futures in the provided Collection. *