From e87a4510d173b14531e52b89b06b2d3de4f5a93f Mon Sep 17 00:00:00 2001 From: Ruguo Yu Date: Mon, 29 Aug 2022 22:15:06 +0800 Subject: [PATCH 1/3] [broker][admin] Add cmd to remove topic properties --- .../admin/impl/PersistentTopicsBase.java | 51 +++++++++++++++++++ .../broker/admin/v2/PersistentTopics.java | 35 +++++++++++++ .../pulsar/broker/admin/AdminApi2Test.java | 7 +++ .../apache/pulsar/client/admin/Topics.java | 18 +++++++ .../client/admin/internal/TopicsImpl.java | 13 +++++ .../pulsar/admin/cli/PulsarAdminToolTest.java | 4 ++ .../apache/pulsar/admin/cli/CmdTopics.java | 16 ++++++ 7 files changed, 144 insertions(+) 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 72249fdc1e089..7d895f81ef594 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 @@ -658,6 +658,57 @@ public void updatePropertiesFailed(ManagedLedgerException exception, Object ctx) return future; } + protected CompletableFuture internalRemovePropertiesAsync(boolean authoritative, String key) { + return validateTopicOwnershipAsync(topicName, authoritative) + .thenCompose(__ -> validateNamespaceOperationAsync(topicName.getNamespaceObject(), + NamespaceOperation.CREATE_TOPIC)) + .thenCompose(__ -> { + if (topicName.isPartitioned()) { + return internalRemoveNonPartitionedTopicProperties(key); + } else { + return pulsar().getBrokerService().fetchPartitionedTopicMetadataAsync(topicName) + .thenCompose(metadata -> { + if (metadata.partitions == 0) { + return internalRemoveNonPartitionedTopicProperties(key); + } + return namespaceResources() + .getPartitionedTopicResources().updatePartitionedTopicAsync(topicName, + p -> { + p.properties.remove(key); + return new PartitionedTopicMetadata(p.partitions, p.properties); + }); + }); + } + }).thenAccept(__ -> + log.info("[{}] remove [{}] properties success with key {}", + clientAppId(), topicName, key)); + } + + private CompletableFuture internalRemoveNonPartitionedTopicProperties(String key) { + CompletableFuture future = new CompletableFuture<>(); + pulsar().getBrokerService().getTopicIfExists(topicName.toString()) + .thenAccept(opt -> { + if (!opt.isPresent()) { + throw new RestException(Status.NOT_FOUND, + getTopicNotFoundErrorMessage(topicName.toString())); + } + ManagedLedger managedLedger = ((PersistentTopic) opt.get()).getManagedLedger(); + managedLedger.asyncDeleteProperty(key, new AsyncCallbacks.UpdatePropertiesCallback() { + + @Override + public void updatePropertiesComplete(Map properties, Object ctx) { + future.complete(null); + } + + @Override + public void updatePropertiesFailed(ManagedLedgerException exception, Object ctx) { + future.completeExceptionally(exception); + } + }, null); + }); + return future; + } + protected CompletableFuture internalCheckTopicExists(TopicName topicName) { return pulsar().getNamespaceService().checkTopicExists(topicName) .thenAccept(exist -> { 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 f9aff2baf6f64..f283bb7aa5c28 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 @@ -1003,6 +1003,41 @@ public void updateProperties( }); } + @DELETE + @Path("/{tenant}/{namespace}/{topic}/properties") + @ApiOperation(value = "Remove the key in properties on the given topic.") + @ApiResponses(value = { + @ApiResponse(code = 307, message = "Current broker doesn't serve the namespace of this topic"), + @ApiResponse(code = 401, message = "Don't have permission to administrate resources on this tenant"), + @ApiResponse(code = 403, message = "Don't have admin permission"), + @ApiResponse(code = 404, message = "Partitioned topic does not exist"), + @ApiResponse(code = 409, message = "Concurrent modification"), + @ApiResponse(code = 412, message = "Partitioned topic name is invalid"), + @ApiResponse(code = 500, message = "Internal server error") + }) + public void removeProperties( + @Suspended final AsyncResponse asyncResponse, + @ApiParam(value = "Specify the tenant", required = true) + @PathParam("tenant") String tenant, + @ApiParam(value = "Specify the namespace", required = true) + @PathParam("namespace") String namespace, + @ApiParam(value = "Specify topic name", required = true) + @PathParam("topic") @Encoded String encodedTopic, + @QueryParam("key") String key, + @QueryParam("authoritative") @DefaultValue("false") boolean authoritative) { + validatePersistentTopicName(tenant, namespace, encodedTopic); + internalRemovePropertiesAsync(authoritative, key) + .thenAccept(__ -> asyncResponse.resume(Response.noContent().build())) + .exceptionally(ex -> { + if (!isRedirectException(ex)) { + log.error("[{}] Failed to remove key {} in properties on topic {}", + clientAppId(), key, topicName, ex); + } + resumeAsyncResponseExceptionally(asyncResponse, ex); + return null; + }); + } + @DELETE @Path("/{tenant}/{namespace}/{topic}/partitions") @ApiOperation(value = "Delete a partitioned topic.", diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApi2Test.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApi2Test.java index d492ad73d7860..e55222c9dd4bb 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApi2Test.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApi2Test.java @@ -924,6 +924,13 @@ public void testUpdatePartitionedTopicProperties() throws Exception { Assert.assertEquals(properties.size(), 2); Assert.assertEquals(properties.get("key1"), "value11"); Assert.assertEquals(properties.get("key2"), "value2"); + + // delete key in properties + admin.topics().removeProperties(topicName, "key1"); + properties = admin.topics().getProperties(topicName); + Assert.assertNotNull(properties); + Assert.assertEquals(properties.size(), 1); + Assert.assertEquals(properties.get("key2"), "value2"); } @Test diff --git a/pulsar-client-admin-api/src/main/java/org/apache/pulsar/client/admin/Topics.java b/pulsar-client-admin-api/src/main/java/org/apache/pulsar/client/admin/Topics.java index 98a071c4680c1..177cae9a9a438 100644 --- a/pulsar-client-admin-api/src/main/java/org/apache/pulsar/client/admin/Topics.java +++ b/pulsar-client-admin-api/src/main/java/org/apache/pulsar/client/admin/Topics.java @@ -754,6 +754,24 @@ void updatePartitionedTopic(String topic, int numPartitions, boolean updateLocal */ CompletableFuture updatePropertiesAsync(String topic, Map properties); + /** + * Remove the key in properties on a topic. + * + * @param topic + * @param key + * @throws PulsarAdminException + */ + void removeProperties(String topic, String key) throws PulsarAdminException; + + /** + * Remove the key in properties on a topic asynchronously. + * + * @param topic + * @param key + * @return + */ + CompletableFuture removePropertiesAsync(String topic, String key); + /** * Delete a partitioned topic and its schemas. *

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 45ca53cbd0abb..87be258dbec08 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 @@ -476,6 +476,19 @@ public void deletePartitionedTopic(String topic) throws PulsarAdminException { deletePartitionedTopic(topic, false); } + @Override + public void removeProperties(String topic, String key) throws PulsarAdminException { + sync(() -> removePropertiesAsync(topic, key)); + } + + @Override + public CompletableFuture removePropertiesAsync(String topic, String key) { + TopicName tn = validateTopic(topic); + WebTarget path = topicPath(tn, "properties") + .queryParam("key", key); + return asyncDeleteRequest(path); + } + @Override public CompletableFuture deletePartitionedTopicAsync(String topic) { return deletePartitionedTopicAsync(topic, false); diff --git a/pulsar-client-tools-test/src/test/java/org/apache/pulsar/admin/cli/PulsarAdminToolTest.java b/pulsar-client-tools-test/src/test/java/org/apache/pulsar/admin/cli/PulsarAdminToolTest.java index 7ca09a2b14d17..d1d858c0e26e8 100644 --- a/pulsar-client-tools-test/src/test/java/org/apache/pulsar/admin/cli/PulsarAdminToolTest.java +++ b/pulsar-client-tools-test/src/test/java/org/apache/pulsar/admin/cli/PulsarAdminToolTest.java @@ -1508,6 +1508,10 @@ public void topics() throws Exception { props.put("x", "y,z"); verify(mockTopics).updateProperties("persistent://myprop/clust/ns1/ds1", props); + cmdTopics = new CmdTopics(() -> admin); + cmdTopics.run(split("remove-properties persistent://myprop/clust/ns1/ds1 --key a")); + verify(mockTopics).removeProperties("persistent://myprop/clust/ns1/ds1", "a"); + cmdTopics = new CmdTopics(() -> admin); cmdTopics.run(split("get-subscription-properties persistent://myprop/clust/ns1/ds1 -s sub1")); verify(mockTopics).getSubscriptionProperties("persistent://myprop/clust/ns1/ds1", "sub1"); 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 dc6313b1be1d9..8b12d70d057e5 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 @@ -123,6 +123,7 @@ public CmdTopics(Supplier admin) { jcommander.addCommand("get-partitioned-topic-metadata", new GetPartitionedTopicMetadataCmd()); jcommander.addCommand("get-properties", new GetPropertiesCmd()); jcommander.addCommand("update-properties", new UpdateProperties()); + jcommander.addCommand("remove-properties", new RemoveProperties()); jcommander.addCommand("delete-partitioned-topic", new DeletePartitionedCmd()); jcommander.addCommand("peek-messages", new PeekMessages()); @@ -649,6 +650,21 @@ void run() throws Exception { } } + @Parameters(commandDescription = "Remove the key in properties of a topic") + private class RemoveProperties extends CliCommand { + @Parameter(description = "persistent://tenant/namespace/topic", required = true) + private java.util.List params; + + @Parameter(names = {"--key", "-k"}, description = "The key to remove in the properties of topic") + private String key; + + @Override + void run() throws Exception { + String topic = validateTopicName(params); + getTopics().removeProperties(topic, key); + } + } + @Parameters(commandDescription = "Delete a partitioned topic. " + "It will also delete all the partitions of the topic if it exists." + "And the application is not able to connect to the topic(delete then re-create with same name) again " From fcba52327a6709d5c2926c574d5e1ec20fa17ff6 Mon Sep 17 00:00:00 2001 From: Ruguo Yu Date: Wed, 31 Aug 2022 15:48:41 +0800 Subject: [PATCH 2/3] address comment --- .../PulsarAuthorizationProvider.java | 1 + .../admin/impl/PersistentTopicsBase.java | 7 ++++--- .../pulsar/broker/admin/AdminApi2Test.java | 21 ++++++++++++++----- .../common/policies/data/TopicOperation.java | 1 + 4 files changed, 22 insertions(+), 8 deletions(-) diff --git a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/authorization/PulsarAuthorizationProvider.java b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/authorization/PulsarAuthorizationProvider.java index 399ecc4a9f0a5..de586345fd0da 100644 --- a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/authorization/PulsarAuthorizationProvider.java +++ b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/authorization/PulsarAuthorizationProvider.java @@ -553,6 +553,7 @@ public CompletableFuture allowTopicOperationAsync(TopicName topicName, case COMPACT: case OFFLOAD: case UNLOAD: + case DELETE_METADATA: case ADD_BUNDLE_RANGE: case GET_BUNDLE_RANGE: case DELETE_BUNDLE_RANGE: 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 7d895f81ef594..9d13c43ed3dbe 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 @@ -660,8 +660,7 @@ public void updatePropertiesFailed(ManagedLedgerException exception, Object ctx) protected CompletableFuture internalRemovePropertiesAsync(boolean authoritative, String key) { return validateTopicOwnershipAsync(topicName, authoritative) - .thenCompose(__ -> validateNamespaceOperationAsync(topicName.getNamespaceObject(), - NamespaceOperation.CREATE_TOPIC)) + .thenCompose(__ -> validateTopicOperationAsync(topicName, TopicOperation.DELETE_METADATA)) .thenCompose(__ -> { if (topicName.isPartitioned()) { return internalRemoveNonPartitionedTopicProperties(key); @@ -674,7 +673,9 @@ protected CompletableFuture internalRemovePropertiesAsync(boolean authorit return namespaceResources() .getPartitionedTopicResources().updatePartitionedTopicAsync(topicName, p -> { - p.properties.remove(key); + if (p.properties != null) { + p.properties.remove(key); + } return new PartitionedTopicMetadata(p.partitions, p.properties); }); }); diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApi2Test.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApi2Test.java index e55222c9dd4bb..3d58c6f682795 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApi2Test.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApi2Test.java @@ -892,6 +892,7 @@ public void testCreateAndGetTopicProperties() throws Exception { public void testUpdatePartitionedTopicProperties() throws Exception { final String namespace = "prop-xyz/ns2"; final String topicName = "persistent://" + namespace + "/testUpdatePartitionedTopicProperties"; + final String topicNameTwo = "persistent://" + namespace + "/testUpdatePartitionedTopicProperties2"; admin.namespaces().createNamespace(namespace, 20); // create partitioned topic without properties @@ -925,12 +926,22 @@ public void testUpdatePartitionedTopicProperties() throws Exception { Assert.assertEquals(properties.get("key1"), "value11"); Assert.assertEquals(properties.get("key2"), "value2"); - // delete key in properties - admin.topics().removeProperties(topicName, "key1"); - properties = admin.topics().getProperties(topicName); - Assert.assertNotNull(properties); + // create topic without properties + admin.topics().createPartitionedTopic(topicNameTwo, 2); + properties = admin.topics().getProperties(topicNameTwo); + Assert.assertNull(properties); + // remove key of properties on this topic + admin.topics().removeProperties(topicNameTwo, "key1"); + Assert.assertNull(properties); + properties = new HashMap<>(); + properties.put("key1", "value1"); + properties.put("key2", "value2"); + admin.topics().updateProperties(topicNameTwo, properties); + properties = admin.topics().getProperties(topicNameTwo); + Assert.assertEquals(properties.size(), 2); + admin.topics().removeProperties(topicNameTwo, "key1"); + properties = admin.topics().getProperties(topicNameTwo); Assert.assertEquals(properties.size(), 1); - Assert.assertEquals(properties.get("key2"), "value2"); } @Test diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/TopicOperation.java b/pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/TopicOperation.java index d4de706e607b7..0184e0efb82d7 100644 --- a/pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/TopicOperation.java +++ b/pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/TopicOperation.java @@ -50,6 +50,7 @@ public enum TopicOperation { GET_STATS, GET_METADATA, + DELETE_METADATA, GET_BACKLOG_SIZE, SET_REPLICATED_SUBSCRIPTION_STATUS, From 0cc4429b751cfb600b5fe3f55e6e90924d70ba20 Mon Sep 17 00:00:00 2001 From: Ruguo Yu Date: Wed, 31 Aug 2022 23:08:52 +0800 Subject: [PATCH 3/3] address comment --- .../apache/pulsar/broker/admin/AdminApi2Test.java | 14 ++++++++------ 1 file changed, 8 insertions(+), 6 deletions(-) diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApi2Test.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApi2Test.java index 3d58c6f682795..2c2d7057cfad3 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApi2Test.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApi2Test.java @@ -932,16 +932,18 @@ public void testUpdatePartitionedTopicProperties() throws Exception { Assert.assertNull(properties); // remove key of properties on this topic admin.topics().removeProperties(topicNameTwo, "key1"); + properties = admin.topics().getProperties(topicNameTwo); Assert.assertNull(properties); - properties = new HashMap<>(); - properties.put("key1", "value1"); - properties.put("key2", "value2"); - admin.topics().updateProperties(topicNameTwo, properties); + Map topicProp = new HashMap<>(); + topicProp.put("key1", "value1"); + topicProp.put("key2", "value2"); + admin.topics().updateProperties(topicNameTwo, topicProp); properties = admin.topics().getProperties(topicNameTwo); - Assert.assertEquals(properties.size(), 2); + Assert.assertEquals(properties, topicProp); admin.topics().removeProperties(topicNameTwo, "key1"); + topicProp.remove("key1"); properties = admin.topics().getProperties(topicNameTwo); - Assert.assertEquals(properties.size(), 1); + Assert.assertEquals(properties, topicProp); } @Test