You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by GitBox <gi...@apache.org> on 2022/08/23 12:51:48 UTC
[GitHub] [pulsar] AnonHxy opened a new pull request, #17238: [broker][admin]Add api for update topic properties
AnonHxy opened a new pull request, #17238:
URL: https://github.com/apache/pulsar/pull/17238
### Motivation
* Maybe this modification should be part of https://github.com/apache/pulsar/issues/12629 . As described in the PIP document "These metadata could be set during topic creation and also updated.", we have implemented the `creation` method, but have not implement the `updated` method.
* This PR will implement the topic meta update
### Modifications
* Add a admin api `updateProperties`
### Verifying this change
- [x] Make sure that the change passes the CI checks.
- Add UT `org.apache.pulsar.broker.admin.AdminApi2Test#testUpdatePartitionedTopicProperties`
- Add UT `org.apache.pulsar.broker.admin.AdminApi2Test#testUpdateNonPartitionedTopicProperties`
### Documentation
Check the box below or label this PR directly.
Need to update docs?
- [ ] `doc-required`
(Your PR needs to update docs and you will update later)
- [x] `doc-not-needed`
(Please explain why)
- [ ] `doc`
(Your PR contains doc changes)
- [ ] `doc-complete`
(Docs have been already added)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] AnonHxy merged pull request #17238: [broker][admin]Add api for update topic properties
Posted by GitBox <gi...@apache.org>.
AnonHxy merged PR #17238:
URL: https://github.com/apache/pulsar/pull/17238
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] heesung-sn commented on a diff in pull request #17238: [broker][admin]Add api for update topic properties
Posted by "heesung-sn (via GitHub)" <gi...@apache.org>.
heesung-sn commented on code in PR #17238:
URL: https://github.com/apache/pulsar/pull/17238#discussion_r1199270887
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java:
##########
@@ -601,6 +602,60 @@ private CompletableFuture<Map<String, String>> getPropertiesAsync() {
});
}
+ protected CompletableFuture<Void> internalUpdatePropertiesAsync(boolean authoritative,
+ Map<String, String> properties) {
+ if (properties == null || properties.isEmpty()) {
+ log.warn("[{}] [{}] properties is empty, ignore update", clientAppId(), topicName);
+ return CompletableFuture.completedFuture(null);
+ }
+ return validateTopicOwnershipAsync(topicName, authoritative)
+ .thenCompose(__ -> validateTopicOperationAsync(topicName, TopicOperation.PRODUCE))
+ .thenCompose(__ -> {
+ if (topicName.isPartitioned()) {
+ return internalUpdateNonPartitionedTopicProperties(properties);
+ } else {
+ return pulsar().getBrokerService().fetchPartitionedTopicMetadataAsync(topicName)
+ .thenCompose(metadata -> {
+ if (metadata.partitions == 0) {
+ return internalUpdateNonPartitionedTopicProperties(properties);
+ }
+ return namespaceResources()
+ .getPartitionedTopicResources().updatePartitionedTopicAsync(topicName,
+ p -> new PartitionedTopicMetadata(p.partitions,
+ MapUtils.putAll(p.properties, properties.entrySet().toArray())));
+ });
+ }
+ }).thenAccept(__ ->
+ log.info("[{}] [{}] update properties success with properties {}",
+ clientAppId(), topicName, properties));
+ }
+
+ private CompletableFuture<Void> internalUpdateNonPartitionedTopicProperties(Map<String, String> properties) {
+ CompletableFuture<Void> 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.asyncSetProperties(properties, new AsyncCallbacks.UpdatePropertiesCallback() {
+
+ @Override
+ public void updatePropertiesComplete(Map<String, String> properties, Object ctx) {
+ managedLedger.getConfig().getProperties().putAll(properties);
Review Comment:
`managedLedger.getConfig().getProperties()` can return NPE if the topic is from the older code without the `properties`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] AnonHxy commented on a diff in pull request #17238: [broker][admin]Add api for update topic properties
Posted by GitBox <gi...@apache.org>.
AnonHxy commented on code in PR #17238:
URL: https://github.com/apache/pulsar/pull/17238#discussion_r953292080
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v2/PersistentTopics.java:
##########
@@ -966,6 +966,42 @@ public void getProperties(
});
}
+ @PUT
+ @Path("/{tenant}/{namespace}/{topic}/properties")
+ @ApiOperation(value = "Replace all the properties on the given subscription")
Review Comment:
Nice catch! Updated
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] Technoboy- commented on a diff in pull request #17238: [broker][admin]Add api for update topic properties
Posted by GitBox <gi...@apache.org>.
Technoboy- commented on code in PR #17238:
URL: https://github.com/apache/pulsar/pull/17238#discussion_r953276852
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v2/PersistentTopics.java:
##########
@@ -966,6 +966,42 @@ public void getProperties(
});
}
+ @PUT
+ @Path("/{tenant}/{namespace}/{topic}/properties")
+ @ApiOperation(value = "Replace all the properties on the given subscription")
Review Comment:
`Replace all the properties` -> `Update the properties`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] Jason918 commented on a diff in pull request #17238: [broker][admin]Add api for update topic properties
Posted by GitBox <gi...@apache.org>.
Jason918 commented on code in PR #17238:
URL: https://github.com/apache/pulsar/pull/17238#discussion_r954434866
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v2/PersistentTopics.java:
##########
@@ -966,6 +966,42 @@ public void getProperties(
});
}
+ @PUT
+ @Path("/{tenant}/{namespace}/{topic}/properties")
+ @ApiOperation(value = "Update the 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 or"
+ + "subscriber is not authorized to access this operation"),
+ @ApiResponse(code = 403, message = "Don't have admin permission"),
+ @ApiResponse(code = 404, message = "Topic/Subscription does not exist"),
+ @ApiResponse(code = 405, message = "Method Not Allowed"),
+ @ApiResponse(code = 500, message = "Internal server error"),
+ @ApiResponse(code = 503, message = "Failed to validate global cluster configuration")
+ })
+ public void updateProperties(
+ @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,
+ @ApiParam(value = "Whether leader broker redirected this call to this broker. For internal use.")
+ @QueryParam("authoritative") @DefaultValue("false") boolean authoritative,
+ @ApiParam(value = "Key value pair properties for the topic metadata") Map<String, String> properties){
+ validatePersistentTopicName(tenant, namespace, encodedTopic);
Review Comment:
Please verify this if you have time. If there is a problem, we can fix them in another PR
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] AnonHxy commented on a diff in pull request #17238: [broker][admin]Add api for update topic properties
Posted by GitBox <gi...@apache.org>.
AnonHxy commented on code in PR #17238:
URL: https://github.com/apache/pulsar/pull/17238#discussion_r953642381
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java:
##########
@@ -601,6 +602,60 @@ private CompletableFuture<Map<String, String>> getPropertiesAsync() {
});
}
+ protected CompletableFuture<Void> internalUpdatePropertiesAsync(boolean authoritative,
+ Map<String, String> properties) {
+ if (properties == null || properties.isEmpty()) {
+ log.warn("[{}] [{}] properties is empty, ignore update", clientAppId(), topicName);
+ return CompletableFuture.completedFuture(null);
Review Comment:
Users can set the value as null to make the key invalidte. Here we don't delete the key because I think it's better to keep same action as what non-partitioned topic do, see L3915 and L3922, the `delete` flag(the 2nd params) is false:
https://github.com/apache/pulsar/blob/05a2ea87371783000f496ce8103c0ac372e4a8fe/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java#L3912-L3923
And if we have to delete the key, I think it's better add another API like `deleteProperty` :) @Jason918
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] AnonHxy commented on a diff in pull request #17238: [broker][admin]Add api for update topic properties
Posted by "AnonHxy (via GitHub)" <gi...@apache.org>.
AnonHxy commented on code in PR #17238:
URL: https://github.com/apache/pulsar/pull/17238#discussion_r1201442171
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java:
##########
@@ -601,6 +602,60 @@ private CompletableFuture<Map<String, String>> getPropertiesAsync() {
});
}
+ protected CompletableFuture<Void> internalUpdatePropertiesAsync(boolean authoritative,
+ Map<String, String> properties) {
+ if (properties == null || properties.isEmpty()) {
+ log.warn("[{}] [{}] properties is empty, ignore update", clientAppId(), topicName);
+ return CompletableFuture.completedFuture(null);
+ }
+ return validateTopicOwnershipAsync(topicName, authoritative)
+ .thenCompose(__ -> validateTopicOperationAsync(topicName, TopicOperation.PRODUCE))
+ .thenCompose(__ -> {
+ if (topicName.isPartitioned()) {
+ return internalUpdateNonPartitionedTopicProperties(properties);
+ } else {
+ return pulsar().getBrokerService().fetchPartitionedTopicMetadataAsync(topicName)
+ .thenCompose(metadata -> {
+ if (metadata.partitions == 0) {
+ return internalUpdateNonPartitionedTopicProperties(properties);
+ }
+ return namespaceResources()
+ .getPartitionedTopicResources().updatePartitionedTopicAsync(topicName,
+ p -> new PartitionedTopicMetadata(p.partitions,
+ MapUtils.putAll(p.properties, properties.entrySet().toArray())));
+ });
+ }
+ }).thenAccept(__ ->
+ log.info("[{}] [{}] update properties success with properties {}",
+ clientAppId(), topicName, properties));
+ }
+
+ private CompletableFuture<Void> internalUpdateNonPartitionedTopicProperties(Map<String, String> properties) {
+ CompletableFuture<Void> 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.asyncSetProperties(properties, new AsyncCallbacks.UpdatePropertiesCallback() {
+
+ @Override
+ public void updatePropertiesComplete(Map<String, String> properties, Object ctx) {
+ managedLedger.getConfig().getProperties().putAll(properties);
Review Comment:
Thanks for you to fix it!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] Jason918 commented on a diff in pull request #17238: [broker][admin]Add api for update topic properties
Posted by GitBox <gi...@apache.org>.
Jason918 commented on code in PR #17238:
URL: https://github.com/apache/pulsar/pull/17238#discussion_r953461822
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java:
##########
@@ -18,6 +18,7 @@
*/
package org.apache.pulsar.broker.admin.impl;
+import static org.apache.commons.collections4.MapUtils.putAll;
Review Comment:
Better avoid import static method.
##########
pulsar-client-admin-api/src/main/java/org/apache/pulsar/client/admin/Topics.java:
##########
@@ -736,6 +736,24 @@ void updatePartitionedTopic(String topic, int numPartitions, boolean updateLocal
*/
CompletableFuture<Map<String, String>> getPropertiesAsync(String topic);
+ /**
+ * Update Topic Properties on a topic.
+ * The new properties will override the existing values, properties that are not passed will be keep.
+ * @param topic
Review Comment:
> "properties that are not passed will be keep."
Not quite clear about what this means.
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v2/PersistentTopics.java:
##########
@@ -966,6 +966,42 @@ public void getProperties(
});
}
+ @PUT
+ @Path("/{tenant}/{namespace}/{topic}/properties")
+ @ApiOperation(value = "Update the 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 or"
+ + "subscriber is not authorized to access this operation"),
+ @ApiResponse(code = 403, message = "Don't have admin permission"),
+ @ApiResponse(code = 404, message = "Topic/Subscription does not exist"),
+ @ApiResponse(code = 405, message = "Method Not Allowed"),
+ @ApiResponse(code = 500, message = "Internal server error"),
+ @ApiResponse(code = 503, message = "Failed to validate global cluster configuration")
+ })
+ public void updateProperties(
+ @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,
+ @ApiParam(value = "Whether leader broker redirected this call to this broker. For internal use.")
+ @QueryParam("authoritative") @DefaultValue("false") boolean authoritative,
+ @ApiParam(value = "Key value pair properties for the topic metadata") Map<String, String> properties){
+ validatePersistentTopicName(tenant, namespace, encodedTopic);
Review Comment:
I wonder what happens if this validation fails.
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java:
##########
@@ -601,6 +602,60 @@ private CompletableFuture<Map<String, String>> getPropertiesAsync() {
});
}
+ protected CompletableFuture<Void> internalUpdatePropertiesAsync(boolean authoritative,
+ Map<String, String> properties) {
+ if (properties == null || properties.isEmpty()) {
+ log.warn("[{}] [{}] properties is empty, ignore update", clientAppId(), topicName);
+ return CompletableFuture.completedFuture(null);
Review Comment:
If so, is there a way to delete the properties?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] Jason918 commented on pull request #17238: [broker][admin]Add api for update topic properties
Posted by GitBox <gi...@apache.org>.
Jason918 commented on PR #17238:
URL: https://github.com/apache/pulsar/pull/17238#issuecomment-1236343562
> @Jason918 we cannot add new APIs on released branches
Sure, I will remove the label.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] eolivelli commented on pull request #17238: [broker][admin]Add api for update topic properties
Posted by GitBox <gi...@apache.org>.
eolivelli commented on PR #17238:
URL: https://github.com/apache/pulsar/pull/17238#issuecomment-1236343097
@Jason918 we cannot add new APIs on released branches
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] AnonHxy commented on a diff in pull request #17238: [broker][admin]Add api for update topic properties
Posted by GitBox <gi...@apache.org>.
AnonHxy commented on code in PR #17238:
URL: https://github.com/apache/pulsar/pull/17238#discussion_r953642381
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java:
##########
@@ -601,6 +602,60 @@ private CompletableFuture<Map<String, String>> getPropertiesAsync() {
});
}
+ protected CompletableFuture<Void> internalUpdatePropertiesAsync(boolean authoritative,
+ Map<String, String> properties) {
+ if (properties == null || properties.isEmpty()) {
+ log.warn("[{}] [{}] properties is empty, ignore update", clientAppId(), topicName);
+ return CompletableFuture.completedFuture(null);
Review Comment:
Users can set the value as null to make the key invalidte. Here we don't delete the key because I think it's better to keep same action as what non-partitioned topic do, see below:
https://github.com/apache/pulsar/blob/05a2ea87371783000f496ce8103c0ac372e4a8fe/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java#L3912-L3923
And if we have to delete the key, I think it's better add another API like `deleteProperty` :) @Jason918
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v2/PersistentTopics.java:
##########
@@ -966,6 +966,42 @@ public void getProperties(
});
}
+ @PUT
+ @Path("/{tenant}/{namespace}/{topic}/properties")
+ @ApiOperation(value = "Update the 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 or"
+ + "subscriber is not authorized to access this operation"),
+ @ApiResponse(code = 403, message = "Don't have admin permission"),
+ @ApiResponse(code = 404, message = "Topic/Subscription does not exist"),
+ @ApiResponse(code = 405, message = "Method Not Allowed"),
+ @ApiResponse(code = 500, message = "Internal server error"),
+ @ApiResponse(code = 503, message = "Failed to validate global cluster configuration")
+ })
+ public void updateProperties(
+ @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,
+ @ApiParam(value = "Whether leader broker redirected this call to this broker. For internal use.")
+ @QueryParam("authoritative") @DefaultValue("false") boolean authoritative,
+ @ApiParam(value = "Key value pair properties for the topic metadata") Map<String, String> properties){
+ validatePersistentTopicName(tenant, namespace, encodedTopic);
Review Comment:
I think that the web service framework should catch all exceptions throw by API methods, but I am not sure.
##########
pulsar-client-admin-api/src/main/java/org/apache/pulsar/client/admin/Topics.java:
##########
@@ -736,6 +736,24 @@ void updatePartitionedTopic(String topic, int numPartitions, boolean updateLocal
*/
CompletableFuture<Map<String, String>> getPropertiesAsync(String topic);
+ /**
+ * Update Topic Properties on a topic.
+ * The new properties will override the existing values, properties that are not passed will be keep.
+ * @param topic
Review Comment:
Updated. PTAL @Jason918
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java:
##########
@@ -18,6 +18,7 @@
*/
package org.apache.pulsar.broker.admin.impl;
+import static org.apache.commons.collections4.MapUtils.putAll;
Review Comment:
OK
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] Jason918 commented on pull request #17238: [broker][admin]Add api for update topic properties
Posted by GitBox <gi...@apache.org>.
Jason918 commented on PR #17238:
URL: https://github.com/apache/pulsar/pull/17238#issuecomment-1236342851
Can you help open a new PR to branch-2.10? There are a lot conflict when cherry-pick directly. @AnonHxy
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org