From dd151a5a5323076d2faa733bd96182ff685b56ee Mon Sep 17 00:00:00 2001 From: Jiwei Guo Date: Tue, 22 Aug 2023 22:08:09 +0800 Subject: [PATCH] [fix][broker] Fix potential case cause retention policy not working on topic level (#21041) (cherry picked from commit d3a6df3c2a8f99ab3e5b7f6d826533ffd23328dc) --- .../service/persistent/PersistentTopic.java | 22 ++++++------- .../persistent/PersistentTopicTest.java | 31 +++++++++++++++++++ 2 files changed, 41 insertions(+), 12 deletions(-) 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 ffb6828c8c311..3e3363b8c5b75 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 @@ -1535,7 +1535,6 @@ private void disposeTopic(CompletableFuture closeFuture) { CompletableFuture checkReplicationAndRetryOnFailure() { CompletableFuture result = new CompletableFuture(); checkReplication().thenAccept(res -> { - log.info("[{}] Policies updated successfully", topic); result.complete(null); }).exceptionally(th -> { log.error("[{}] Policies update failed {}, scheduled retry in {} seconds", topic, th.getMessage(), @@ -1555,7 +1554,8 @@ public CompletableFuture checkDeduplicationStatus() { return messageDeduplication.checkStatus(); } - private CompletableFuture checkPersistencePolicies() { + @VisibleForTesting + CompletableFuture checkPersistencePolicies() { TopicName topicName = TopicName.get(topic); CompletableFuture future = new CompletableFuture<>(); brokerService.getManagedLedgerConfig(topicName).thenAccept(config -> { @@ -3486,16 +3486,14 @@ public void onUpdate(TopicPolicies policies) { replicators.forEach((name, replicator) -> replicator.updateRateLimiter()); shadowReplicators.forEach((name, replicator) -> replicator.updateRateLimiter()); checkMessageExpiry(); - checkReplicationAndRetryOnFailure(); - - checkDeduplicationStatus(); - - preCreateSubscriptionForCompactionIfNeeded(); - - // update managed ledger config - checkPersistencePolicies(); - }).exceptionally(e -> { - Throwable t = e instanceof CompletionException ? e.getCause() : e; + }) + .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; }); diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentTopicTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentTopicTest.java index 412b8207e34c7..41704af0b8bb2 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentTopicTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentTopicTest.java @@ -60,6 +60,7 @@ import org.apache.bookkeeper.mledger.ManagedLedger; import org.apache.pulsar.broker.service.BrokerService; import org.apache.pulsar.broker.service.BrokerTestBase; +import org.apache.pulsar.broker.service.TopicPoliciesService; import org.apache.pulsar.broker.stats.PrometheusMetricsTest; import org.apache.pulsar.broker.stats.prometheus.PrometheusMetricsGenerator; import org.apache.pulsar.client.admin.PulsarAdminException; @@ -77,7 +78,9 @@ import org.apache.pulsar.common.naming.TopicName; import org.apache.pulsar.common.policies.data.ClusterData; import org.apache.pulsar.common.policies.data.Policies; +import org.apache.pulsar.common.policies.data.RetentionPolicies; import org.apache.pulsar.common.policies.data.TenantInfo; +import org.apache.pulsar.common.policies.data.TopicPolicies; import org.apache.pulsar.common.policies.data.TopicStats; import org.awaitility.Awaitility; import org.junit.Assert; @@ -603,4 +606,32 @@ public void testCreateTopicWithZombieReplicatorCursor(boolean topicLevelPolicy) return !topic.getManagedLedger().getCursors().iterator().hasNext(); }); } + + @Test + public void testCheckPersistencePolicies() throws Exception { + final String myNamespace = "prop/ns"; + admin.namespaces().createNamespace(myNamespace, Sets.newHashSet("test")); + final String topic = "persistent://" + myNamespace + "/testConfig" + UUID.randomUUID(); + conf.setForceDeleteNamespaceAllowed(true); + pulsarClient.newProducer().topic(topic).create().close(); + RetentionPolicies retentionPolicies = new RetentionPolicies(1, 1); + PersistentTopic persistentTopic = spy((PersistentTopic) pulsar.getBrokerService().getTopicIfExists(topic).get().get()); + TopicPoliciesService policiesService = spy(pulsar.getTopicPoliciesService()); + doReturn(policiesService).when(pulsar).getTopicPoliciesService(); + TopicPolicies policies = new TopicPolicies(); + policies.setRetentionPolicies(retentionPolicies); + doReturn(policies).when(policiesService).getTopicPoliciesIfExists(TopicName.get(topic)); + persistentTopic.onUpdate(policies); + verify(persistentTopic, times(1)).checkPersistencePolicies(); + Awaitility.await().untilAsserted(() -> { + assertEquals(persistentTopic.getManagedLedger().getConfig().getRetentionSizeInMB(), 1L); + assertEquals(persistentTopic.getManagedLedger().getConfig().getRetentionTimeMillis(), TimeUnit.MINUTES.toMillis(1)); + }); + // throw exception + doReturn(CompletableFuture.failedFuture(new RuntimeException())).when(persistentTopic).checkPersistencePolicies(); + policies.setRetentionPolicies(new RetentionPolicies(2, 2)); + persistentTopic.onUpdate(policies); + assertEquals(persistentTopic.getManagedLedger().getConfig().getRetentionSizeInMB(), 1L); + assertEquals(persistentTopic.getManagedLedger().getConfig().getRetentionTimeMillis(), TimeUnit.MINUTES.toMillis(1)); + } }