Skip to content

[Issue 2688]Support set retention on topic level. #7747

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
codelipenghui merged 7 commits into from
Aug 11, 2020
Merged

[Issue 2688]Support set retention on topic level. #7747

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
7 commits
Select commit Hold shift + click to select a range
b133646
Support set retention on topic level.
jianyun8023 Aug 4, 2020
74c8d4d
Support remove retention on topic level.
jianyun8023 Aug 4, 2020
477d08a
Fix and rename to `TopicPoliciesTest`
jianyun8023 Aug 5, 2020
e739133
fix comment and modified test
jianyun8023 Aug 6, 2020
dbd7771
fix log class，because `TopicBacklogQuotaTest` renamed to `TopicPolici…
jianyun8023 Aug 6, 2020
c8f1043
modify the exception is throw
jianyun8023 Aug 10, 2020
bebd6f7
modify the use of asyncResponse
jianyun8023 Aug 10, 2020
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 4 additions & 0 deletions pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/AdminResource.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -517,6 +518,9 @@ protected Optional<TopicPolicies> 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);
Expand Down
63 changes: 63 additions & 0 deletions pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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();
Comment thread
jianyun8023 marked this conversation as resolved.

@zhanghaou zhanghaou Aug 13, 2020

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

checkTopicLevelPolicyEnable() is also called by getTopicPolicies(), so we can delete it here.

@jianyun8023 jianyun8023 Aug 19, 2020

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

okay, let me solve these problems

Optional<RetentionPolicies> retention = getTopicPolicies(topicName)
.map(TopicPolicies::getRetentionPolicies);
if (!retention.isPresent()) {
asyncResponse.resume(Response.noContent().build());
}else {
asyncResponse.resume(retention.get());
}
}

protected CompletableFuture<Void> internalSetRetention(RetentionPolicies retention) {
validateAdminAccessForTenant(namespaceName.getTenant());
validatePoliciesReadOnlyAccess();
if (topicName.isGlobal()) {
validateGlobalNamespaceOwnership(namespaceName);
}
checkTopicLevelPolicyEnable();

@zhanghaou zhanghaou Aug 13, 2020

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Same to the above.

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,
Comment thread
jianyun8023 marked this conversation as resolved.
"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<Void> internalRemoveRetention() {
validateAdminAccessForTenant(namespaceName.getTenant());
validatePoliciesReadOnlyAccess();
if (topicName.isGlobal()) {
validateGlobalNamespaceOwnership(namespaceName);
}
checkTopicLevelPolicyEnable();

@zhanghaou zhanghaou Aug 13, 2020

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Same to the above.

Optional<TopicPolicies> 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);
Expand Down
86 changes: 85 additions & 1 deletion pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v2/PersistentTopics.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand All @@ -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;

Expand Down Expand Up @@ -1005,7 +1007,7 @@ public Map<BacklogQuota.BacklogQuotaType, BacklogQuota> 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"),
Expand Down Expand Up @@ -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) {
Comment thread
jianyun8023 marked this conversation as resolved.
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") })
Comment thread
jianyun8023 marked this conversation as resolved.
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 "
Expand Down
36 changes: 22 additions & 14 deletions pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -1757,28 +1758,35 @@ public void checkBackloggedCursors() {
*/
private boolean shouldTopicBeRetained() {
TopicName name = TopicName.get(topic);
RetentionPolicies retentionPolicies = null;
try {
Optional<Policies> 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
Expand Down
2 changes: 0 additions & 2 deletions pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/TopicMessageTTLTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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";
Expand Down
40 changes: 27 additions & 13 deletions ...r/admin/TopicBacklogQuotaDisableTest.java → ...roker/admin/TopicPoliciesDisableTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,32 +21,26 @@
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";

private final String testNamespace = "my-namespace";

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
Expand All @@ -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
Expand All @@ -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);
Expand Down
Loading