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 d4055b5 [Broker] Add verification when terminating non-persistent topic (#11903) d4055b5 is described below commit d4055b5877c7935b999314eafc85b80b5e219c23 Author: 包子 <wudixiaolong...@icloud.com> AuthorDate: Tue Sep 7 00:20:19 2021 +0800 [Broker] Add verification when terminating non-persistent topic (#11903) * [Broker] Add verification when terminating non-persistent * [Broker] add unit test * code format --- .../apache/pulsar/broker/admin/AdminResource.java | 15 ++++++++++ .../pulsar/broker/admin/v1/PersistentTopics.java | 3 +- .../pulsar/broker/admin/v2/PersistentTopics.java | 3 +- .../pulsar/broker/admin/AdminResourceTest.java | 35 ++++++++++++++++++++++ .../pulsar/broker/admin/PersistentTopicsTest.java | 29 ++++++++++++++++++ 5 files changed, 83 insertions(+), 2 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 de13b54..cf96e92 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 @@ -241,6 +241,13 @@ public abstract class AdminResource extends PulsarWebResource { } } + protected void validatePersistentTopicName(String property, String namespace, String encodedTopic) { + validateTopicName(property, namespace, encodedTopic); + if (topicName.getDomain() != TopicDomain.persistent) { + throw new RestException(Status.NOT_ACCEPTABLE, "Need to provide a persistent topic name"); + } + } + protected void validatePartitionedTopicName(String tenant, String namespace, String encodedTopic) { // first, it has to be a validate topic name validateTopicName(tenant, namespace, encodedTopic); @@ -278,6 +285,14 @@ public abstract class AdminResource extends PulsarWebResource { } } + @Deprecated + protected void validatePersistentTopicName(String property, String cluster, String namespace, String encodedTopic) { + validateTopicName(property, cluster, namespace, encodedTopic); + if (topicName.getDomain() != TopicDomain.persistent) { + throw new RestException(Status.NOT_ACCEPTABLE, "Need to provide a persistent topic name"); + } + } + protected Policies getNamespacePolicies(NamespaceName namespaceName) { try { final String namespace = namespaceName.toString(); diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v1/PersistentTopics.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v1/PersistentTopics.java index fee8707..32f5ab3 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v1/PersistentTopics.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v1/PersistentTopics.java @@ -673,11 +673,12 @@ public class PersistentTopics extends PersistentTopicsBase { @ApiResponse(code = 307, message = "Current broker doesn't serve the namespace of this topic"), @ApiResponse(code = 403, message = "Don't have admin permission"), @ApiResponse(code = 405, message = "Operation not allowed on non-persistent topic"), + @ApiResponse(code = 406, message = "Need to provide a persistent topic name"), @ApiResponse(code = 404, message = "Topic does not exist") }) public MessageId terminate(@PathParam("property") String property, @PathParam("cluster") String cluster, @PathParam("namespace") String namespace, @PathParam("topic") @Encoded String encodedTopic, @QueryParam("authoritative") @DefaultValue("false") boolean authoritative) { - validateTopicName(property, cluster, namespace, encodedTopic); + validatePersistentTopicName(property, cluster, namespace, encodedTopic); return internalTerminate(authoritative); } diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v2/PersistentTopics.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v2/PersistentTopics.java index 4930915..1e26907 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 @@ -2442,6 +2442,7 @@ public class PersistentTopics extends PersistentTopicsBase { @ApiResponse(code = 403, message = "Don't have admin permission"), @ApiResponse(code = 404, message = "Topic does not exist"), @ApiResponse(code = 405, message = "Termination of a partitioned topic is not allowed"), + @ApiResponse(code = 406, message = "Need to provide a persistent topic name"), @ApiResponse(code = 412, message = "Topic name is not valid"), @ApiResponse(code = 500, message = "Internal server error"), @ApiResponse(code = 503, message = "Failed to validate global cluster configuration")}) @@ -2454,7 +2455,7 @@ public class PersistentTopics extends PersistentTopicsBase { @PathParam("topic") @Encoded String encodedTopic, @ApiParam(value = "Is authentication required to perform this operation") @QueryParam("authoritative") @DefaultValue("false") boolean authoritative) { - validateTopicName(tenant, namespace, encodedTopic); + validatePersistentTopicName(tenant, namespace, encodedTopic); return internalTerminate(authoritative); } diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminResourceTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminResourceTest.java index 982c0b2..e047d31 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminResourceTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminResourceTest.java @@ -57,6 +57,41 @@ public class AdminResourceTest extends BrokerTestBase { }; } + private static AdminResource mockNonPersistentResource() { + return new AdminResource() { + + @Override + protected String domain() { + return "non-persistent"; + } + }; + } + + @Test + public void testValidatePersistentTopicNameSuccess() { + String tenant = "test-tenant"; + String namespace = "test-namespace"; + String topic = Codec.encode("test-topic"); + + AdminResource resource = mockResource(); + resource.validatePersistentTopicName(tenant, namespace, topic); + } + + @Test + public void testValidatePersistentTopicNameInvalid() { + String tenant = "test-tenant"; + String namespace = "test-namespace"; + String topic = Codec.encode("test-topic"); + + AdminResource nPResource = mockNonPersistentResource(); + try { + nPResource.validatePersistentTopicName(tenant, namespace, topic); + fail("Should fail validation on non-persistent topic"); + } catch (RestException e) { + assertEquals(Status.NOT_ACCEPTABLE.getStatusCode(), e.getResponse().getStatus()); + } + } + @Test public void testValidatePartitionedTopicNameSuccess() { String tenant = "test-tenant"; diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/PersistentTopicsTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/PersistentTopicsTest.java index 182d2cc..9871b7d 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/PersistentTopicsTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/PersistentTopicsTest.java @@ -333,6 +333,35 @@ public class PersistentTopicsTest extends MockedPulsarServiceBaseTest { } @Test + public void testTerminate() { + String testLocalTopicName = "topic-not-found"; + + // 1) Create the nonPartitionTopic topic + persistentTopics.createNonPartitionedTopic(testTenant, testNamespace, testLocalTopicName, true); + + // 2) Create a subscription + AsyncResponse response = mock(AsyncResponse.class); + persistentTopics.createSubscription(response, testTenant, testNamespace, testLocalTopicName, "test", true, + (MessageIdImpl) MessageId.earliest, false); + ArgumentCaptor<Response> responseCaptor = ArgumentCaptor.forClass(Response.class); + verify(response, timeout(5000).times(1)).resume(responseCaptor.capture()); + Assert.assertEquals(responseCaptor.getValue().getStatus(), Response.Status.NO_CONTENT.getStatusCode()); + + // 3) Assert terminate persistent topic + MessageId messageId = persistentTopics.terminate(testTenant, testNamespace, testLocalTopicName, true); + Assert.assertEquals(messageId, new MessageIdImpl(3, -1, -1)); + + // 4) Assert terminate non-persistent topic + String nonPersistentTopicName = "non-persistent-topic"; + try { + nonPersistentTopic.terminate(testTenant, testNamespace, nonPersistentTopicName, true); + Assert.fail("Should fail validation on non-persistent topic"); + } catch (RestException e) { + Assert.assertEquals(Response.Status.NOT_ACCEPTABLE.getStatusCode(), e.getResponse().getStatus()); + } + } + + @Test public void testNonPartitionedTopics() { final String nonPartitionTopic = "non-partitioned-topic"; AsyncResponse response = mock(AsyncResponse.class);