From 2ecb9c2e94251faa63165fd855c2578ea5cbfb0c Mon Sep 17 00:00:00 2001 From: "jianyun.zhao" Date: Mon, 20 Jul 2020 21:27:16 +0800 Subject: [PATCH 1/9] adding policy of backlog quota on topic level --- .../broker/service/BacklogQuotaManager.java | 35 ++++++++++++++----- .../service/persistent/PersistentTopic.java | 8 ++--- .../common/policies/data/TopicPolicies.java | 2 +- 3 files changed, 29 insertions(+), 16 deletions(-) diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BacklogQuotaManager.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BacklogQuotaManager.java index fd47425bfaac5..2315afc3f774a 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BacklogQuotaManager.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BacklogQuotaManager.java @@ -20,6 +20,7 @@ import java.util.List; import java.util.Map; +import java.util.Optional; import java.util.concurrent.CompletableFuture; import org.apache.bookkeeper.mledger.ManagedCursor; @@ -32,9 +33,8 @@ import org.apache.pulsar.common.policies.data.BacklogQuota; import org.apache.pulsar.common.policies.data.Policies; import org.apache.pulsar.common.policies.data.BacklogQuota.BacklogQuotaType; -import org.apache.pulsar.common.policies.data.BacklogQuota.RetentionPolicy; +import org.apache.pulsar.common.policies.data.TopicPolicies; import org.apache.pulsar.common.util.FutureUtil; -import org.apache.pulsar.common.util.collections.ConcurrentOpenHashSet; import org.apache.pulsar.zookeeper.ZooKeeperDataCache; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -47,12 +47,17 @@ public class BacklogQuotaManager { private static final Logger log = LoggerFactory.getLogger(BacklogQuotaManager.class); private final BacklogQuota defaultQuota; private final ZooKeeperDataCache zkCache; + private final TopicPoliciesService topicPoliciesService; + private final boolean isTopicLevelPoliciesEnable; + public BacklogQuotaManager(PulsarService pulsar) { + this.isTopicLevelPoliciesEnable = pulsar.getConfiguration().isTopicLevelPoliciesEnabled(); this.defaultQuota = new BacklogQuota( pulsar.getConfiguration().getBacklogQuotaDefaultLimitGB() * 1024 * 1024 * 1024, pulsar.getConfiguration().getBacklogQuotaDefaultRetentionPolicy()); this.zkCache = pulsar.getConfigurationCache().policiesCache(); + this.topicPoliciesService = pulsar.getTopicPoliciesService(); } public BacklogQuota getDefaultQuota() { @@ -70,9 +75,24 @@ public BacklogQuota getBacklogQuota(String namespace, String policyPath) { } } - public long getBacklogQuotaLimit(String namespace) { - String policyPath = AdminResource.path(POLICIES, namespace); - return getBacklogQuota(namespace, policyPath).getLimit(); + public BacklogQuota getBacklogQuota(TopicName topicName) { + String policyPath = AdminResource.path(POLICIES, topicName.getNamespace()); + if (!isTopicLevelPoliciesEnable) { + return getBacklogQuota(topicName.getNamespace(),policyPath); + } + + Optional optional = Optional.empty(); + try { + TopicPolicies topicPolicies = topicPoliciesService.getTopicPolicies(topicName); + optional = Optional.ofNullable(topicPolicies.getBackLogQuotaMap().get(BacklogQuotaType.destination_storage.name())); + } catch (Exception e) { + log.error("Failed to read policies data, will apply the default backlog quota: topicName={}", topicName, e); + } + return optional.orElseGet(() -> getBacklogQuota(topicName.getNamespace(),policyPath)); + } + + public long getBacklogQuotaLimit(TopicName topicName) { + return getBacklogQuota(topicName).getLimit(); } /** @@ -83,10 +103,7 @@ public long getBacklogQuotaLimit(String namespace) { */ public void handleExceededBacklogQuota(PersistentTopic persistentTopic) { TopicName topicName = TopicName.get(persistentTopic.getName()); - String namespace = topicName.getNamespace(); - String policyPath = AdminResource.path(POLICIES, namespace); - - BacklogQuota quota = getBacklogQuota(namespace, policyPath); + BacklogQuota quota = getBacklogQuota(topicName); log.info("Backlog quota exceeded for topic [{}]. Applying [{}] policy", persistentTopic.getName(), quota.getPolicy()); switch (quota.getPolicy()) { 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 8983238d2dbed..5f4dc63cade30 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 @@ -1831,11 +1831,7 @@ public CompletableFuture onPoliciesUpdate(Policies data) { @Override public BacklogQuota getBacklogQuota() { TopicName topicName = TopicName.get(this.getName()); - String namespace = topicName.getNamespace(); - String policyPath = AdminResource.path(POLICIES, namespace); - - BacklogQuota backlogQuota = brokerService.getBacklogQuotaManager().getBacklogQuota(namespace, policyPath); - return backlogQuota; + return brokerService.getBacklogQuotaManager().getBacklogQuota(topicName); } /** @@ -1866,7 +1862,7 @@ && isBacklogExceeded()) { */ public boolean isBacklogExceeded() { TopicName topicName = TopicName.get(getName()); - long backlogQuotaLimitInBytes = brokerService.getBacklogQuotaManager().getBacklogQuotaLimit(topicName.getNamespace()); + long backlogQuotaLimitInBytes = brokerService.getBacklogQuotaManager().getBacklogQuotaLimit(topicName); if (backlogQuotaLimitInBytes < 0) { return false; } diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/TopicPolicies.java b/pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/TopicPolicies.java index 56fbf8368eed8..f9fcaf860747e 100644 --- a/pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/TopicPolicies.java +++ b/pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/TopicPolicies.java @@ -37,7 +37,7 @@ @AllArgsConstructor public class TopicPolicies { - private Map backLogQuotaMap = Maps.newHashMap(); + private Map backLogQuotaMap = Maps.newHashMap(); private PersistencePolicies persistence = null; private RetentionPolicies retentionPolicies = null; private Boolean deduplicationEnabled = null; From 6286822508bc31f57e26aa5e53f6205fa7cbdb27 Mon Sep 17 00:00:00 2001 From: "jianyun.zhao" Date: Thu, 23 Jul 2020 11:42:18 +0800 Subject: [PATCH 2/9] Add management functions to topic backquote --- .../pulsar/broker/admin/AdminResource.java | 21 ++ .../admin/impl/PersistentTopicsBase.java | 92 ++++++++ .../broker/admin/v2/PersistentTopics.java | 47 ++++ .../broker/service/BacklogQuotaManager.java | 13 +- .../broker/admin/TopicBacklogQuotaTest.java | 215 ++++++++++++++++++ .../apache/pulsar/client/admin/Topics.java | 79 +++++++ .../client/admin/internal/TopicsImpl.java | 38 ++++ .../apache/pulsar/admin/cli/CmdTopics.java | 62 +++++ 8 files changed, 561 insertions(+), 6 deletions(-) create mode 100644 pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/TopicBacklogQuotaTest.java 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 9124ebb890e98..007571918a26e 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 @@ -28,6 +28,7 @@ import java.net.URI; import java.util.ArrayList; import java.util.List; +import java.util.Optional; import java.util.Set; import java.util.concurrent.CompletableFuture; import java.util.concurrent.CountDownLatch; @@ -66,6 +67,7 @@ import org.apache.pulsar.common.policies.data.Policies; import org.apache.pulsar.common.policies.data.SubscribeRate; import org.apache.pulsar.common.policies.data.TenantInfo; +import org.apache.pulsar.common.policies.data.TopicPolicies; import org.apache.pulsar.common.policies.impl.NamespaceIsolationPolicies; import org.apache.pulsar.common.util.Codec; import org.apache.pulsar.common.util.FutureUtil; @@ -508,6 +510,25 @@ protected BacklogQuota namespaceBacklogQuota(String namespace, String namespaceP return pulsar().getBrokerService().getBacklogQuotaManager().getBacklogQuota(namespace, namespacePath); } + protected Optional getTopicPolicies(TopicName topicName) { + try { + checkTopicLevelPolicyEnable(); + return Optional.ofNullable(pulsar().getTopicPoliciesService().getTopicPolicies(topicName)); + } catch (RestException re) { + throw re; + } catch (Exception e) { + log.error("[{}] Failed to get topic policies {}", clientAppId(), topicName, e); + throw new RestException(e); + } + } + + protected void checkTopicLevelPolicyEnable() { + if (!config().isTopicLevelPoliciesEnabled()) { + throw new RestException(Status.METHOD_NOT_ALLOWED, + "Topic level policies is disabled, to enable the topic level policy and retry."); + } + } + protected DispatchRate dispatchRate() { return new DispatchRate( pulsar().getConfiguration().getDispatchThrottlingRatePerTopicInMsg(), 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 cdbca4542af6d..f05e7563b6320 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 @@ -71,6 +71,7 @@ import org.apache.pulsar.broker.admin.AdminResource; import org.apache.pulsar.broker.admin.ZkAdminPaths; import org.apache.pulsar.broker.authentication.AuthenticationDataSource; +import org.apache.pulsar.broker.service.BrokerServiceException; import org.apache.pulsar.broker.service.BrokerServiceException.AlreadyRunningException; import org.apache.pulsar.broker.service.BrokerServiceException.NotAllowedException; import org.apache.pulsar.broker.service.BrokerServiceException.SubscriptionBusyException; @@ -104,12 +105,15 @@ import org.apache.pulsar.common.partition.PartitionedTopicMetadata; import org.apache.pulsar.common.policies.data.AuthAction; import org.apache.pulsar.common.policies.data.AuthPolicies; +import org.apache.pulsar.common.policies.data.BacklogQuota; import org.apache.pulsar.common.policies.data.PartitionedTopicInternalStats; import org.apache.pulsar.common.policies.data.PartitionedTopicStats; import org.apache.pulsar.common.policies.data.PersistentOfflineTopicStats; import org.apache.pulsar.common.policies.data.PersistentTopicInternalStats; import org.apache.pulsar.common.policies.data.Policies; +import org.apache.pulsar.common.policies.data.RetentionPolicies; import org.apache.pulsar.common.policies.data.SubscriptionStats; +import org.apache.pulsar.common.policies.data.TopicPolicies; import org.apache.pulsar.common.policies.data.TopicStats; import org.apache.pulsar.common.util.DateFormatter; import org.apache.pulsar.common.util.FutureUtil; @@ -2000,6 +2004,94 @@ protected PersistentOfflineTopicStats internalGetBacklog(boolean authoritative) return offlineTopicStats; } + protected void internalSetBacklogQuota(AsyncResponse asyncResponse, BacklogQuota.BacklogQuotaType backlogQuotaType, BacklogQuota backlogQuota) { + validateAdminAccessForTenant(namespaceName.getTenant()); + validatePoliciesReadOnlyAccess(); + if (topicName.isGlobal()) { + validateGlobalNamespaceOwnership(namespaceName); + } + if (backlogQuotaType == null) { + backlogQuotaType = BacklogQuota.BacklogQuotaType.destination_storage; + } + checkTopicLevelPolicyEnable(); + TopicPolicies topicPolicies; + try { + topicPolicies = pulsar().getTopicPoliciesService().getTopicPolicies(topicName); + } catch (BrokerServiceException.TopicPoliciesCacheNotInitException e) { + log.warn("Topic {} policies cache have not init.", topicName); + asyncResponse.resume(new RestException(e)); + return; + } + if (topicPolicies == null){ + topicPolicies = new TopicPolicies(); + } + + RetentionPolicies retentionPolicies = getRetentionPolicies(topicName, topicPolicies); + if(!checkQuotas(backlogQuota,retentionPolicies)){ + log.warn( + "[{}] Failed to update backlog configuration for topic {}: conflicts with retention quota", + clientAppId(), topicName); + throw new RestException(Status.PRECONDITION_FAILED, + "Backlog Quota exceeds configured retention quota for topic. Please increase retention quota and retry"); + } + + if(backlogQuota!=null){ + topicPolicies.getBackLogQuotaMap().put(backlogQuotaType.name(), backlogQuota); + }else { + topicPolicies.getBackLogQuotaMap().remove(backlogQuotaType.name()); + } + Map backLogQuotaMap = topicPolicies.getBackLogQuotaMap(); + 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={}, map={}", + clientAppId(), + namespaceName, + topicName.getLocalName(), + jsonMapper().writeValueAsString(backLogQuotaMap)); + } catch (JsonProcessingException ignore) { } + asyncResponse.resume(Response.noContent().build()); + } + }); + } + + private RetentionPolicies getRetentionPolicies(TopicName topicName, TopicPolicies topicPolicies) { + RetentionPolicies retentionPolicies = topicPolicies.getRetentionPolicies(); + if (retentionPolicies == null){ + try { + retentionPolicies = getNamespacePoliciesAsync(topicName.getNamespaceObject()) + .thenApply(policies -> policies.retention_policies) + .get(1L, TimeUnit.SECONDS); + } catch (Exception e) { + throw new RestException(e); + } + } + return retentionPolicies; + } + + protected void internalRemoveBacklogQuota(AsyncResponse asyncResponse, + BacklogQuota.BacklogQuotaType backlogQuotaType) { + internalSetBacklogQuota(asyncResponse, backlogQuotaType, null); + } + + private boolean checkQuotas(BacklogQuota quota, RetentionPolicies retention) { + if (retention==null||retention.getRetentionSizeInMB() == 0 || + retention.getRetentionSizeInMB() == -1) { + return true; + } + if (quota == null) { + quota = pulsar().getBrokerService().getBacklogQuotaManager().getDefaultQuota(); + } + if (quota.getLimit() >= ( retention.getRetentionSizeInMB() * 1024 * 1024)) { + return false; + } + return true; + } + 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 fbfea19662411..d605fd52ac263 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 @@ -38,6 +38,7 @@ import javax.ws.rs.core.MediaType; import javax.ws.rs.core.Response; +import com.google.common.collect.Maps; import org.apache.pulsar.broker.admin.impl.PersistentTopicsBase; import org.apache.pulsar.broker.web.RestException; import org.apache.pulsar.client.admin.LongRunningProcessStatus; @@ -46,8 +47,10 @@ import org.apache.pulsar.client.impl.MessageIdImpl; import org.apache.pulsar.common.partition.PartitionedTopicMetadata; import org.apache.pulsar.common.policies.data.AuthAction; +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.TopicPolicies; import org.apache.pulsar.common.policies.data.TopicStats; import io.swagger.annotations.Api; @@ -977,6 +980,50 @@ public PersistentOfflineTopicStats getBacklog( return internalGetBacklog(authoritative); } + @GET + @Path("/{tenant}/{namespace}/{topic}/backlogQuotaMap") + @ApiOperation(value = "Get backlog quota map on a topic.") + @ApiResponses(value = { @ApiResponse(code = 403, message = "Don't have admin permission"), + @ApiResponse(code = 404, message = "Topic policy does not exist"), + @ApiResponse(code = 405, message = "Topic level policy is disabled, to enable the topic level policy and retry")}) + public Map getBacklogQuotaMap(@PathParam("tenant") String tenant, + @PathParam("namespace") String namespace, + @PathParam("topic") @Encoded String encodedTopic) { + validateTopicName(tenant, namespace, encodedTopic); + return getTopicPolicies(topicName) + .map(TopicPolicies::getBackLogQuotaMap) + .orElse(Maps.newHashMap()); + } + + @POST + @Path("/{tenant}/{namespace}/{topic}/backlogQuota") + @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"), + @ApiResponse(code = 405, message = "Topic level policy is disabled, to enable the topic level policy and retry"), + @ApiResponse(code = 412, message = "Specified backlog quota exceeds retention quota. Increase retention quota and retry request") }) + public void setBacklogQuota(@Suspended final AsyncResponse asyncResponse, @PathParam("tenant") String tenant, @PathParam("namespace") String namespace, + @PathParam("topic") @Encoded String encodedTopic, + @QueryParam("backlogQuotaType") BacklogQuota.BacklogQuotaType backlogQuotaType, BacklogQuota backlogQuota) { + validateTopicName(tenant, namespace, encodedTopic); + internalSetBacklogQuota(asyncResponse, backlogQuotaType, backlogQuota); + } + + @DELETE + @Path("/{tenant}/{namespace}/{topic}/backlogQuota") + @ApiOperation(value = "Remove a backlog quota policy from 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 removeBacklogQuota(@Suspended final AsyncResponse asyncResponse, @PathParam("tenant") String tenant, @PathParam("namespace") String namespace, + @PathParam("topic") @Encoded String encodedTopic, + @QueryParam("backlogQuotaType") BacklogQuota.BacklogQuotaType backlogQuotaType) { + validateTopicName(tenant, namespace, encodedTopic); + internalRemoveBacklogQuota(asyncResponse, backlogQuotaType); + } + @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/BacklogQuotaManager.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BacklogQuotaManager.java index 2315afc3f774a..8eafa5d574332 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BacklogQuotaManager.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BacklogQuotaManager.java @@ -47,7 +47,7 @@ public class BacklogQuotaManager { private static final Logger log = LoggerFactory.getLogger(BacklogQuotaManager.class); private final BacklogQuota defaultQuota; private final ZooKeeperDataCache zkCache; - private final TopicPoliciesService topicPoliciesService; + private final PulsarService pulsar; private final boolean isTopicLevelPoliciesEnable; @@ -57,7 +57,7 @@ public BacklogQuotaManager(PulsarService pulsar) { pulsar.getConfiguration().getBacklogQuotaDefaultLimitGB() * 1024 * 1024 * 1024, pulsar.getConfiguration().getBacklogQuotaDefaultRetentionPolicy()); this.zkCache = pulsar.getConfigurationCache().policiesCache(); - this.topicPoliciesService = pulsar.getTopicPoliciesService(); + this.pulsar = pulsar; } public BacklogQuota getDefaultQuota() { @@ -81,14 +81,15 @@ public BacklogQuota getBacklogQuota(TopicName topicName) { return getBacklogQuota(topicName.getNamespace(),policyPath); } - Optional optional = Optional.empty(); try { - TopicPolicies topicPolicies = topicPoliciesService.getTopicPolicies(topicName); - optional = Optional.ofNullable(topicPolicies.getBackLogQuotaMap().get(BacklogQuotaType.destination_storage.name())); + return Optional.ofNullable(pulsar.getTopicPoliciesService().getTopicPolicies(topicName)) + .map(TopicPolicies::getBackLogQuotaMap) + .map(map -> map.get(BacklogQuotaType.destination_storage.name())) + .orElseGet(() -> getBacklogQuota(topicName.getNamespace(),policyPath)); } catch (Exception e) { log.error("Failed to read policies data, will apply the default backlog quota: topicName={}", topicName, e); } - return optional.orElseGet(() -> getBacklogQuota(topicName.getNamespace(),policyPath)); + return getBacklogQuota(topicName.getNamespace(),policyPath); } public long getBacklogQuotaLimit(TopicName topicName) { 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/TopicBacklogQuotaTest.java new file mode 100644 index 0000000000000..7c2a2167d067d --- /dev/null +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/TopicBacklogQuotaTest.java @@ -0,0 +1,215 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.pulsar.broker.admin; + +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.Test; + +@Slf4j +public class TopicBacklogQuotaTest extends MockedPulsarServiceBaseTest { + + private static final Logger LOG = LoggerFactory.getLogger(TopicBacklogQuotaTest.class); + + private final String testTenant = "my-tenant"; + + private final String testNamespace = "my-namespace"; + + private final String myNamespace = testTenant + "/" + testNamespace; + + private final String backlogQuotaTopic = "persistent://" + myNamespace + "/test-set-backlog-quota"; + + public void enableTopicLevelPolicies() throws Exception { + this.conf.setSystemTopicEnabled(true); + this.conf.setTopicLevelPoliciesEnabled(true); + 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")); + } + + 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")); + } + + @Override + protected void setup() throws Exception { + + } + + @AfterMethod + @Override + public void cleanup() throws Exception { + super.internalCleanup(); + } + + @Test + public void testSetBacklogQuota() throws Exception { + enableTopicLevelPolicies(); + 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); + + admin.topics().setBacklogQuota(backlogQuotaTopic, backlogQuota); + log.info("Backlog quota set success on topic: {}", backlogQuotaTopic); + + Thread.sleep(3000); + + BacklogQuota getBacklogQuota = admin.topics().getBacklogQuotaMap(backlogQuotaTopic) + .get(BacklogQuota.BacklogQuotaType.destination_storage); + log.info("Backlog quota {} get on topic: {}", getBacklogQuota, backlogQuotaTopic); + 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); + Assert.assertEquals(backlogQuotaInManager, backlogQuota); + + admin.topics().deletePartitionedTopic(backlogQuotaTopic, true); + } + + @Test + public void testRemoveBacklogQuota() throws Exception { + enableTopicLevelPolicies(); + 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); + admin.topics().setBacklogQuota(backlogQuotaTopic, backlogQuota); + log.info("Backlog quota set success on topic: {}", backlogQuotaTopic); + + Thread.sleep(3000); + + BacklogQuota getBacklogQuota = admin.topics().getBacklogQuotaMap(backlogQuotaTopic) + .get(BacklogQuota.BacklogQuotaType.destination_storage); + log.info("Backlog quota {} get on topic: {}", getBacklogQuota, backlogQuotaTopic); + 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); + Assert.assertEquals(backlogQuota, backlogQuotaInManager); + + admin.topics().removeBacklogQuota(backlogQuotaTopic); + getBacklogQuota = admin.topics().getBacklogQuotaMap(backlogQuotaTopic) + .get(BacklogQuota.BacklogQuotaType.destination_storage); + log.info("Backlog quota {} get on topic: {} after remove", getBacklogQuota, backlogQuotaTopic); + Assert.assertNull(getBacklogQuota); + + backlogQuotaInManager = backlogQuotaManager.getBacklogQuota(TopicName.get(backlogQuotaTopic)); + log.info("Backlog quota {} in backlog quota manager on topic: {} after remove", backlogQuotaInManager, + backlogQuotaTopic); + Assert.assertEquals(backlogQuotaManager.getDefaultQuota(), backlogQuotaInManager); + + admin.topics().deletePartitionedTopic(backlogQuotaTopic, true); + } + + @Test + public void testCheckQuota() throws Exception { + enableTopicLevelPolicies(); + admin.topics().createPartitionedTopic(backlogQuotaTopic, 2); + RetentionPolicies retentionPolicies = new RetentionPolicies(10, 10); + String namespace = TopicName.get(backlogQuotaTopic).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); + try { + admin.topics().setBacklogQuota(backlogQuotaTopic, backlogQuota); + Assert.fail(); + } catch (PulsarAdminException e) { + Assert.assertEquals(e.getStatusCode(), 412); + } + + backlogQuota = + new BacklogQuota(10 * 1024 * 1024 + 1, BacklogQuota.RetentionPolicy.consumer_backlog_eviction); + log.info("Backlog quota: {} will set to the topic: {}", backlogQuota, backlogQuotaTopic); + try { + admin.topics().setBacklogQuota(backlogQuotaTopic, backlogQuota); + Assert.fail(); + } catch (PulsarAdminException e) { + Assert.assertEquals(e.getStatusCode(), 412); + } + + 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); + Thread.sleep(3000); + BacklogQuota getBacklogQuota = admin.topics().getBacklogQuotaMap(backlogQuotaTopic) + .get(BacklogQuota.BacklogQuotaType.destination_storage); + log.info("Backlog quota {} get on topic: {} after remove", getBacklogQuota, backlogQuotaTopic); + Assert.assertEquals(getBacklogQuota, backlogQuota); + + admin.topics().deletePartitionedTopic(backlogQuotaTopic, 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); + + try { + admin.topics().setBacklogQuota(backlogQuotaTopic, backlogQuota); + Assert.fail(); + } catch (PulsarAdminException e) { + Assert.assertEquals(e.getStatusCode(), 405); + } + + try { + admin.topics().removeBacklogQuota(backlogQuotaTopic); + Assert.fail(); + } catch (PulsarAdminException e) { + Assert.assertEquals(e.getStatusCode(), 405); + } + + try { + admin.topics().getBacklogQuotaMap(backlogQuotaTopic); + Assert.fail(); + } catch (PulsarAdminException e) { + Assert.assertEquals(e.getStatusCode(), 405); + } + } +} 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 f4e997d8774d9..bf6ac821b0a73 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 @@ -34,6 +34,7 @@ import org.apache.pulsar.client.api.MessageId; import org.apache.pulsar.common.partition.PartitionedTopicMetadata; import org.apache.pulsar.common.policies.data.AuthAction; +import org.apache.pulsar.common.policies.data.BacklogQuota; import org.apache.pulsar.common.policies.data.PartitionedTopicInternalStats; import org.apache.pulsar.common.policies.data.PartitionedTopicStats; import org.apache.pulsar.common.policies.data.PersistentTopicInternalStats; @@ -1354,4 +1355,82 @@ void createSubscription(String topic, String subscriptionName, MessageId message * @return */ CompletableFuture getLastMessageIdAsync(String topic); + + /** + * Get backlog quota map for a topic. + * Response example: + * + * 
+     * 
+     *  {
+     *      "namespace_memory" : {
+     *          "limit" : "134217728",
+     *          "policy" : "consumer_backlog_eviction"
+     *      },
+     *      "destination_storage" : {
+     *          "limit" : "-1",
+     *          "policy" : "producer_exception"
+     *      }
+     *  }
+     * 
+     *
+ * + * @param topic + * Topic name + * + * @throws NotAuthorizedException + * Permission denied + * @throws NotFoundException + * Topic does not exist + * @throws PulsarAdminException + * Unexpected error + */ + Map getBacklogQuotaMap(String topic) throws PulsarAdminException; + + /** + * Set a backlog quota for a topic. + * The backlog quota can be set on this resource: + * + *

+ * Request parameter example: + *

+ * + * 
+     * 
+     * {
+     *     "limit" : "134217728",
+     *     "policy" : "consumer_backlog_eviction"
+     * }
+     * 
+     *
+ * + * @param topic + * Topic name + * @param backlogQuota + * the new BacklogQuota + * + * @throws NotAuthorizedException + * Don't have admin permission + * @throws NotFoundException + * Topic does not exist + * @throws PulsarAdminException + * Unexpected error + */ + void setBacklogQuota(String topic, BacklogQuota backlogQuota) throws PulsarAdminException; + + /** + * Remove a backlog quota policy from a topic. + * The namespace backlog policy will fall back to the default. + * + * @param topic + * Topic name + * + * @throws NotAuthorizedException + * Don't have admin permission + * @throws NotFoundException + * Topic does not exist + * @throws PulsarAdminException + * Unexpected error + */ + void removeBacklogQuota(String topic) throws PulsarAdminException; } 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 f6d4fae78ee15..b5ac14467c145 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 @@ -45,6 +45,7 @@ import javax.ws.rs.client.Entity; import javax.ws.rs.client.InvocationCallback; import javax.ws.rs.client.WebTarget; +import javax.ws.rs.core.GenericType; import javax.ws.rs.core.MediaType; import javax.ws.rs.core.MultivaluedMap; import javax.ws.rs.core.Response; @@ -69,6 +70,8 @@ import org.apache.pulsar.common.naming.TopicName; import org.apache.pulsar.common.partition.PartitionedTopicMetadata; import org.apache.pulsar.common.policies.data.AuthAction; +import org.apache.pulsar.common.policies.data.BacklogQuota; +import org.apache.pulsar.common.policies.data.BacklogQuota.BacklogQuotaType; import org.apache.pulsar.common.policies.data.ErrorData; import org.apache.pulsar.common.policies.data.PartitionedTopicInternalStats; import org.apache.pulsar.common.policies.data.PartitionedTopicStats; @@ -1391,5 +1394,40 @@ public void failed(Throwable throwable) { return future; } + @Override + public Map getBacklogQuotaMap(String topic) throws PulsarAdminException { + try { + TopicName tn = validateTopic(topic); + WebTarget path = topicPath(tn, "backlogQuotaMap"); + return request(path).get(new GenericType>() { + }); + } catch (Exception e) { + throw getApiException(e); + } + } + + @Override + public void setBacklogQuota(String topic, BacklogQuota backlogQuota) throws PulsarAdminException { + try { + TopicName tn = validateTopic(topic); + WebTarget path = topicPath(tn, "backlogQuota"); + request(path).post(Entity.entity(backlogQuota, MediaType.APPLICATION_JSON), ErrorData.class); + } catch (Exception e) { + throw getApiException(e); + } + } + + @Override + public void removeBacklogQuota(String topic) throws PulsarAdminException { + try { + TopicName tn = validateTopic(topic); + WebTarget path = topicPath(tn, "backlogQuota"); + request(path.queryParam("backlogQuotaType", BacklogQuotaType.destination_storage.toString())) + .delete(ErrorData.class); + } catch (Exception e) { + throw getApiException(e); + } + } + 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 733a31d06d8a8..4f2e39986f2f0 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 @@ -21,6 +21,7 @@ import static org.apache.commons.lang3.StringUtils.isNotBlank; import com.beust.jcommander.Parameter; +import com.beust.jcommander.ParameterException; import com.beust.jcommander.Parameters; import com.beust.jcommander.converters.CommaParameterSplitter; import com.google.common.collect.Lists; @@ -32,6 +33,7 @@ import io.netty.buffer.ByteBufUtil; import io.netty.buffer.Unpooled; +import java.util.Arrays; import java.util.LinkedList; import java.util.List; import java.util.concurrent.ExecutionException; @@ -45,6 +47,7 @@ import org.apache.pulsar.client.api.MessageId; import org.apache.pulsar.client.impl.BatchMessageIdImpl; 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.util.RelativeTimeUtil; @@ -98,6 +101,9 @@ public CmdTopics(PulsarAdmin admin) { jcommander.addCommand("offload", new Offload()); jcommander.addCommand("offload-status", new OffloadStatusCmd()); jcommander.addCommand("last-message-id", new GetLastMessageId()); + jcommander.addCommand("get-backlog-quotas", new GetBacklogQuotaMap()); + jcommander.addCommand("set-backlog-quota", new SetBacklogQuota()); + jcommander.addCommand("remove-backlog-quota", new RemoveBacklogQuota()); } @Parameters(commandDescription = "Get the list of topics under a namespace.") @@ -818,4 +824,60 @@ void run() throws PulsarAdminException { print(topics.getLastMessageId(persistentTopic)); } } + + @Parameters(commandDescription = "Get the backlog quota policies for a topic") + private class GetBacklogQuotaMap 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().getBacklogQuotaMap(persistentTopic)); + } + } + + @Parameters(commandDescription = "Set a backlog quota policy for a topic") + private class SetBacklogQuota extends CliCommand { + @Parameter(description = "persistent://tenant/namespace/topic", required = true) + private java.util.List params; + + @Parameter(names = { "-l", "--limit" }, description = "Size limit (eg: 10M, 16G)", required = true) + private String limitStr; + + @Parameter(names = { "-p", "--policy" }, description = "Retention policy to enforce when the limit is reached. " + + "Valid options are: [producer_request_hold, producer_exception, consumer_backlog_eviction]", required = true) + private String policyStr; + + @Override + void run() throws PulsarAdminException { + BacklogQuota.RetentionPolicy policy; + long limit; + + try { + policy = BacklogQuota.RetentionPolicy.valueOf(policyStr); + } catch (IllegalArgumentException e) { + throw new ParameterException(String.format("Invalid retention policy type '%s'. Valid options are: %s", + policyStr, Arrays.toString(BacklogQuota.RetentionPolicy.values()))); + } + + limit = validateSizeString(limitStr); + + String persistentTopic = validatePersistentTopic(params); + admin.topics().setBacklogQuota(persistentTopic, new BacklogQuota(limit, policy)); + } + } + + @Parameters(commandDescription = "Remove a backlog quota policy from a topic") + private class RemoveBacklogQuota 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().removeBacklogQuota(persistentTopic); + } + } } From afddf87c3bff31f019ba6fa50c35e1e1536ee766 Mon Sep 17 00:00:00 2001 From: "jianyun.zhao" Date: Thu, 23 Jul 2020 22:14:41 +0800 Subject: [PATCH 3/9] fix TopicBacklogQuotaTest header --- .../org/apache/pulsar/broker/admin/TopicBacklogQuotaTest.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) 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/TopicBacklogQuotaTest.java index 7c2a2167d067d..16e3f6f7ccc65 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/TopicBacklogQuotaTest.java @@ -7,7 +7,7 @@ * "License"); you may not use this file except in compliance * with the License. You may obtain a copy of the License at * - * http://www.apache.org/licenses/LICENSE-2.0 + * http://www.apache.org/licenses/LICENSE-2.0 * * Unless required by applicable law or agreed to in writing, * software distributed under the License is distributed on an From 3ccc2151a311a3eafcbfa64b98ac6e5c879a6a68 Mon Sep 17 00:00:00 2001 From: "jianyun.zhao" Date: Thu, 23 Jul 2020 22:57:19 +0800 Subject: [PATCH 4/9] fix code style --- .../pulsar/broker/admin/impl/PersistentTopicsBase.java | 7 ++++--- 1 file changed, 4 insertions(+), 3 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 f05e7563b6320..59e67bd95bb78 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 @@ -2031,11 +2031,12 @@ protected void internalSetBacklogQuota(AsyncResponse asyncResponse, BacklogQuota log.warn( "[{}] Failed to update backlog configuration for topic {}: conflicts with retention quota", clientAppId(), topicName); - throw new RestException(Status.PRECONDITION_FAILED, - "Backlog Quota exceeds configured retention quota for topic. Please increase retention quota and retry"); + asyncResponse.resume(new RestException(Status.PRECONDITION_FAILED, + "Backlog Quota exceeds configured retention quota for topic. " + + "Please increase retention quota and retry")); } - if(backlogQuota!=null){ + if(backlogQuota != null){ topicPolicies.getBackLogQuotaMap().put(backlogQuotaType.name(), backlogQuota); }else { topicPolicies.getBackLogQuotaMap().remove(backlogQuotaType.name()); From d3855de1a74cf8856c214d46cf27d053eb4b5a7c Mon Sep 17 00:00:00 2001 From: "jianyun.zhao" Date: Thu, 23 Jul 2020 23:00:13 +0800 Subject: [PATCH 5/9] Change the return value of `getBacklogQuotaMap` to make it consistent with the namespace --- .../pulsar/broker/admin/v2/PersistentTopics.java | 10 +++++++++- 1 file changed, 9 insertions(+), 1 deletion(-) 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 d605fd52ac263..b939bde85eb09 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 @@ -18,6 +18,7 @@ */ package org.apache.pulsar.broker.admin.v2; +import java.util.HashMap; import java.util.List; import java.util.Map; import java.util.Set; @@ -986,12 +987,19 @@ public PersistentOfflineTopicStats getBacklog( @ApiResponses(value = { @ApiResponse(code = 403, message = "Don't have admin permission"), @ApiResponse(code = 404, message = "Topic policy does not exist"), @ApiResponse(code = 405, message = "Topic level policy is disabled, to enable the topic level policy and retry")}) - public Map getBacklogQuotaMap(@PathParam("tenant") String tenant, + public Map getBacklogQuotaMap(@PathParam("tenant") String tenant, @PathParam("namespace") String namespace, @PathParam("topic") @Encoded String encodedTopic) { validateTopicName(tenant, namespace, encodedTopic); return getTopicPolicies(topicName) .map(TopicPolicies::getBackLogQuotaMap) + .map(map -> { + HashMap hashMap = Maps.newHashMap(); + map.forEach((key,value) -> { + hashMap.put(BacklogQuota.BacklogQuotaType.valueOf(key),value); + }); + return hashMap; + }) .orElse(Maps.newHashMap()); } From cf248c14c4d8fedcb98bb9380d6d332449e5cafc Mon Sep 17 00:00:00 2001 From: "jianyun.zhao" Date: Thu, 23 Jul 2020 23:00:58 +0800 Subject: [PATCH 6/9] add waiting interval for test --- .../org/apache/pulsar/broker/admin/TopicBacklogQuotaTest.java | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) 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/TopicBacklogQuotaTest.java index 16e3f6f7ccc65..5e3e4a97de6a6 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/TopicBacklogQuotaTest.java @@ -159,7 +159,7 @@ public void testCheckQuota() throws Exception { } catch (PulsarAdminException e) { Assert.assertEquals(e.getStatusCode(), 412); } - + 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); @@ -169,7 +169,7 @@ public void testCheckQuota() throws Exception { } catch (PulsarAdminException e) { Assert.assertEquals(e.getStatusCode(), 412); } - + 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); From 3dc2dab80811451de6efa3464c0de3206228885e Mon Sep 17 00:00:00 2001 From: "jianyun.zhao" Date: Fri, 24 Jul 2020 14:50:40 +0800 Subject: [PATCH 7/9] extract checkQuota to AdminResource --- .../pulsar/broker/admin/AdminResource.java | 15 +++++++++++++++ .../broker/admin/impl/NamespacesBase.java | 17 ++++------------- .../broker/admin/impl/PersistentTopicsBase.java | 16 +--------------- 3 files changed, 20 insertions(+), 28 deletions(-) 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 007571918a26e..590c03282d50b 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 @@ -65,6 +65,7 @@ import org.apache.pulsar.common.policies.data.FailureDomain; import org.apache.pulsar.common.policies.data.LocalPolicies; import org.apache.pulsar.common.policies.data.Policies; +import org.apache.pulsar.common.policies.data.RetentionPolicies; import org.apache.pulsar.common.policies.data.SubscribeRate; import org.apache.pulsar.common.policies.data.TenantInfo; import org.apache.pulsar.common.policies.data.TopicPolicies; @@ -522,6 +523,20 @@ protected Optional getTopicPolicies(TopicName topicName) { } } + protected boolean checkBacklogQuota(BacklogQuota quota, RetentionPolicies retention) { + if (retention == null || retention.getRetentionSizeInMB() == 0 || + retention.getRetentionSizeInMB() == -1) { + return true; + } + if (quota == null) { + quota = pulsar().getBrokerService().getBacklogQuotaManager().getDefaultQuota(); + } + if (quota.getLimit() >= ( retention.getRetentionSizeInMB() * 1024 * 1024)) { + return false; + } + return true; + } + protected void checkTopicLevelPolicyEnable() { if (!config().isTopicLevelPoliciesEnabled()) { throw new RestException(Status.METHOD_NOT_ALLOWED, diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/NamespacesBase.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/NamespacesBase.java index ec2c1bd028e30..13e5e55230f30 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/NamespacesBase.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/NamespacesBase.java @@ -2008,21 +2008,12 @@ protected RetentionPolicies internalGetRetention() { } private boolean checkQuotas(Policies policies, RetentionPolicies retention) { - Map backlog_quota_map = policies.backlog_quota_map; - if (backlog_quota_map.isEmpty() || retention.getRetentionSizeInMB() == 0 || retention.getRetentionSizeInMB() == -1) { + Map backlogQuotaMap = policies.backlog_quota_map; + if (backlogQuotaMap.isEmpty()) { return true; } - BacklogQuota quota = backlog_quota_map.get(BacklogQuotaType.destination_storage); - if (quota == null) { - quota = pulsar().getBrokerService().getBacklogQuotaManager().getDefaultQuota(); - } - if (quota.getLimit() < 0 && (retention.getRetentionSizeInMB() > 0 || retention.getRetentionTimeInMinutes() > 0)) { - return false; - } - if (quota.getLimit() >= (retention.getRetentionSizeInMB() * 1024 * 1024)) { - return false; - } - return true; + BacklogQuota quota = backlogQuotaMap.get(BacklogQuotaType.destination_storage); + return checkBacklogQuota(quota, retention); } private void clearBacklog(NamespaceName nsName, String bundleRange, String subscription) { 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 59e67bd95bb78..40663993e21c8 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 @@ -2027,7 +2027,7 @@ protected void internalSetBacklogQuota(AsyncResponse asyncResponse, BacklogQuota } RetentionPolicies retentionPolicies = getRetentionPolicies(topicName, topicPolicies); - if(!checkQuotas(backlogQuota,retentionPolicies)){ + if(!checkBacklogQuota(backlogQuota,retentionPolicies)){ log.warn( "[{}] Failed to update backlog configuration for topic {}: conflicts with retention quota", clientAppId(), topicName); @@ -2079,20 +2079,6 @@ protected void internalRemoveBacklogQuota(AsyncResponse asyncResponse, internalSetBacklogQuota(asyncResponse, backlogQuotaType, null); } - private boolean checkQuotas(BacklogQuota quota, RetentionPolicies retention) { - if (retention==null||retention.getRetentionSizeInMB() == 0 || - retention.getRetentionSizeInMB() == -1) { - return true; - } - if (quota == null) { - quota = pulsar().getBrokerService().getBacklogQuotaManager().getDefaultQuota(); - } - if (quota.getLimit() >= ( retention.getRetentionSizeInMB() * 1024 * 1024)) { - return false; - } - return true; - } - protected MessageId internalTerminate(boolean authoritative) { if (topicName.isGlobal()) { validateGlobalNamespaceOwnership(namespaceName); From 6bd3df4dc83181a7673a095bb83ac32feedf2354 Mon Sep 17 00:00:00 2001 From: "jianyun.zhao" Date: Thu, 30 Jul 2020 17:02:45 +0800 Subject: [PATCH 8/9] add waiting interval for test --- .../apache/pulsar/broker/admin/TopicBacklogQuotaTest.java | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) 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/TopicBacklogQuotaTest.java index 5e3e4a97de6a6..a743507d539fd 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/TopicBacklogQuotaTest.java @@ -84,6 +84,7 @@ public void cleanup() throws Exception { public void testSetBacklogQuota() throws Exception { enableTopicLevelPolicies(); admin.topics().createPartitionedTopic(backlogQuotaTopic, 2); + Thread.sleep(3000); BacklogQuota backlogQuota = new BacklogQuota(1024, BacklogQuota.RetentionPolicy.consumer_backlog_eviction); log.info("Backlog quota: {} will set to the topic: {}", backlogQuota, backlogQuotaTopic); @@ -92,7 +93,6 @@ public void testSetBacklogQuota() throws Exception { log.info("Backlog quota set success on topic: {}", backlogQuotaTopic); Thread.sleep(3000); - BacklogQuota getBacklogQuota = admin.topics().getBacklogQuotaMap(backlogQuotaTopic) .get(BacklogQuota.BacklogQuotaType.destination_storage); log.info("Backlog quota {} get on topic: {}", getBacklogQuota, backlogQuotaTopic); @@ -110,6 +110,7 @@ public void testSetBacklogQuota() throws Exception { public void testRemoveBacklogQuota() throws Exception { enableTopicLevelPolicies(); admin.topics().createPartitionedTopic(backlogQuotaTopic, 2); + Thread.sleep(3000); BacklogQuota backlogQuota = new BacklogQuota(1024, BacklogQuota.RetentionPolicy.consumer_backlog_eviction); log.info("Backlog quota: {} will set to the topic: {}", backlogQuota, backlogQuotaTopic); @@ -117,7 +118,6 @@ public void testRemoveBacklogQuota() throws Exception { log.info("Backlog quota set success on topic: {}", backlogQuotaTopic); Thread.sleep(3000); - BacklogQuota getBacklogQuota = admin.topics().getBacklogQuotaMap(backlogQuotaTopic) .get(BacklogQuota.BacklogQuotaType.destination_storage); log.info("Backlog quota {} get on topic: {}", getBacklogQuota, backlogQuotaTopic); @@ -146,6 +146,8 @@ public void testRemoveBacklogQuota() throws Exception { public void testCheckQuota() throws Exception { enableTopicLevelPolicies(); admin.topics().createPartitionedTopic(backlogQuotaTopic, 2); + Thread.sleep(3000); + RetentionPolicies retentionPolicies = new RetentionPolicies(10, 10); String namespace = TopicName.get(backlogQuotaTopic).getNamespace(); admin.namespaces().setRetention(namespace, retentionPolicies); From 8d68ad87004bae8781e97748f2d9648a0ffe3e60 Mon Sep 17 00:00:00 2001 From: "jianyun.zhao" Date: Fri, 31 Jul 2020 11:16:24 +0800 Subject: [PATCH 9/9] Split test cases `TopicBacklogQuotaTest` --- .../admin/TopicBacklogQuotaDisableTest.java | 97 +++++++++++++++++++ .../broker/admin/TopicBacklogQuotaTest.java | 28 ++---- 2 files changed, 105 insertions(+), 20 deletions(-) create mode 100644 pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/TopicBacklogQuotaDisableTest.java 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/TopicBacklogQuotaDisableTest.java new file mode 100644 index 0000000000000..0f8b2cbb57c5a --- /dev/null +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/TopicBacklogQuotaDisableTest.java @@ -0,0 +1,97 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.pulsar.broker.admin; + +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); + + private final String testTenant = "my-tenant"; + + private final String testNamespace = "my-namespace"; + + private final String myNamespace = testTenant + "/" + testNamespace; + + private final String backlogQuotaTopic = "persistent://" + myNamespace + "/test-set-backlog-quota"; + + @BeforeMethod + @Override + protected void setup() 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")); + admin.topics().createPartitionedTopic(backlogQuotaTopic, 2); + } + + @AfterMethod + @Override + public void cleanup() throws Exception { + super.internalCleanup(); + } + + @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); + + try { + admin.topics().setBacklogQuota(backlogQuotaTopic, backlogQuota); + Assert.fail(); + } catch (PulsarAdminException e) { + Assert.assertEquals(e.getStatusCode(), 405); + } + + try { + admin.topics().removeBacklogQuota(backlogQuotaTopic); + Assert.fail(); + } catch (PulsarAdminException e) { + Assert.assertEquals(e.getStatusCode(), 405); + } + + try { + admin.topics().getBacklogQuotaMap(backlogQuotaTopic); + 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/TopicBacklogQuotaTest.java index a743507d539fd..1104ab6f397e6 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/TopicBacklogQuotaTest.java @@ -32,6 +32,7 @@ import org.slf4j.LoggerFactory; import org.testng.Assert; import org.testng.annotations.AfterMethod; +import org.testng.annotations.BeforeMethod; import org.testng.annotations.Test; @Slf4j @@ -47,9 +48,9 @@ public class TopicBacklogQuotaTest extends MockedPulsarServiceBaseTest { private final String backlogQuotaTopic = "persistent://" + myNamespace + "/test-set-backlog-quota"; - public void enableTopicLevelPolicies() throws Exception { + public void disableTopicLevelPolicies() throws Exception { this.conf.setSystemTopicEnabled(true); - this.conf.setTopicLevelPoliciesEnabled(true); + this.conf.setTopicLevelPoliciesEnabled(false); super.internalSetup(); admin.clusters().createCluster("test", new ClusterData(pulsar.getWebServiceAddress())); @@ -58,20 +59,18 @@ public void enableTopicLevelPolicies() throws Exception { admin.namespaces().createNamespace(testTenant + "/" + testNamespace, Sets.newHashSet("test")); } - public void disableTopicLevelPolicies() throws Exception { + @BeforeMethod + @Override + protected void setup() throws Exception { this.conf.setSystemTopicEnabled(true); - this.conf.setTopicLevelPoliciesEnabled(false); + this.conf.setTopicLevelPoliciesEnabled(true); 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")); - } - - @Override - protected void setup() throws Exception { - + admin.topics().createPartitionedTopic(backlogQuotaTopic, 2); } @AfterMethod @@ -82,9 +81,6 @@ public void cleanup() throws Exception { @Test public void testSetBacklogQuota() throws Exception { - enableTopicLevelPolicies(); - admin.topics().createPartitionedTopic(backlogQuotaTopic, 2); - Thread.sleep(3000); BacklogQuota backlogQuota = new BacklogQuota(1024, BacklogQuota.RetentionPolicy.consumer_backlog_eviction); log.info("Backlog quota: {} will set to the topic: {}", backlogQuota, backlogQuotaTopic); @@ -108,10 +104,6 @@ public void testSetBacklogQuota() throws Exception { @Test public void testRemoveBacklogQuota() throws Exception { - enableTopicLevelPolicies(); - admin.topics().createPartitionedTopic(backlogQuotaTopic, 2); - Thread.sleep(3000); - 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); @@ -144,10 +136,6 @@ public void testRemoveBacklogQuota() throws Exception { @Test public void testCheckQuota() throws Exception { - enableTopicLevelPolicies(); - admin.topics().createPartitionedTopic(backlogQuotaTopic, 2); - Thread.sleep(3000); - RetentionPolicies retentionPolicies = new RetentionPolicies(10, 10); String namespace = TopicName.get(backlogQuotaTopic).getNamespace(); admin.namespaces().setRetention(namespace, retentionPolicies);