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..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 @@ -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,67 @@ 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 CompletableFuture internalSetRetention(RetentionPolicies retention) { + validateAdminAccessForTenant(namespaceName.getTenant()); + validatePoliciesReadOnlyAccess(); + if (topicName.isGlobal()) { + validateGlobalNamespaceOwnership(namespaceName); + } + checkTopicLevelPolicyEnable(); + if (retention == null) { + return CompletableFuture.completedFuture(null); + } + 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); + return pulsar().getTopicPoliciesService().updateTopicPoliciesAsync(topicName, topicPolicies); + } + + protected CompletableFuture internalRemoveRetention() { + validateAdminAccessForTenant(namespaceName.getTenant()); + validatePoliciesReadOnlyAccess(); + if (topicName.isGlobal()) { + validateGlobalNamespaceOwnership(namespaceName); + } + checkTopicLevelPolicyEnable(); + Optional topicPolicies = getTopicPolicies(topicName); + if (!topicPolicies.isPresent()) { + return CompletableFuture.completedFuture(null); + } + topicPolicies.get().setRetentionPolicies(null); + return pulsar().getTopicPoliciesService().updateTopicPoliciesAsync(topicName, topicPolicies.get()); + } + 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..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; @@ -51,6 +52,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; @@ -1005,7 +1007,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"), @@ -1079,6 +1081,88 @@ public void removeMessageTTL(@Suspended final AsyncResponse asyncResponse, internalSetMessageTTL(asyncResponse, null); } + @GET + @Path("/{tenant}/{namespace}/{topic}/retention") + @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"), + @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 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"), + @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); + 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 + @Path("/{tenant}/{namespace}/{topic}/retention") + @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"), + @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); + 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 @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..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 @@ -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,35 @@ 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 = Optional.ofNullable(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; } @Override 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"; 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/TopicPoliciesTest.java similarity index 58% 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/TopicPoliciesTest.java index e847a7d25df0c..8e5aa5118c26c 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/TopicPoliciesTest.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 TopicPoliciesTest 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,8 +56,8 @@ 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); - Producer producer = pulsarClient.newProducer().topic(testTenant + "/" + testNamespace + "/" + "lookup-topic").create(); + admin.topics().createPartitionedTopic(testTopic, 2); + Producer producer = pulsarClient.newProducer().topic(testTopic).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 testCheckBacklogQuota() 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,90 @@ 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); + 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 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, retention); + + 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, 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 8a1618eed0119..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 @@ -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,142 @@ void createSubscription(String topic, String subscriptionName, MessageId message * Unexpected error */ void removeMessageTTL(String topic) throws PulsarAdminException; + + /** + * Set the retention configuration 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. + *

+ * + * @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: + *

+ * + * 

+     * 
+     * {
+     *     "retentionTimeInMinutes" : 60,            // how long to retain messages
+     *     "retentionSizeInMB" : 1024,              // retention backlog limit
+     * }
+     * 
+     *
+ * + * @param topic + * Topic name + */ + 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 5831deea6e610..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 @@ -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,81 @@ 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; + } + + @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 b3622d0884650..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 @@ -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,9 @@ 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()); + jcommander.addCommand("remove-retention", new RemoveRetention()); } @Parameters(commandDescription = "Get the list of topics under a namespace.") @@ -927,4 +931,65 @@ 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)); + } + } + + @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); + } + } }