From b133646d1747901d6e8a547d85bef998d9be5899 Mon Sep 17 00:00:00 2001 From: "jianyun.zhao" Date: Tue, 4 Aug 2020 23:48:00 +0800 Subject: [PATCH 1/7] Support set retention on topic level. --- .../pulsar/broker/admin/AdminResource.java | 4 + .../admin/impl/PersistentTopicsBase.java | 65 ++++++++ .../broker/admin/v2/PersistentTopics.java | 45 ++++++ .../service/persistent/PersistentTopic.java | 49 ++++-- ...est.java => TopicPoliciesDisableTest.java} | 40 +++-- ....java => TopicTopicPoliciesQuotaTest.java} | 144 ++++++++++-------- .../apache/pulsar/client/admin/Topics.java | 107 +++++++++++++ .../client/admin/internal/TopicsImpl.java | 56 +++++++ .../apache/pulsar/admin/cli/CmdTopics.java | 52 +++++++ 9 files changed, 469 insertions(+), 93 deletions(-) rename pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/{TopicBacklogQuotaDisableTest.java => TopicPoliciesDisableTest.java} (73%) rename pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/{TopicBacklogQuotaTest.java => TopicTopicPoliciesQuotaTest.java} (62%) 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 590c03282d50b..cd08b04a5134f 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 @@ -47,6 +47,7 @@ import org.apache.pulsar.broker.PulsarService; import org.apache.pulsar.broker.ServiceConfiguration; import org.apache.pulsar.broker.cache.LocalZooKeeperCacheService; +import org.apache.pulsar.broker.service.BrokerServiceException; import org.apache.pulsar.broker.web.PulsarWebResource; import org.apache.pulsar.broker.web.RestException; import org.apache.pulsar.common.api.proto.PulsarApi; @@ -517,6 +518,9 @@ protected Optional getTopicPolicies(TopicName topicName) { return Optional.ofNullable(pulsar().getTopicPoliciesService().getTopicPolicies(topicName)); } catch (RestException re) { throw re; + } catch (BrokerServiceException.TopicPoliciesCacheNotInitException e){ + log.error("Topic {} policies cache have not init.", topicName); + throw new RestException(e); } catch (Exception e) { log.error("[{}] Failed to get topic policies {}", clientAppId(), topicName, e); throw new RestException(e); diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java index 004ca6faaf5b0..3c3d18ef31caa 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java @@ -94,6 +94,8 @@ import org.apache.pulsar.client.impl.MessageIdImpl; import org.apache.pulsar.common.allocator.PulsarByteBufAllocator; import org.apache.pulsar.common.naming.PartitionedManagedLedgerInfo; +import org.apache.pulsar.common.policies.data.PolicyName; +import org.apache.pulsar.common.policies.data.PolicyOperation; import org.apache.pulsar.common.protocol.Commands; import org.apache.pulsar.common.api.proto.PulsarApi.CommandSubscribe.InitialPosition; import org.apache.pulsar.common.api.proto.PulsarApi.KeyValue; @@ -2156,6 +2158,69 @@ protected void internalRemoveBacklogQuota(AsyncResponse asyncResponse, internalSetBacklogQuota(asyncResponse, backlogQuotaType, null); } + protected void internalGetRetention(AsyncResponse asyncResponse){ + validateAdminAccessForTenant(namespaceName.getTenant()); + validatePoliciesReadOnlyAccess(); + if (topicName.isGlobal()) { + validateGlobalNamespaceOwnership(namespaceName); + } + checkTopicLevelPolicyEnable(); + Optional retention = getTopicPolicies(topicName) + .map(TopicPolicies::getRetentionPolicies); + if (!retention.isPresent()) { + asyncResponse.resume(Response.noContent().build()); + }else { + asyncResponse.resume(retention.get()); + } + } + + protected void internalSetRetention(AsyncResponse asyncResponse, + RetentionPolicies retention){ + validateNamespacePolicyOperation(namespaceName, PolicyName.RETENTION, PolicyOperation.WRITE); + validateAdminAccessForTenant(namespaceName.getTenant()); + validatePoliciesReadOnlyAccess(); + if (topicName.isGlobal()) { + validateGlobalNamespaceOwnership(namespaceName); + } + checkTopicLevelPolicyEnable(); + if (retention == null) { + asyncResponse.resume(Response.noContent().build()); + } + TopicPolicies topicPolicies = getTopicPolicies(topicName) + .orElseGet(TopicPolicies::new); + BacklogQuota backlogQuota = + topicPolicies.getBackLogQuotaMap().get(BacklogQuota.BacklogQuotaType.destination_storage.name()); + if (backlogQuota == null){ + Policies policies = getNamespacePolicies(topicName.getNamespaceObject()); + backlogQuota = policies.backlog_quota_map.get(BacklogQuota.BacklogQuotaType.destination_storage); + } + if(!checkBacklogQuota(backlogQuota, retention)){ + log.warn( + "[{}] Failed to update retention quota configuration for topic {}: conflicts with retention quota", + clientAppId(), topicName); + throw new RestException(Status.PRECONDITION_FAILED, + "Retention Quota must exceed configured backlog quota for topic. " + + "Please increase retention quota and retry"); + } + topicPolicies.setRetentionPolicies(retention); + pulsar().getTopicPoliciesService().updateTopicPoliciesAsync(topicName, topicPolicies) + .whenComplete((r, ex) -> { + if (ex != null) { + log.error("Failed updated backlog quota map",ex); + asyncResponse.resume(new RestException(ex)); + } else { + try { + log.info("[{}] Successfully updated backlog quota map: namespace={}, topic={}, retention={}", + clientAppId(), + namespaceName, + topicName.getLocalName(), + jsonMapper().writeValueAsString(retention)); + } catch (JsonProcessingException ignore) { } + asyncResponse.resume(Response.noContent().build()); + } + }); + } + protected MessageId internalTerminate(boolean authoritative) { if (topicName.isGlobal()) { validateGlobalNamespaceOwnership(namespaceName); diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v2/PersistentTopics.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v2/PersistentTopics.java index cd0dae6d10123..e82c46d9446a3 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v2/PersistentTopics.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v2/PersistentTopics.java @@ -51,6 +51,7 @@ import org.apache.pulsar.common.policies.data.BacklogQuota; import org.apache.pulsar.common.policies.data.PersistentOfflineTopicStats; import org.apache.pulsar.common.policies.data.PersistentTopicInternalStats; +import org.apache.pulsar.common.policies.data.RetentionPolicies; import org.apache.pulsar.common.policies.data.TopicPolicies; import org.apache.pulsar.common.policies.data.TopicStats; @@ -1079,6 +1080,50 @@ public void removeMessageTTL(@Suspended final AsyncResponse asyncResponse, internalSetMessageTTL(asyncResponse, null); } + @GET + @Path("/{tenant}/{namespace}/{topic}/retention") + @ApiOperation(value = "Get retention config on a topic.") + @ApiResponses(value = { @ApiResponse(code = 403, message = "Don't have admin permission"), + @ApiResponse(code = 404, message = "Topic does not exist"), + @ApiResponse(code = 405, message = "Topic level policy is disabled, to enable the topic level policy and retry"), + @ApiResponse(code = 409, message = "Concurrent modification") }) + public void getRetention(@Suspended final AsyncResponse asyncResponse, + @PathParam("tenant") String tenant, + @PathParam("namespace") String namespace, + @PathParam("topic") @Encoded String encodedTopic) { + validateTopicName(tenant, namespace, encodedTopic); + try { + internalGetRetention(asyncResponse); + } catch (RestException e) { + asyncResponse.resume(e); + } catch (Exception e) { + asyncResponse.resume(new RestException(e)); + } + } + + @POST + @Path("/{tenant}/{namespace}/{topic}/retention") + @ApiOperation(value = " Set retention configuration on a topic.") + @ApiResponses(value = { @ApiResponse(code = 403, message = "Don't have admin permission"), + @ApiResponse(code = 404, message = "Topic does not exist"), + @ApiResponse(code = 405, message = "Topic level policy is disabled, to enable the topic level policy and retry"), + @ApiResponse(code = 409, message = "Concurrent modification"), + @ApiResponse(code = 412, message = "Retention Quota must exceed backlog quota") }) + public void setRetention(@Suspended final AsyncResponse asyncResponse, + @PathParam("tenant") String tenant, + @PathParam("namespace") String namespace, + @PathParam("topic") @Encoded String encodedTopic, + @ApiParam(value = "Retention policies for the specified namespace") RetentionPolicies retention) { + validateTopicName(tenant, namespace, encodedTopic); + try { + internalSetRetention(asyncResponse, retention); + } catch (RestException e) { + asyncResponse.resume(e); + } catch (Exception e) { + asyncResponse.resume(new RestException(e)); + } + } + @POST @Path("/{tenant}/{namespace}/{topic}/terminate") @ApiOperation(value = "Terminate a topic. A topic that is terminated will not accept any more " 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 767b23677668d..6ffc04ae288e6 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 @@ -87,6 +87,7 @@ import org.apache.pulsar.broker.service.StreamingStats; import org.apache.pulsar.broker.service.Subscription; import org.apache.pulsar.broker.service.Topic; +import org.apache.pulsar.broker.service.TopicPoliciesService; import org.apache.pulsar.broker.service.persistent.DispatchRateLimiter.Type; import org.apache.pulsar.broker.stats.ClusterReplicationMetrics; import org.apache.pulsar.broker.stats.NamespaceStats; @@ -1757,28 +1758,48 @@ public void checkBackloggedCursors() { */ private boolean shouldTopicBeRetained() { TopicName name = TopicName.get(topic); + RetentionPolicies retentionPolicies = null; try { - Optional policies = brokerService.pulsar().getConfigurationCache().policiesCache() - .get(AdminResource.path(POLICIES, name.getNamespace())); - // If no policies, the default is to have no retention and delete the inactive topic - RetentionPolicies retentionPolicies = policies.map(p -> p.retention_policies).orElseGet( - () -> new RetentionPolicies( - brokerService.pulsar().getConfiguration().getDefaultRetentionTimeInMinutes(), - brokerService.pulsar().getConfiguration().getDefaultRetentionSizeInMB()) - ); - long retentionTime = TimeUnit.MINUTES.toNanos(retentionPolicies.getRetentionTimeInMinutes()); - - // Negative retention time means the topic should be retained indefinitely, - // because its own data has to be retained - return retentionTime < 0 || (System.nanoTime() - lastActive) < retentionTime; + retentionPolicies = getTopicPolicies(name) + .map(TopicPolicies::getRetentionPolicies) + .orElse( null); + if (retentionPolicies == null){ + retentionPolicies = brokerService.pulsar().getConfigurationCache().policiesCache() + .get(AdminResource.path(POLICIES, name.getNamespace())) + .map(p -> p.retention_policies) + .orElse(null); + } + if (retentionPolicies == null){ + // If no policies, the default is to have no retention and delete the inactive topic + retentionPolicies = new RetentionPolicies( + brokerService.pulsar().getConfiguration().getDefaultRetentionTimeInMinutes(), + brokerService.pulsar().getConfiguration().getDefaultRetentionSizeInMB()); + } } catch (Exception e) { if (log.isDebugEnabled()) { log.debug("[{}] Error getting policies", topic); } - // Don't delete in case we cannot get the policies return true; } + + long retentionTime = TimeUnit.MINUTES.toNanos(retentionPolicies.getRetentionTimeInMinutes()); + // Negative retention time means the topic should be retained indefinitely, + // because its own data has to be retained + return retentionTime < 0 || (System.nanoTime() - lastActive) < retentionTime; + } + + private Optional getTopicPolicies(TopicName topicName){ + if (!brokerService.pulsar().getConfiguration().isTopicLevelPoliciesEnabled()) { + return Optional.empty(); + } + try { + TopicPoliciesService topicPoliciesService = brokerService.pulsar().getTopicPoliciesService(); + return Optional.ofNullable(topicPoliciesService.getTopicPolicies(topicName)); + } catch (BrokerServiceException.TopicPoliciesCacheNotInitException e) { + log.warn("Topic {} policies cache have not init.", topicName); + return Optional.empty(); + } } @Override diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/TopicBacklogQuotaDisableTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/TopicPoliciesDisableTest.java similarity index 73% rename from pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/TopicBacklogQuotaDisableTest.java rename to pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/TopicPoliciesDisableTest.java index 0f8b2cbb57c5a..e473382584d19 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/TopicBacklogQuotaDisableTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/TopicPoliciesDisableTest.java @@ -21,24 +21,18 @@ import com.google.common.collect.Sets; import lombok.extern.slf4j.Slf4j; import org.apache.pulsar.broker.auth.MockedPulsarServiceBaseTest; -import org.apache.pulsar.broker.service.BacklogQuotaManager; import org.apache.pulsar.client.admin.PulsarAdminException; -import org.apache.pulsar.common.naming.TopicName; import org.apache.pulsar.common.policies.data.BacklogQuota; import org.apache.pulsar.common.policies.data.ClusterData; import org.apache.pulsar.common.policies.data.RetentionPolicies; import org.apache.pulsar.common.policies.data.TenantInfo; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; import org.testng.Assert; import org.testng.annotations.AfterMethod; import org.testng.annotations.BeforeMethod; import org.testng.annotations.Test; @Slf4j -public class TopicBacklogQuotaDisableTest extends MockedPulsarServiceBaseTest { - - private static final Logger LOG = LoggerFactory.getLogger(TopicBacklogQuotaDisableTest.class); +public class TopicPoliciesDisableTest extends MockedPulsarServiceBaseTest { private final String testTenant = "my-tenant"; @@ -46,7 +40,7 @@ public class TopicBacklogQuotaDisableTest extends MockedPulsarServiceBaseTest { private final String myNamespace = testTenant + "/" + testNamespace; - private final String backlogQuotaTopic = "persistent://" + myNamespace + "/test-set-backlog-quota"; + private final String testTopic = "persistent://" + myNamespace + "/test-set-backlog-quota"; @BeforeMethod @Override @@ -59,7 +53,7 @@ protected void setup() throws Exception { TenantInfo tenantInfo = new TenantInfo(Sets.newHashSet("role1", "role2"), Sets.newHashSet("test")); admin.tenants().createTenant(this.testTenant, tenantInfo); admin.namespaces().createNamespace(testTenant + "/" + testNamespace, Sets.newHashSet("test")); - admin.topics().createPartitionedTopic(backlogQuotaTopic, 2); + admin.topics().createPartitionedTopic(testTopic, 2); } @AfterMethod @@ -71,24 +65,44 @@ public void cleanup() throws Exception { @Test public void testBacklogQuotaDisabled() throws Exception { BacklogQuota backlogQuota = new BacklogQuota(1024, BacklogQuota.RetentionPolicy.consumer_backlog_eviction); - log.info("Backlog quota: {} will set to the topic: {}", backlogQuota, backlogQuotaTopic); + log.info("Backlog quota: {} will set to the topic: {}", backlogQuota, testTopic); + + try { + admin.topics().setBacklogQuota(testTopic, backlogQuota); + Assert.fail(); + } catch (PulsarAdminException e) { + Assert.assertEquals(e.getStatusCode(), 405); + } + + try { + admin.topics().removeBacklogQuota(testTopic); + Assert.fail(); + } catch (PulsarAdminException e) { + Assert.assertEquals(e.getStatusCode(), 405); + } try { - admin.topics().setBacklogQuota(backlogQuotaTopic, backlogQuota); + admin.topics().getBacklogQuotaMap(testTopic); Assert.fail(); } catch (PulsarAdminException e) { Assert.assertEquals(e.getStatusCode(), 405); } + } + + @Test + public void testRetentionDisabled() throws Exception { + RetentionPolicies retention = new RetentionPolicies(); + log.info("Retention: {} will set to the topic: {}", retention, testTopic); try { - admin.topics().removeBacklogQuota(backlogQuotaTopic); + admin.topics().setRetention(testTopic, retention); Assert.fail(); } catch (PulsarAdminException e) { Assert.assertEquals(e.getStatusCode(), 405); } try { - admin.topics().getBacklogQuotaMap(backlogQuotaTopic); + admin.topics().getRetention(testTopic); Assert.fail(); } catch (PulsarAdminException e) { Assert.assertEquals(e.getStatusCode(), 405); diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/TopicBacklogQuotaTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/TopicTopicPoliciesQuotaTest.java similarity index 62% rename from pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/TopicBacklogQuotaTest.java rename to pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/TopicTopicPoliciesQuotaTest.java index e847a7d25df0c..c6432f0aacca8 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/TopicBacklogQuotaTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/TopicTopicPoliciesQuotaTest.java @@ -29,17 +29,13 @@ import org.apache.pulsar.common.policies.data.ClusterData; import org.apache.pulsar.common.policies.data.RetentionPolicies; import org.apache.pulsar.common.policies.data.TenantInfo; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; import org.testng.Assert; import org.testng.annotations.AfterMethod; import org.testng.annotations.BeforeMethod; import org.testng.annotations.Test; @Slf4j -public class TopicBacklogQuotaTest extends MockedPulsarServiceBaseTest { - - private static final Logger LOG = LoggerFactory.getLogger(TopicBacklogQuotaTest.class); +public class TopicTopicPoliciesQuotaTest extends MockedPulsarServiceBaseTest { private final String testTenant = "my-tenant"; @@ -47,18 +43,7 @@ public class TopicBacklogQuotaTest extends MockedPulsarServiceBaseTest { private final String myNamespace = testTenant + "/" + testNamespace; - private final String backlogQuotaTopic = "persistent://" + myNamespace + "/test-set-backlog-quota"; - - public void disableTopicLevelPolicies() throws Exception { - this.conf.setSystemTopicEnabled(true); - this.conf.setTopicLevelPoliciesEnabled(false); - super.internalSetup(); - - admin.clusters().createCluster("test", new ClusterData(pulsar.getWebServiceAddress())); - TenantInfo tenantInfo = new TenantInfo(Sets.newHashSet("role1", "role2"), Sets.newHashSet("test")); - admin.tenants().createTenant(this.testTenant, tenantInfo); - admin.namespaces().createNamespace(testTenant + "/" + testNamespace, Sets.newHashSet("test")); - } + private final String testTopic = "persistent://" + myNamespace + "/test-set-backlog-quota"; @BeforeMethod @Override @@ -71,7 +56,7 @@ protected void setup() throws Exception { TenantInfo tenantInfo = new TenantInfo(Sets.newHashSet("role1", "role2"), Sets.newHashSet("test")); admin.tenants().createTenant(this.testTenant, tenantInfo); admin.namespaces().createNamespace(testTenant + "/" + testNamespace, Sets.newHashSet("test")); - admin.topics().createPartitionedTopic(backlogQuotaTopic, 2); + admin.topics().createPartitionedTopic(testTopic, 2); Producer producer = pulsarClient.newProducer().topic(testTenant + "/" + testNamespace + "/" + "lookup-topic").create(); producer.close(); Thread.sleep(3000); @@ -87,68 +72,68 @@ public void cleanup() throws Exception { public void testSetBacklogQuota() throws Exception { BacklogQuota backlogQuota = new BacklogQuota(1024, BacklogQuota.RetentionPolicy.consumer_backlog_eviction); - log.info("Backlog quota: {} will set to the topic: {}", backlogQuota, backlogQuotaTopic); + log.info("Backlog quota: {} will set to the topic: {}", backlogQuota, testTopic); - admin.topics().setBacklogQuota(backlogQuotaTopic, backlogQuota); - log.info("Backlog quota set success on topic: {}", backlogQuotaTopic); + admin.topics().setBacklogQuota(testTopic, backlogQuota); + log.info("Backlog quota set success on topic: {}", testTopic); Thread.sleep(3000); - BacklogQuota getBacklogQuota = admin.topics().getBacklogQuotaMap(backlogQuotaTopic) + BacklogQuota getBacklogQuota = admin.topics().getBacklogQuotaMap(testTopic) .get(BacklogQuota.BacklogQuotaType.destination_storage); - log.info("Backlog quota {} get on topic: {}", getBacklogQuota, backlogQuotaTopic); + log.info("Backlog quota {} get on topic: {}", getBacklogQuota, testTopic); Assert.assertEquals(getBacklogQuota, backlogQuota); BacklogQuotaManager backlogQuotaManager = pulsar.getBrokerService().getBacklogQuotaManager(); - BacklogQuota backlogQuotaInManager = backlogQuotaManager.getBacklogQuota(TopicName.get(backlogQuotaTopic)); - log.info("Backlog quota {} in backlog quota manager on topic: {}", backlogQuotaInManager, backlogQuotaTopic); + BacklogQuota backlogQuotaInManager = backlogQuotaManager.getBacklogQuota(TopicName.get(testTopic)); + log.info("Backlog quota {} in backlog quota manager on topic: {}", backlogQuotaInManager, testTopic); Assert.assertEquals(backlogQuotaInManager, backlogQuota); - admin.topics().deletePartitionedTopic(backlogQuotaTopic, true); + admin.topics().deletePartitionedTopic(testTopic, true); } @Test public void testRemoveBacklogQuota() throws Exception { BacklogQuota backlogQuota = new BacklogQuota(1024, BacklogQuota.RetentionPolicy.consumer_backlog_eviction); - log.info("Backlog quota: {} will set to the topic: {}", backlogQuota, backlogQuotaTopic); - admin.topics().setBacklogQuota(backlogQuotaTopic, backlogQuota); - log.info("Backlog quota set success on topic: {}", backlogQuotaTopic); + log.info("Backlog quota: {} will set to the topic: {}", backlogQuota, testTopic); + admin.topics().setBacklogQuota(testTopic, backlogQuota); + log.info("Backlog quota set success on topic: {}", testTopic); Thread.sleep(3000); - BacklogQuota getBacklogQuota = admin.topics().getBacklogQuotaMap(backlogQuotaTopic) + BacklogQuota getBacklogQuota = admin.topics().getBacklogQuotaMap(testTopic) .get(BacklogQuota.BacklogQuotaType.destination_storage); - log.info("Backlog quota {} get on topic: {}", getBacklogQuota, backlogQuotaTopic); + log.info("Backlog quota {} get on topic: {}", getBacklogQuota, testTopic); Assert.assertEquals(backlogQuota, getBacklogQuota); BacklogQuotaManager backlogQuotaManager = pulsar.getBrokerService().getBacklogQuotaManager(); - BacklogQuota backlogQuotaInManager = backlogQuotaManager.getBacklogQuota(TopicName.get(backlogQuotaTopic)); - log.info("Backlog quota {} in backlog quota manager on topic: {}", backlogQuotaInManager, backlogQuotaTopic); + BacklogQuota backlogQuotaInManager = backlogQuotaManager.getBacklogQuota(TopicName.get(testTopic)); + log.info("Backlog quota {} in backlog quota manager on topic: {}", backlogQuotaInManager, testTopic); Assert.assertEquals(backlogQuota, backlogQuotaInManager); - admin.topics().removeBacklogQuota(backlogQuotaTopic); - getBacklogQuota = admin.topics().getBacklogQuotaMap(backlogQuotaTopic) + admin.topics().removeBacklogQuota(testTopic); + getBacklogQuota = admin.topics().getBacklogQuotaMap(testTopic) .get(BacklogQuota.BacklogQuotaType.destination_storage); - log.info("Backlog quota {} get on topic: {} after remove", getBacklogQuota, backlogQuotaTopic); + log.info("Backlog quota {} get on topic: {} after remove", getBacklogQuota, testTopic); Assert.assertNull(getBacklogQuota); - backlogQuotaInManager = backlogQuotaManager.getBacklogQuota(TopicName.get(backlogQuotaTopic)); + backlogQuotaInManager = backlogQuotaManager.getBacklogQuota(TopicName.get(testTopic)); log.info("Backlog quota {} in backlog quota manager on topic: {} after remove", backlogQuotaInManager, - backlogQuotaTopic); + testTopic); Assert.assertEquals(backlogQuotaManager.getDefaultQuota(), backlogQuotaInManager); - admin.topics().deletePartitionedTopic(backlogQuotaTopic, true); + admin.topics().deletePartitionedTopic(testTopic, true); } @Test - public void testCheckQuota() throws Exception { + public void testCheckBlcklogQuota() throws Exception { RetentionPolicies retentionPolicies = new RetentionPolicies(10, 10); - String namespace = TopicName.get(backlogQuotaTopic).getNamespace(); + String namespace = TopicName.get(testTopic).getNamespace(); admin.namespaces().setRetention(namespace, retentionPolicies); BacklogQuota backlogQuota = new BacklogQuota(10 * 1024 * 1024, BacklogQuota.RetentionPolicy.consumer_backlog_eviction); - log.info("Backlog quota: {} will set to the topic: {}", backlogQuota, backlogQuotaTopic); + log.info("Backlog quota: {} will set to the topic: {}", backlogQuota, testTopic); try { - admin.topics().setBacklogQuota(backlogQuotaTopic, backlogQuota); + admin.topics().setBacklogQuota(testTopic, backlogQuota); Assert.fail(); } catch (PulsarAdminException e) { Assert.assertEquals(e.getStatusCode(), 412); @@ -156,9 +141,9 @@ public void testCheckQuota() throws Exception { Thread.sleep(3000); backlogQuota = new BacklogQuota(10 * 1024 * 1024 + 1, BacklogQuota.RetentionPolicy.consumer_backlog_eviction); - log.info("Backlog quota: {} will set to the topic: {}", backlogQuota, backlogQuotaTopic); + log.info("Backlog quota: {} will set to the topic: {}", backlogQuota, testTopic); try { - admin.topics().setBacklogQuota(backlogQuotaTopic, backlogQuota); + admin.topics().setBacklogQuota(testTopic, backlogQuota); Assert.fail(); } catch (PulsarAdminException e) { Assert.assertEquals(e.getStatusCode(), 412); @@ -166,44 +151,71 @@ public void testCheckQuota() throws Exception { Thread.sleep(3000); backlogQuota = new BacklogQuota(10 * 1024 * 1024 - 1, BacklogQuota.RetentionPolicy.consumer_backlog_eviction); - log.info("Backlog quota: {} will set to the topic: {}", backlogQuota, backlogQuotaTopic); - admin.topics().setBacklogQuota(backlogQuotaTopic, backlogQuota); + log.info("Backlog quota: {} will set to the topic: {}", backlogQuota, testTopic); + admin.topics().setBacklogQuota(testTopic, backlogQuota); Thread.sleep(3000); - BacklogQuota getBacklogQuota = admin.topics().getBacklogQuotaMap(backlogQuotaTopic) + BacklogQuota getBacklogQuota = admin.topics().getBacklogQuotaMap(testTopic) .get(BacklogQuota.BacklogQuotaType.destination_storage); - log.info("Backlog quota {} get on topic: {} after remove", getBacklogQuota, backlogQuotaTopic); + log.info("Backlog quota {} get on topic: {} after remove", getBacklogQuota, testTopic); Assert.assertEquals(getBacklogQuota, backlogQuota); - admin.topics().deletePartitionedTopic(backlogQuotaTopic, true); + admin.topics().deletePartitionedTopic(testTopic, true); } @Test - public void testBacklogQuotaDisabled() throws Exception { - disableTopicLevelPolicies(); - admin.topics().createPartitionedTopic(backlogQuotaTopic, 2); - - BacklogQuota backlogQuota = new BacklogQuota(1024, BacklogQuota.RetentionPolicy.consumer_backlog_eviction); - log.info("Backlog quota: {} will set to the topic: {}", backlogQuota, backlogQuotaTopic); + public void testCheckRetention() throws Exception { + BacklogQuota backlogQuota = + new BacklogQuota(10 * 1024 * 1024, BacklogQuota.RetentionPolicy.consumer_backlog_eviction); + RetentionPolicies retentionPolicies = new RetentionPolicies(10, 11); + String namespace = TopicName.get(testTopic).getNamespace(); + admin.namespaces().setRetention(namespace, retentionPolicies); + admin.topics().setBacklogQuota(testTopic, backlogQuota); + Thread.sleep(3000); + RetentionPolicies retention = new RetentionPolicies(10, 10); + log.info("Retention: {} will set to the topic: {}", retention, testTopic); try { - admin.topics().setBacklogQuota(backlogQuotaTopic, backlogQuota); + admin.topics().setRetention(testTopic, retention); Assert.fail(); } catch (PulsarAdminException e) { - Assert.assertEquals(e.getStatusCode(), 405); + Assert.assertEquals(e.getStatusCode(), 412); } + retention = new RetentionPolicies(10, 9); + log.info("Retention: {} will set to the topic: {}", retention, testTopic); try { - admin.topics().removeBacklogQuota(backlogQuotaTopic); + admin.topics().setRetention(testTopic, retention); Assert.fail(); } catch (PulsarAdminException e) { - Assert.assertEquals(e.getStatusCode(), 405); + Assert.assertEquals(e.getStatusCode(), 412); } - try { - admin.topics().getBacklogQuotaMap(backlogQuotaTopic); - Assert.fail(); - } catch (PulsarAdminException e) { - Assert.assertEquals(e.getStatusCode(), 405); - } + Thread.sleep(3000); + retention = new RetentionPolicies(10, 12); + log.info("Backlog quota: {} will set to the topic: {}", backlogQuota, testTopic); + admin.topics().setRetention(testTopic, retention); + Thread.sleep(3000); + RetentionPolicies getRetention = admin.topics().getRetention(testTopic); + log.info("Backlog quota {} get on topic: {}", getRetention, testTopic); + Assert.assertEquals(getRetention, retention); + + admin.topics().deletePartitionedTopic(testTopic, true); + } + + @Test + public void testSetRetention() throws Exception { + + RetentionPolicies retentionPolicies = new RetentionPolicies(); + log.info("Retention: {} will set to the topic: {}", retentionPolicies, testTopic); + + admin.topics().setRetention(testTopic, retentionPolicies); + log.info("Retention set success on topic: {}", testTopic); + + Thread.sleep(3000); + RetentionPolicies getRetention = admin.topics().getRetention(testTopic); + log.info("Retention {} get on topic: {}", getRetention, testTopic); + Assert.assertEquals(getRetention, retentionPolicies); + + admin.topics().deletePartitionedTopic(testTopic, true); } } diff --git a/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/Topics.java b/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/Topics.java index 8a1618eed0119..1099964668ceb 100644 --- a/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/Topics.java +++ b/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/Topics.java @@ -38,6 +38,7 @@ import org.apache.pulsar.common.policies.data.PartitionedTopicInternalStats; import org.apache.pulsar.common.policies.data.PartitionedTopicStats; import org.apache.pulsar.common.policies.data.PersistentTopicInternalStats; +import org.apache.pulsar.common.policies.data.RetentionPolicies; import org.apache.pulsar.common.policies.data.TopicStats; /** @@ -1476,4 +1477,110 @@ void createSubscription(String topic, String subscriptionName, MessageId message * Unexpected error */ void removeMessageTTL(String topic) throws PulsarAdminException; + + /** + * Set the retention configuration for all the topics on a topic. + *

+ * Set the retention configuration on a topic. This operation requires Pulsar super-user access. + *

+ * Request parameter example: + *

+ * + * 

+     * 
+     * {
+     *     "retentionTimeInMinutes" : 60,            // how long to retain messages
+     *     "retentionSizeInMB" : 1024,              // retention backlog limit
+     * }
+     * 
+     *
+ * + * @param topic + * Topic name + * + * @throws NotAuthorizedException + * Don't have admin permission + * @throws NotFoundException + * Topic does not exist + * @throws ConflictException + * Concurrent modification + * @throws PulsarAdminException + * Unexpected error + */ + void setRetention(String topic, RetentionPolicies retention) throws PulsarAdminException; + + /** + * Set the retention configuration for all the topics on a topic asynchronously. + *

+ * Set the retention configuration on a topic. This operation requires Pulsar super-user access. + *

+ * Request parameter example: + *

+ * + * 

+     * 
+     * {
+     *     "retentionTimeInMinutes" : 60,            // how long to retain messages
+     *     "retentionSizeInMB" : 1024,              // retention backlog limit
+     * }
+     * 
+     *
+ * + * @param topic + * Topic name + */ + CompletableFuture setRetentionAsync(String topic, RetentionPolicies retention); + + /** + * Get the retention configuration for a topic. + *

+ * Get the retention configuration for a topic. + *

+ * Response example: + *

+ * + * 

+     * 
+     * {
+     *     "retentionTimeInMinutes" : 60,            // how long to retain messages
+     *     "retentionSizeInMB" : 1024,              // retention backlog limit
+     * }
+     * 
+     *
+ * + * @param topic + * Topic name + * @throws NotAuthorizedException + * Don't have admin permission + * @throws NotFoundException + * Topic does not exist + * @throws ConflictException + * Concurrent modification + * @throws PulsarAdminException + * Unexpected error + */ + RetentionPolicies getRetention(String topic) throws PulsarAdminException; + + /** + * Get the retention configuration for a topic asynchronously. + *

+ * Get the retention configuration for a topic. + *

+ * Response example: + *

+ * + * 

+     * 
+     * {
+     *     "retentionTimeInMinutes" : 60,            // how long to retain messages
+     *     "retentionSizeInMB" : 1024,              // retention backlog limit
+     * }
+     * 
+     *
+ * + * @param topic + * Topic name + */ + CompletableFuture getRetentionAsync(String topic); + } diff --git a/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/TopicsImpl.java b/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/TopicsImpl.java index 5831deea6e610..600a902d3db06 100644 --- a/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/TopicsImpl.java +++ b/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/TopicsImpl.java @@ -76,6 +76,7 @@ import org.apache.pulsar.common.policies.data.PartitionedTopicInternalStats; import org.apache.pulsar.common.policies.data.PartitionedTopicStats; import org.apache.pulsar.common.policies.data.PersistentTopicInternalStats; +import org.apache.pulsar.common.policies.data.RetentionPolicies; import org.apache.pulsar.common.policies.data.TopicStats; import org.apache.pulsar.common.protocol.Commands; import org.apache.pulsar.common.util.Codec; @@ -1463,5 +1464,60 @@ public void removeMessageTTL(String topic) throws PulsarAdminException { } } + @Override + public void setRetention(String topic, RetentionPolicies retention) throws PulsarAdminException { + try { + setRetentionAsync(topic, retention).get(this.readTimeoutMs, TimeUnit.MILLISECONDS); + } catch (ExecutionException e) { + throw (PulsarAdminException) e.getCause(); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new PulsarAdminException(e); + } catch (TimeoutException e) { + throw new PulsarAdminException.TimeoutException(e); + } + } + + @Override + public CompletableFuture setRetentionAsync(String topic, RetentionPolicies retention) { + TopicName tn = validateTopic(topic); + WebTarget path = topicPath(tn, "retention"); + return asyncPostRequest(path, Entity.entity(retention, MediaType.APPLICATION_JSON)); + } + + @Override + public RetentionPolicies getRetention(String topic) throws PulsarAdminException { + try { + return getRetentionAsync(topic).get(this.readTimeoutMs, TimeUnit.MILLISECONDS); + } catch (ExecutionException e) { + throw (PulsarAdminException) e.getCause(); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new PulsarAdminException(e); + } catch (TimeoutException e) { + throw new PulsarAdminException.TimeoutException(e); + } + } + + @Override + public CompletableFuture getRetentionAsync(String topic) { + TopicName tn = validateTopic(topic); + WebTarget path = topicPath(tn, "retention"); + final CompletableFuture future = new CompletableFuture<>(); + asyncGetRequest(path, + new InvocationCallback() { + @Override + public void completed(RetentionPolicies retentionPolicies) { + future.complete(retentionPolicies); + } + + @Override + public void failed(Throwable throwable) { + future.completeExceptionally(getApiException(throwable.getCause())); + } + }); + return future; + } + private static final Logger log = LoggerFactory.getLogger(TopicsImpl.class); } diff --git a/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdTopics.java b/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdTopics.java index b3622d0884650..b08cdbd941710 100644 --- a/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdTopics.java +++ b/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdTopics.java @@ -49,6 +49,7 @@ import org.apache.pulsar.client.impl.MessageIdImpl; import org.apache.pulsar.common.policies.data.BacklogQuota; import org.apache.pulsar.common.policies.data.PersistentTopicInternalStats; +import org.apache.pulsar.common.policies.data.RetentionPolicies; import org.apache.pulsar.common.util.RelativeTimeUtil; @Parameters(commandDescription = "Operations on persistent topics") @@ -107,6 +108,8 @@ public CmdTopics(PulsarAdmin admin) { jcommander.addCommand("get-message-ttl", new GetMessageTTL()); jcommander.addCommand("set-message-ttl", new SetMessageTTL()); jcommander.addCommand("remove-message-ttl", new RemoveMessageTTL()); + jcommander.addCommand("get-retention", new GetRetention()); + jcommander.addCommand("set-retention", new SetRetention()); } @Parameters(commandDescription = "Get the list of topics under a namespace.") @@ -927,4 +930,53 @@ void run() throws PulsarAdminException { admin.topics().removeMessageTTL(persistentTopic); } } + + @Parameters(commandDescription = "Get the retention policy for a topic") + private class GetRetention extends CliCommand { + @Parameter(description = "persistent://tenant/namespace/topic", required = true) + private java.util.List params; + + @Override + void run() throws PulsarAdminException { + String persistentTopic = validatePersistentTopic(params); + print(admin.topics().getRetention(persistentTopic)); + } + } + + @Parameters(commandDescription = "Set the retention policy for a topic") + private class SetRetention extends CliCommand { + @Parameter(description = "persistent://tenant/namespace/topic", required = true) + private java.util.List params; + + @Parameter(names = { "--time", + "-t" }, description = "Retention time in minutes (or minutes, hours,days,weeks eg: 100m, 3h, 2d, 5w). " + + "0 means no retention and -1 means infinite time retention", required = true) + private String retentionTimeStr; + + @Parameter(names = { "--size", "-s" }, description = "Retention size limit (eg: 10M, 16G, 3T). " + + "0 or less than 1MB means no retention and -1 means infinite size retention", required = true) + private String limitStr; + + @Override + void run() throws PulsarAdminException { + String persistentTopic = validatePersistentTopic(params); + long sizeLimit = validateSizeString(limitStr); + long retentionTimeInSec = RelativeTimeUtil.parseRelativeTimeInSeconds(retentionTimeStr); + + final int retentionTimeInMin; + if (retentionTimeInSec != -1) { + retentionTimeInMin = (int) TimeUnit.SECONDS.toMinutes(retentionTimeInSec); + } else { + retentionTimeInMin = -1; + } + + final int retentionSizeInMB; + if (sizeLimit != -1) { + retentionSizeInMB = (int) (sizeLimit / (1024 * 1024)); + } else { + retentionSizeInMB = -1; + } + admin.topics().setRetention(persistentTopic, new RetentionPolicies(retentionTimeInMin, retentionSizeInMB)); + } + } } From 74c8d4dcbf74e335aab63cc9d38c409b139f0d00 Mon Sep 17 00:00:00 2001 From: "jianyun.zhao" Date: Wed, 5 Aug 2020 06:13:01 +0800 Subject: [PATCH 2/7] Support remove retention on topic level. --- .../admin/impl/PersistentTopicsBase.java | 35 +++++++++++++++-- .../broker/admin/v2/PersistentTopics.java | 24 +++++++++++- .../admin/TopicTopicPoliciesQuotaTest.java | 30 +++++++++++++-- .../apache/pulsar/client/admin/Topics.java | 38 +++++++++++++++++-- .../client/admin/internal/TopicsImpl.java | 21 ++++++++++ .../apache/pulsar/admin/cli/CmdTopics.java | 13 +++++++ 6 files changed, 149 insertions(+), 12 deletions(-) diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java index 3c3d18ef31caa..612d0b32cff74 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java @@ -2175,8 +2175,7 @@ protected void internalGetRetention(AsyncResponse asyncResponse){ } protected void internalSetRetention(AsyncResponse asyncResponse, - RetentionPolicies retention){ - validateNamespacePolicyOperation(namespaceName, PolicyName.RETENTION, PolicyOperation.WRITE); + RetentionPolicies retention) { validateAdminAccessForTenant(namespaceName.getTenant()); validatePoliciesReadOnlyAccess(); if (topicName.isGlobal()) { @@ -2206,11 +2205,11 @@ protected void internalSetRetention(AsyncResponse asyncResponse, pulsar().getTopicPoliciesService().updateTopicPoliciesAsync(topicName, topicPolicies) .whenComplete((r, ex) -> { if (ex != null) { - log.error("Failed updated backlog quota map",ex); + log.error("Failed updated retention",ex); asyncResponse.resume(new RestException(ex)); } else { try { - log.info("[{}] Successfully updated backlog quota map: namespace={}, topic={}, retention={}", + log.info("[{}] Successfully updated retention: namespace={}, topic={}, retention={}", clientAppId(), namespaceName, topicName.getLocalName(), @@ -2221,6 +2220,34 @@ protected void internalSetRetention(AsyncResponse asyncResponse, }); } + protected void internalRemoveRetention(AsyncResponse asyncResponse) { + validateAdminAccessForTenant(namespaceName.getTenant()); + validatePoliciesReadOnlyAccess(); + if (topicName.isGlobal()) { + validateGlobalNamespaceOwnership(namespaceName); + } + checkTopicLevelPolicyEnable(); + Optional topicPolicies = getTopicPolicies(topicName); + if (!topicPolicies.isPresent()) { + asyncResponse.resume(Response.noContent().build()); + return; + } + topicPolicies.get().setRetentionPolicies(null); + pulsar().getTopicPoliciesService().updateTopicPoliciesAsync(topicName, topicPolicies.get()) + .whenComplete((r, ex) -> { + if (ex != null) { + log.error("Failed updated retention",ex); + asyncResponse.resume(new RestException(ex)); + } else { + log.info("[{}] Successfully remove retention: namespace={}, topic={}", + clientAppId(), + namespaceName, + topicName.getLocalName()); + asyncResponse.resume(Response.noContent().build()); + } + }); + } + protected MessageId internalTerminate(boolean authoritative) { if (topicName.isGlobal()) { validateGlobalNamespaceOwnership(namespaceName); diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v2/PersistentTopics.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v2/PersistentTopics.java index e82c46d9446a3..71f00c08ff3df 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v2/PersistentTopics.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v2/PersistentTopics.java @@ -1103,7 +1103,7 @@ public void getRetention(@Suspended final AsyncResponse asyncResponse, @POST @Path("/{tenant}/{namespace}/{topic}/retention") - @ApiOperation(value = " Set retention configuration on a topic.") + @ApiOperation(value = "Set retention configuration on a topic.") @ApiResponses(value = { @ApiResponse(code = 403, message = "Don't have admin permission"), @ApiResponse(code = 404, message = "Topic does not exist"), @ApiResponse(code = 405, message = "Topic level policy is disabled, to enable the topic level policy and retry"), @@ -1124,6 +1124,28 @@ public void setRetention(@Suspended final AsyncResponse asyncResponse, } } + @DELETE + @Path("/{tenant}/{namespace}/{topic}/retention") + @ApiOperation(value = "Remove retention configuration on a topic.") + @ApiResponses(value = { @ApiResponse(code = 403, message = "Don't have admin permission"), + @ApiResponse(code = 404, message = "Topic does not exist"), + @ApiResponse(code = 405, message = "Topic level policy is disabled, to enable the topic level policy and retry"), + @ApiResponse(code = 409, message = "Concurrent modification"), + @ApiResponse(code = 412, message = "Retention Quota must exceed backlog quota") }) + public void removeRetention(@Suspended final AsyncResponse asyncResponse, + @PathParam("tenant") String tenant, + @PathParam("namespace") String namespace, + @PathParam("topic") @Encoded String encodedTopic) { + validateTopicName(tenant, namespace, encodedTopic); + try { + internalRemoveRetention(asyncResponse); + } catch (RestException e) { + asyncResponse.resume(e); + } catch (Exception e) { + asyncResponse.resume(new RestException(e)); + } + } + @POST @Path("/{tenant}/{namespace}/{topic}/terminate") @ApiOperation(value = "Terminate a topic. A topic that is terminated will not accept any more " diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/TopicTopicPoliciesQuotaTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/TopicTopicPoliciesQuotaTest.java index c6432f0aacca8..76dba1e3b2773 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/TopicTopicPoliciesQuotaTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/TopicTopicPoliciesQuotaTest.java @@ -204,17 +204,39 @@ public void testCheckRetention() throws Exception { @Test public void testSetRetention() throws Exception { + RetentionPolicies retention = new RetentionPolicies(60, 1024); + log.info("Retention: {} will set to the topic: {}", retention, testTopic); - RetentionPolicies retentionPolicies = new RetentionPolicies(); - log.info("Retention: {} will set to the topic: {}", retentionPolicies, testTopic); + admin.topics().setRetention(testTopic, retention); + log.info("Retention set success on topic: {}", testTopic); + + Thread.sleep(3000); + RetentionPolicies getRetention = admin.topics().getRetention(testTopic); + log.info("Retention {} get on topic: {}", getRetention, testTopic); + Assert.assertEquals(getRetention, retention); - admin.topics().setRetention(testTopic, retentionPolicies); + admin.topics().deletePartitionedTopic(testTopic, true); + } + + @Test + public void testRemoveRetention() throws Exception { + + RetentionPolicies retention = new RetentionPolicies(60, 1024); + log.info("Retention: {} will set to the topic: {}", retention, testTopic); + + admin.topics().setRetention(testTopic, retention); log.info("Retention set success on topic: {}", testTopic); Thread.sleep(3000); RetentionPolicies getRetention = admin.topics().getRetention(testTopic); log.info("Retention {} get on topic: {}", getRetention, testTopic); - Assert.assertEquals(getRetention, retentionPolicies); + Assert.assertEquals(getRetention, retention); + + admin.topics().removeRetention(testTopic); + Thread.sleep(3000); + log.info("Retention {} get on topic: {} after remove", getRetention, testTopic); + getRetention = admin.topics().getRetention(testTopic); + Assert.assertNull(getRetention); admin.topics().deletePartitionedTopic(testTopic, true); } diff --git a/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/Topics.java b/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/Topics.java index 1099964668ceb..9dd46d31b3945 100644 --- a/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/Topics.java +++ b/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/Topics.java @@ -1566,7 +1566,40 @@ void createSubscription(String topic, String subscriptionName, MessageId message *

* Get the retention configuration for a topic. *

- * Response example: + * + * @param topic + * Topic name + */ + CompletableFuture getRetentionAsync(String topic); + + /** + * Remove the retention configuration for all the topics on a topic. + *

+ * Remove the retention configuration on a topic. This operation requires Pulsar super-user access. + *

+ * Request parameter example: + *

+ * + * @param topic + * Topic name + * + * @throws NotAuthorizedException + * Don't have admin permission + * @throws NotFoundException + * Topic does not exist + * @throws ConflictException + * Concurrent modification + * @throws PulsarAdminException + * Unexpected error + */ + void removeRetention(String topic) throws PulsarAdminException; + + /** + * Remove the retention configuration for all the topics on a topic asynchronously. + *

+ * Remove the retention configuration on a topic. This operation requires Pulsar super-user access. + *

+ * Request parameter example: *

* * 

@@ -1581,6 +1614,5 @@ void createSubscription(String topic, String subscriptionName, MessageId message
      * @param topic
      *            Topic name
      */
-    CompletableFuture getRetentionAsync(String topic);
-
+    CompletableFuture removeRetentionAsync(String topic);
 }
diff --git a/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/TopicsImpl.java b/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/TopicsImpl.java
index 600a902d3db06..4df515af9bc99 100644
--- a/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/TopicsImpl.java
+++ b/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/TopicsImpl.java
@@ -1519,5 +1519,26 @@ public void failed(Throwable throwable) {
         return future;
     }
 
+    @Override
+    public void removeRetention(String topic) throws PulsarAdminException {
+        try {
+            removeRetentionAsync(topic).get(this.readTimeoutMs, TimeUnit.MILLISECONDS);
+        } catch (ExecutionException e) {
+            throw (PulsarAdminException) e.getCause();
+        } catch (InterruptedException e) {
+            Thread.currentThread().interrupt();
+            throw new PulsarAdminException(e);
+        } catch (TimeoutException e) {
+            throw new PulsarAdminException.TimeoutException(e);
+        }
+    }
+
+    @Override
+    public CompletableFuture removeRetentionAsync(String topic) {
+        TopicName tn = validateTopic(topic);
+        WebTarget path = topicPath(tn, "retention");
+        return asyncDeleteRequest(path);
+    }
+
     private static final Logger log = LoggerFactory.getLogger(TopicsImpl.class);
 }
diff --git a/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdTopics.java b/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdTopics.java
index b08cdbd941710..658233f3448b3 100644
--- a/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdTopics.java
+++ b/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdTopics.java
@@ -110,6 +110,7 @@ public CmdTopics(PulsarAdmin admin) {
         jcommander.addCommand("remove-message-ttl", new RemoveMessageTTL());
         jcommander.addCommand("get-retention", new GetRetention());
         jcommander.addCommand("set-retention", new SetRetention());
+        jcommander.addCommand("remove-retention", new RemoveRetention());
     }
 
     @Parameters(commandDescription = "Get the list of topics under a namespace.")
@@ -979,4 +980,16 @@ void run() throws PulsarAdminException {
             admin.topics().setRetention(persistentTopic, new RetentionPolicies(retentionTimeInMin, retentionSizeInMB));
         }
     }
+
+    @Parameters(commandDescription = "Remove the retention policy for a topic")
+    private class RemoveRetention extends CliCommand {
+        @Parameter(description = "persistent://tenant/namespace/topic", required = true)
+        private java.util.List params;
+
+        @Override
+        void run() throws PulsarAdminException {
+            String persistentTopic = validatePersistentTopic(params);
+            admin.topics().removeRetention(persistentTopic);
+        }
+    }
 }

From 477d08a70e9043f455ea2079062f964c248c26f3 Mon Sep 17 00:00:00 2001
From: "jianyun.zhao" 
Date: Wed, 5 Aug 2020 10:10:42 +0800
Subject: [PATCH 3/7] Fix and rename to `TopicPoliciesTest`

---
 ...opicTopicPoliciesQuotaTest.java => TopicPoliciesTest.java} | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)
 rename pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/{TopicTopicPoliciesQuotaTest.java => TopicPoliciesTest.java} (98%)

diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/TopicTopicPoliciesQuotaTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/TopicPoliciesTest.java
similarity index 98%
rename from pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/TopicTopicPoliciesQuotaTest.java
rename to pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/TopicPoliciesTest.java
index 76dba1e3b2773..734e1fbaa350b 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/TopicTopicPoliciesQuotaTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/TopicPoliciesTest.java
@@ -35,7 +35,7 @@
 import org.testng.annotations.Test;
 
 @Slf4j
-public class TopicTopicPoliciesQuotaTest extends MockedPulsarServiceBaseTest {
+public class TopicPoliciesTest extends MockedPulsarServiceBaseTest {
 
     private final String testTenant = "my-tenant";
 
@@ -57,7 +57,7 @@ protected void setup() throws Exception {
         admin.tenants().createTenant(this.testTenant, tenantInfo);
         admin.namespaces().createNamespace(testTenant + "/" + testNamespace, Sets.newHashSet("test"));
         admin.topics().createPartitionedTopic(testTopic, 2);
-        Producer producer = pulsarClient.newProducer().topic(testTenant + "/" + testNamespace + "/" + "lookup-topic").create();
+        Producer producer = pulsarClient.newProducer().topic(testTopic).create();
         producer.close();
         Thread.sleep(3000);
     }

From e73913329298d23cc58f296971115a28d0dc8d90 Mon Sep 17 00:00:00 2001
From: "jianyun.zhao" 
Date: Thu, 6 Aug 2020 17:49:23 +0800
Subject: [PATCH 4/7] fix comment and modified test

---
 .../broker/admin/v2/PersistentTopics.java       |  8 ++++----
 .../service/persistent/PersistentTopic.java     | 17 ++---------------
 .../pulsar/broker/admin/TopicPoliciesTest.java  |  5 +----
 .../org/apache/pulsar/client/admin/Topics.java  |  2 +-
 4 files changed, 8 insertions(+), 24 deletions(-)

diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v2/PersistentTopics.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v2/PersistentTopics.java
index 71f00c08ff3df..9c3bcad1bc33d 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v2/PersistentTopics.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v2/PersistentTopics.java
@@ -1006,7 +1006,7 @@ public Map getBacklogQuotaMap(@Path
 
     @POST
     @Path("/{tenant}/{namespace}/{topic}/backlogQuota")
-    @ApiOperation(value = " Set a backlog quota for a topic.")
+    @ApiOperation(value = "Set a backlog quota for a topic.")
     @ApiResponses(value = { @ApiResponse(code = 403, message = "Don't have admin permission"),
             @ApiResponse(code = 404, message = "Topic does not exist"),
             @ApiResponse(code = 409, message = "Concurrent modification"),
@@ -1082,7 +1082,7 @@ public void removeMessageTTL(@Suspended final AsyncResponse asyncResponse,
 
     @GET
     @Path("/{tenant}/{namespace}/{topic}/retention")
-    @ApiOperation(value = "Get retention config on a topic.")
+    @ApiOperation(value = "Get retention configuration for specified topic.")
     @ApiResponses(value = { @ApiResponse(code = 403, message = "Don't have admin permission"),
             @ApiResponse(code = 404, message = "Topic does not exist"),
             @ApiResponse(code = 405, message = "Topic level policy is disabled, to enable the topic level policy and retry"),
@@ -1103,7 +1103,7 @@ public void getRetention(@Suspended final AsyncResponse asyncResponse,
 
     @POST
     @Path("/{tenant}/{namespace}/{topic}/retention")
-    @ApiOperation(value = "Set retention configuration on a topic.")
+    @ApiOperation(value = "Set retention configuration for specified topic.")
     @ApiResponses(value = { @ApiResponse(code = 403, message = "Don't have admin permission"),
             @ApiResponse(code = 404, message = "Topic does not exist"),
             @ApiResponse(code = 405, message = "Topic level policy is disabled, to enable the topic level policy and retry"),
@@ -1126,7 +1126,7 @@ public void setRetention(@Suspended final AsyncResponse asyncResponse,
 
     @DELETE
     @Path("/{tenant}/{namespace}/{topic}/retention")
-    @ApiOperation(value = "Remove retention configuration on a topic.")
+    @ApiOperation(value = "Remove retention configuration for specified topic.")
     @ApiResponses(value = { @ApiResponse(code = 403, message = "Don't have admin permission"),
             @ApiResponse(code = 404, message = "Topic does not exist"),
             @ApiResponse(code = 405, message = "Topic level policy is disabled, to enable the topic level policy and retry"),
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 6ffc04ae288e6..c2cc1c7e08e93 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
@@ -1760,9 +1760,9 @@ private boolean shouldTopicBeRetained() {
         TopicName name = TopicName.get(topic);
         RetentionPolicies retentionPolicies = null;
         try {
-            retentionPolicies = getTopicPolicies(name)
+            retentionPolicies = Optional.ofNullable(getTopicPolicies(name))
                     .map(TopicPolicies::getRetentionPolicies)
-                    .orElse( null);
+                    .orElse(null);
             if (retentionPolicies == null){
                 retentionPolicies = brokerService.pulsar().getConfigurationCache().policiesCache()
                         .get(AdminResource.path(POLICIES, name.getNamespace()))
@@ -1789,19 +1789,6 @@ private boolean shouldTopicBeRetained() {
         return retentionTime < 0 || (System.nanoTime() - lastActive) < retentionTime;
     }
 
-    private Optional getTopicPolicies(TopicName topicName){
-        if (!brokerService.pulsar().getConfiguration().isTopicLevelPoliciesEnabled()) {
-            return Optional.empty();
-        }
-        try {
-            TopicPoliciesService topicPoliciesService = brokerService.pulsar().getTopicPoliciesService();
-            return Optional.ofNullable(topicPoliciesService.getTopicPolicies(topicName));
-        } catch (BrokerServiceException.TopicPoliciesCacheNotInitException e) {
-            log.warn("Topic {} policies cache have not init.", topicName);
-            return Optional.empty();
-        }
-    }
-
     @Override
     public CompletableFuture onPoliciesUpdate(Policies data) {
         if (log.isDebugEnabled()) {
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 734e1fbaa350b..8e5aa5118c26c 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
@@ -124,7 +124,7 @@ public void testRemoveBacklogQuota() throws Exception {
     }
 
     @Test
-    public void testCheckBlcklogQuota() throws Exception {
+    public void testCheckBacklogQuota() throws Exception {
         RetentionPolicies retentionPolicies = new RetentionPolicies(10, 10);
         String namespace = TopicName.get(testTopic).getNamespace();
         admin.namespaces().setRetention(namespace, retentionPolicies);
@@ -166,9 +166,6 @@ public void testCheckBlcklogQuota() throws Exception {
     public void testCheckRetention() throws Exception {
         BacklogQuota backlogQuota =
                 new BacklogQuota(10 * 1024 * 1024, BacklogQuota.RetentionPolicy.consumer_backlog_eviction);
-        RetentionPolicies retentionPolicies = new RetentionPolicies(10, 11);
-        String namespace = TopicName.get(testTopic).getNamespace();
-        admin.namespaces().setRetention(namespace, retentionPolicies);
         admin.topics().setBacklogQuota(testTopic, backlogQuota);
         Thread.sleep(3000);
 
diff --git a/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/Topics.java b/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/Topics.java
index 9dd46d31b3945..72098ee64056a 100644
--- a/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/Topics.java
+++ b/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/Topics.java
@@ -1479,7 +1479,7 @@ void createSubscription(String topic, String subscriptionName, MessageId message
     void removeMessageTTL(String topic) throws PulsarAdminException;
 
     /**
-     * Set the retention configuration for all the topics on a topic.
+     * Set the retention configuration on a topic.
      * 

      * Set the retention configuration on a topic. This operation requires Pulsar super-user access.
      * 


From dbd777142809dc73c243dba853da5eaf0420acbc Mon Sep 17 00:00:00 2001
From: "jianyun.zhao" 
Date: Thu, 6 Aug 2020 17:52:22 +0800
Subject: [PATCH 5/7] =?UTF-8?q?fix=20log=20class=EF=BC=8Cbecause=20`TopicB?=
 =?UTF-8?q?acklogQuotaTest`=20renamed=20to=20`TopicPoliciesTest`?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 .../org/apache/pulsar/broker/admin/TopicMessageTTLTest.java     | 2 --
 1 file changed, 2 deletions(-)

diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/TopicMessageTTLTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/TopicMessageTTLTest.java
index 0ceb8f4e23c14..f7b33fc4272fa 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/TopicMessageTTLTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/TopicMessageTTLTest.java
@@ -39,8 +39,6 @@
 @Slf4j
 public class TopicMessageTTLTest extends MockedPulsarServiceBaseTest {
 
-    private static final Logger LOG = LoggerFactory.getLogger(TopicBacklogQuotaTest.class);
-
     private final String testTenant = "my-tenant";
 
     private final String testNamespace = "my-namespace";

From c8f10438f372628e8f1fe51ddd88200777a8d5de Mon Sep 17 00:00:00 2001
From: "jianyun.zhao" 
Date: Mon, 10 Aug 2020 20:58:11 +0800
Subject: [PATCH 6/7] modify the exception is throw

---
 .../pulsar/broker/admin/impl/PersistentTopicsBase.java       | 5 +++--
 1 file changed, 3 insertions(+), 2 deletions(-)

diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java
index 612d0b32cff74..2e20d6ddcb91b 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java
@@ -2197,9 +2197,10 @@ protected void internalSetRetention(AsyncResponse asyncResponse,
             log.warn(
                     "[{}] Failed to update retention quota configuration for topic {}: conflicts with retention quota",
                     clientAppId(), topicName);
-            throw new RestException(Status.PRECONDITION_FAILED,
+            asyncResponse.resume(new RestException(Status.PRECONDITION_FAILED,
                     "Retention Quota must exceed configured backlog quota for topic. " +
-                            "Please increase retention quota and retry");
+                            "Please increase retention quota and retry"));
+            return;
         }
         topicPolicies.setRetentionPolicies(retention);
         pulsar().getTopicPoliciesService().updateTopicPoliciesAsync(topicName, topicPolicies)

From bebd6f7455e8af5d512ca40c476ac400cbf08dac Mon Sep 17 00:00:00 2001
From: "jianyun.zhao" 
Date: Mon, 10 Aug 2020 21:18:01 +0800
Subject: [PATCH 7/7] modify the use of asyncResponse

---
 .../admin/impl/PersistentTopicsBase.java      | 46 ++++---------------
 .../broker/admin/v2/PersistentTopics.java     | 45 ++++++++++++------
 2 files changed, 39 insertions(+), 52 deletions(-)

diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java
index 2e20d6ddcb91b..a915546778ba2 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java
@@ -2174,8 +2174,7 @@ protected void internalGetRetention(AsyncResponse asyncResponse){
         }
     }
 
-    protected void internalSetRetention(AsyncResponse asyncResponse,
-            RetentionPolicies retention) {
+    protected CompletableFuture internalSetRetention(RetentionPolicies retention) {
         validateAdminAccessForTenant(namespaceName.getTenant());
         validatePoliciesReadOnlyAccess();
         if (topicName.isGlobal()) {
@@ -2183,7 +2182,7 @@ protected void internalSetRetention(AsyncResponse asyncResponse,
         }
         checkTopicLevelPolicyEnable();
         if (retention == null) {
-            asyncResponse.resume(Response.noContent().build());
+            return CompletableFuture.completedFuture(null);
         }
         TopicPolicies topicPolicies = getTopicPolicies(topicName)
                 .orElseGet(TopicPolicies::new);
@@ -2197,31 +2196,15 @@ protected void internalSetRetention(AsyncResponse asyncResponse,
             log.warn(
                     "[{}] Failed to update retention quota configuration for topic {}: conflicts with retention quota",
                     clientAppId(), topicName);
-            asyncResponse.resume(new RestException(Status.PRECONDITION_FAILED,
+            throw new RestException(Status.PRECONDITION_FAILED,
                     "Retention Quota must exceed configured backlog quota for topic. " +
-                            "Please increase retention quota and retry"));
-            return;
+                            "Please increase retention quota and retry");
         }
         topicPolicies.setRetentionPolicies(retention);
-        pulsar().getTopicPoliciesService().updateTopicPoliciesAsync(topicName, topicPolicies)
-                .whenComplete((r, ex) -> {
-                    if (ex != null) {
-                        log.error("Failed updated retention",ex);
-                        asyncResponse.resume(new RestException(ex));
-                    } else {
-                        try {
-                            log.info("[{}] Successfully updated retention: namespace={}, topic={}, retention={}",
-                                    clientAppId(),
-                                    namespaceName,
-                                    topicName.getLocalName(),
-                                    jsonMapper().writeValueAsString(retention));
-                        } catch (JsonProcessingException ignore) { }
-                        asyncResponse.resume(Response.noContent().build());
-                    }
-                });
+        return pulsar().getTopicPoliciesService().updateTopicPoliciesAsync(topicName, topicPolicies);
     }
 
-    protected void internalRemoveRetention(AsyncResponse asyncResponse) {
+    protected CompletableFuture internalRemoveRetention() {
         validateAdminAccessForTenant(namespaceName.getTenant());
         validatePoliciesReadOnlyAccess();
         if (topicName.isGlobal()) {
@@ -2230,23 +2213,10 @@ protected void internalRemoveRetention(AsyncResponse asyncResponse) {
         checkTopicLevelPolicyEnable();
         Optional topicPolicies = getTopicPolicies(topicName);
         if (!topicPolicies.isPresent()) {
-            asyncResponse.resume(Response.noContent().build());
-            return;
+            return CompletableFuture.completedFuture(null);
         }
         topicPolicies.get().setRetentionPolicies(null);
-        pulsar().getTopicPoliciesService().updateTopicPoliciesAsync(topicName, topicPolicies.get())
-                .whenComplete((r, ex) -> {
-                    if (ex != null) {
-                        log.error("Failed updated retention",ex);
-                        asyncResponse.resume(new RestException(ex));
-                    } else {
-                        log.info("[{}] Successfully remove retention: namespace={}, topic={}",
-                                clientAppId(),
-                                namespaceName,
-                                topicName.getLocalName());
-                        asyncResponse.resume(Response.noContent().build());
-                    }
-                });
+        return pulsar().getTopicPoliciesService().updateTopicPoliciesAsync(topicName, topicPolicies.get());
     }
 
     protected MessageId internalTerminate(boolean authoritative) {
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v2/PersistentTopics.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v2/PersistentTopics.java
index 9c3bcad1bc33d..eb396ecae18e7 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v2/PersistentTopics.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v2/PersistentTopics.java
@@ -39,6 +39,7 @@
 import javax.ws.rs.core.MediaType;
 import javax.ws.rs.core.Response;
 
+import com.fasterxml.jackson.core.JsonProcessingException;
 import com.google.common.collect.Maps;
 import org.apache.pulsar.broker.admin.impl.PersistentTopicsBase;
 import org.apache.pulsar.broker.web.RestException;
@@ -1115,13 +1116,24 @@ public void setRetention(@Suspended final AsyncResponse asyncResponse,
             @PathParam("topic") @Encoded String encodedTopic,
             @ApiParam(value = "Retention policies for the specified namespace") RetentionPolicies retention) {
         validateTopicName(tenant, namespace, encodedTopic);
-        try {
-            internalSetRetention(asyncResponse, retention);
-        } catch (RestException e) {
-            asyncResponse.resume(e);
-        } catch (Exception e) {
-            asyncResponse.resume(new RestException(e));
-        }
+        internalSetRetention(retention).whenComplete((r, ex) -> {
+            if (ex instanceof RestException) {
+                log.error("Failed updated retention", ex);
+                asyncResponse.resume(ex);
+            }else if (ex != null) {
+                log.error("Failed updated retention", ex);
+                asyncResponse.resume(new RestException(ex));
+            } else {
+                try {
+                    log.info("[{}] Successfully updated retention: namespace={}, topic={}, retention={}",
+                            clientAppId(),
+                            namespaceName,
+                            topicName.getLocalName(),
+                            jsonMapper().writeValueAsString(retention));
+                } catch (JsonProcessingException ignore) { }
+                asyncResponse.resume(Response.noContent().build());
+            }
+        });
     }
 
     @DELETE
@@ -1137,13 +1149,18 @@ public void removeRetention(@Suspended final AsyncResponse asyncResponse,
                              @PathParam("namespace") String namespace,
                              @PathParam("topic") @Encoded String encodedTopic) {
         validateTopicName(tenant, namespace, encodedTopic);
-        try {
-            internalRemoveRetention(asyncResponse);
-        } catch (RestException e) {
-            asyncResponse.resume(e);
-        } catch (Exception e) {
-            asyncResponse.resume(new RestException(e));
-        }
+        internalRemoveRetention().whenComplete((r, ex) -> {
+            if (ex != null) {
+                log.error("Failed updated retention", ex);
+                asyncResponse.resume(new RestException(ex));
+            } else {
+                log.info("[{}] Successfully remove retention: namespace={}, topic={}",
+                        clientAppId(),
+                        namespaceName,
+                        topicName.getLocalName());
+                asyncResponse.resume(Response.noContent().build());
+            }
+        });
     }
 
     @POST