This is an automated email from the ASF dual-hosted git repository.
mmerli pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pulsar.git
The following commit(s) were added to refs/heads/master by this push:
new cbe8900 [pulsar-broker] support remove-ttl api for V1 namespace
(#12121)
cbe8900 is described below
commit cbe8900ae3ed857a14412b4e23c273d97adb7747
Author: Rajan Dhabalia <rdhaba...@apache.org>
AuthorDate: Wed Sep 22 15:44:36 2021 -0700
[pulsar-broker] support remove-ttl api for V1 namespace (#12121)
---
.../apache/pulsar/broker/admin/v1/Namespaces.java | 14 +++++++++-
.../pulsar/broker/admin/TopicMessageTTLTest.java | 30 +++++++++++++++++-----
2 files changed, 37 insertions(+), 7 deletions(-)
diff --git
a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v1/Namespaces.java
b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v1/Namespaces.java
index 03849ec..dac7e44 100644
---
a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v1/Namespaces.java
+++
b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v1/Namespaces.java
@@ -331,7 +331,7 @@ public class Namespaces extends NamespacesBase {
@ApiOperation(hidden = true, value = "Get the message TTL for the
namespace")
@ApiResponses(value = { @ApiResponse(code = 403, message = "Don't have
admin permission"),
@ApiResponse(code = 404, message = "Property or cluster or
namespace doesn't exist") })
- public int getNamespaceMessageTTL(@PathParam("property") String property,
@PathParam("cluster") String cluster,
+ public Integer getNamespaceMessageTTL(@PathParam("property") String
property, @PathParam("cluster") String cluster,
@PathParam("namespace") String namespace) {
validateNamespaceName(property, cluster, namespace);
validateNamespacePolicyOperation(NamespaceName.get(property,
namespace), PolicyName.TTL, PolicyOperation.READ);
@@ -352,6 +352,18 @@ public class Namespaces extends NamespacesBase {
internalSetNamespaceMessageTTL(messageTTL);
}
+ @DELETE
+ @Path("/{property}/{cluster}/{namespace}/messageTTL")
+ @ApiOperation(value = "Set message TTL in seconds for namespace")
+ @ApiResponses(value = { @ApiResponse(code = 403, message = "Don't have
admin permission"),
+ @ApiResponse(code = 404, message = "Tenant or cluster or namespace
doesn't exist"),
+ @ApiResponse(code = 412, message = "Invalid TTL") })
+ public void removeNamespaceMessageTTL(@PathParam("property") String
property, @PathParam("cluster") String cluster,
+ @PathParam("namespace") String namespace) {
+ validateNamespaceName(property, cluster, namespace);
+ internalSetNamespaceMessageTTL(null);
+ }
+
@GET
@Path("/{property}/{cluster}/{namespace}/subscriptionExpirationTime")
@ApiOperation(hidden = true, value = "Get the subscription expiration time
for the namespace")
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 4a2c728..746b1a1 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
@@ -33,6 +33,7 @@ import org.eclipse.jetty.http.HttpStatus;
import org.testng.Assert;
import org.testng.annotations.AfterMethod;
import org.testng.annotations.BeforeMethod;
+import org.testng.annotations.DataProvider;
import org.testng.annotations.Test;
import java.lang.reflect.Method;
@@ -43,11 +44,9 @@ import java.util.UUID;
public class TopicMessageTTLTest extends MockedPulsarServiceBaseTest {
private final String testTenant = "my-tenant";
-
+ private final String testCluster = "test";
private final String testNamespace = "my-namespace";
-
private final String myNamespace = testTenant + "/" + testNamespace;
-
private final String testTopic = "persistent://" + myNamespace +
"/test-topic-message-ttl";
@BeforeMethod
@@ -59,10 +58,10 @@ public class TopicMessageTTLTest extends
MockedPulsarServiceBaseTest {
this.conf.setTtlDurationDefaultInSeconds(3600);
super.internalSetup();
- admin.clusters().createCluster("test",
ClusterData.builder().serviceUrl(pulsar.getWebServiceAddress()).build());
- TenantInfoImpl tenantInfo = new
TenantInfoImpl(Sets.newHashSet("role1", "role2"), Sets.newHashSet("test"));
+ admin.clusters().createCluster(testCluster,
ClusterData.builder().serviceUrl(pulsar.getWebServiceAddress()).build());
+ TenantInfoImpl tenantInfo = new
TenantInfoImpl(Sets.newHashSet("role1", "role2"), Sets.newHashSet(testCluster));
admin.tenants().createTenant(this.testTenant, tenantInfo);
- admin.namespaces().createNamespace(testTenant + "/" + testNamespace,
Sets.newHashSet("test"));
+ admin.namespaces().createNamespace(testTenant + "/" + testNamespace,
Sets.newHashSet(testCluster));
admin.topics().createPartitionedTopic(testTopic, 2);
Producer producer = pulsarClient.newProducer().topic(testTenant + "/"
+ testNamespace + "/" + "dummy-topic").create();
producer.close();
@@ -75,6 +74,11 @@ public class TopicMessageTTLTest extends
MockedPulsarServiceBaseTest {
super.internalCleanup();
}
+ @DataProvider(name = "isV1")
+ public Object[][] isV1() {
+ return new Object[][] { { true }, { false } };
+ }
+
@Test
public void testSetThenRemoveMessageTTL() throws Exception {
admin.topics().setMessageTTL(testTopic, 100);
@@ -181,6 +185,20 @@ public class TopicMessageTTLTest extends
MockedPulsarServiceBaseTest {
Assert.assertEquals((int) ((CompletableFuture<Integer>)
method.invoke(persistentTopic)).join(), 3600);
}
+ @Test(dataProvider = "isV1")
+ public void testNamespaceTTL(boolean isV1) throws Exception {
+ String myNamespace = testTenant + "/" + (isV1 ? testCluster + "/" :
"") + "n1"+isV1;
+ admin.namespaces().createNamespace(myNamespace,
Sets.newHashSet(testCluster));
+
+ admin.namespaces().setNamespaceMessageTTL(myNamespace, 10);
+ Awaitility.await().untilAsserted(()
+ ->
Assert.assertEquals(admin.namespaces().getNamespaceMessageTTL(myNamespace).intValue(),
10));
+
+ admin.namespaces().removeNamespaceMessageTTL(myNamespace);
+ Awaitility.await().untilAsserted(()
+ ->
Assert.assertNull(admin.namespaces().getNamespaceMessageTTL(myNamespace)));
+ }
+
@Test(timeOut = 20000)
public void testDifferentLevelPolicyApplied() throws Exception {
final String topicName = testTopic + UUID.randomUUID();
