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/01/30 10:09:52 UTC
[GitHub] [pulsar] RobertIndie opened a new pull request #14069: [Broker] Make PersistentTopicsBase#internalDeleteTopic async
RobertIndie opened a new pull request #14069:
URL: https://github.com/apache/pulsar/pull/14069
Master Issue: #14013
### Motivation
See #14013
### Modifications
* Make PersistentTopicsBase#internalDeleteTopic async
### Verifying this change
This change is a trivial rework / code cleanup without any test coverage.
### Documentation
- [x] `no-need-doc`
--
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] nodece commented on a change in pull request #14069: [Broker] Make PersistentTopicsBase#internalDeleteTopic async
Posted by GitBox <gi...@apache.org>.
nodece commented on a change in pull request #14069:
URL: https://github.com/apache/pulsar/pull/14069#discussion_r801264888
##########
File path: pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/PersistentTopicsTest.java
##########
@@ -1151,28 +1151,29 @@ public void testDeleteTopic() throws Exception {
BrokerService brokerService = spy(pulsar.getBrokerService());
doReturn(brokerService).when(pulsar).getBrokerService();
persistentTopics.createNonPartitionedTopic(testTenant, testNamespace, topicName, false, null);
+
CompletableFuture<Void> deleteTopicFuture = new CompletableFuture<>();
deleteTopicFuture.completeExceptionally(new MetadataStoreException.NotFoundException());
doReturn(deleteTopicFuture).when(brokerService).deleteTopic(anyString(), anyBoolean(), anyBoolean());
- persistentTopics.deleteTopic(testTenant, testNamespace, topicName, true, true, true);
- //
+ AsyncResponse response = mock(AsyncResponse.class);
+ ArgumentCaptor<Response> responseCaptor = ArgumentCaptor.forClass(Response.class);
+ persistentTopics.deleteTopic(response, testTenant, testNamespace, topicName, true, true, true);
+ verify(response, timeout(5000).times(1)).resume(responseCaptor.capture());
+
+ ArgumentCaptor<RestException> errorCaptor = ArgumentCaptor.forClass(RestException.class);
CompletableFuture<Void> deleteTopicFuture2 = new CompletableFuture<>();
deleteTopicFuture2.completeExceptionally(new MetadataStoreException("test exception"));
doReturn(deleteTopicFuture2).when(brokerService).deleteTopic(anyString(), anyBoolean(), anyBoolean());
- try {
- persistentTopics.deleteTopic(testTenant, testNamespace, topicName, true, true, true);
- } catch (Exception e) {
- Assert.assertTrue(e instanceof RestException);
- }
- //
+ response = mock(AsyncResponse.class);
+ persistentTopics.deleteTopic(response, testTenant, testNamespace, topicName, true, true, true);
+ verify(response, timeout(5000).times(1)).resume(errorCaptor.capture());
+
Review comment:
Please check the `errorCaptor.getValue()`: has exception and instanceof RestException
--
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] nodece commented on a change in pull request #14069: [Broker] Make PersistentTopicsBase#internalDeleteTopic async
Posted by GitBox <gi...@apache.org>.
nodece commented on a change in pull request #14069:
URL: https://github.com/apache/pulsar/pull/14069#discussion_r801247057
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java
##########
@@ -1007,32 +1013,36 @@ private void internalUnloadTransactionCoordinatorAsync(AsyncResponse asyncRespon
});
}
- protected void internalDeleteTopic(boolean authoritative, boolean force, boolean deleteSchema) {
+ protected void internalDeleteTopic(AsyncResponse asyncResponse, boolean authoritative, boolean force,
+ boolean deleteSchema) {
if (force) {
- internalDeleteTopicForcefully(authoritative, deleteSchema);
+ internalDeleteTopicForcefully(asyncResponse, authoritative, deleteSchema);
} else {
- internalDeleteTopic(authoritative, deleteSchema);
+ internalDeleteTopic(asyncResponse, authoritative, deleteSchema);
}
}
- protected void internalDeleteTopic(boolean authoritative, boolean deleteSchema) {
- validateNamespaceOperation(topicName.getNamespaceObject(), NamespaceOperation.DELETE_TOPIC);
- validateTopicOwnership(topicName, authoritative);
-
- try {
- pulsar().getBrokerService().deleteTopic(topicName.toString(), false, deleteSchema).get();
- log.info("[{}] Successfully removed topic {}", clientAppId(), topicName);
- } catch (Exception e) {
- Throwable t = e.getCause();
- log.error("[{}] Failed to delete topic {}", clientAppId(), topicName, t);
- if (t instanceof TopicBusyException) {
- throw new RestException(Status.PRECONDITION_FAILED, "Topic has active producers/subscriptions");
- } else if (isManagedLedgerNotFoundException(e)) {
- throw new RestException(Status.NOT_FOUND, "Topic not found");
- } else {
- throw new RestException(t);
- }
- }
+ protected void internalDeleteTopic(AsyncResponse asyncResponse, boolean authoritative, boolean deleteSchema) {
+ validateNamespaceOperationAsync(topicName.getNamespaceObject(), NamespaceOperation.DELETE_TOPIC)
+ .thenCompose(__ -> validateTopicOwnershipAsync(topicName, authoritative))
+ .thenCompose(__ -> pulsar().getBrokerService().deleteTopic(topicName.toString(), false, deleteSchema))
+ .thenRun(() -> {
+ log.info("[{}] Successfully removed topic {}", clientAppId(), topicName);
+ asyncResponse.resume(Response.noContent().build());
+ })
+ .exceptionally(e -> {
+ Throwable t = e.getCause();
Review comment:
```suggestion
Throwable t = FutureUtil.unwrapCompletionException(e);
```
--
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] nodece commented on a change in pull request #14069: [Broker] Make PersistentTopicsBase#internalDeleteTopic async
Posted by GitBox <gi...@apache.org>.
nodece commented on a change in pull request #14069:
URL: https://github.com/apache/pulsar/pull/14069#discussion_r801265908
##########
File path: pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/PersistentTopicsTest.java
##########
@@ -1151,28 +1151,29 @@ public void testDeleteTopic() throws Exception {
BrokerService brokerService = spy(pulsar.getBrokerService());
doReturn(brokerService).when(pulsar).getBrokerService();
persistentTopics.createNonPartitionedTopic(testTenant, testNamespace, topicName, false, null);
+
CompletableFuture<Void> deleteTopicFuture = new CompletableFuture<>();
deleteTopicFuture.completeExceptionally(new MetadataStoreException.NotFoundException());
doReturn(deleteTopicFuture).when(brokerService).deleteTopic(anyString(), anyBoolean(), anyBoolean());
- persistentTopics.deleteTopic(testTenant, testNamespace, topicName, true, true, true);
- //
+ AsyncResponse response = mock(AsyncResponse.class);
+ ArgumentCaptor<Response> responseCaptor = ArgumentCaptor.forClass(Response.class);
+ persistentTopics.deleteTopic(response, testTenant, testNamespace, topicName, true, true, true);
+ verify(response, timeout(5000).times(1)).resume(responseCaptor.capture());
+
+ ArgumentCaptor<RestException> errorCaptor = ArgumentCaptor.forClass(RestException.class);
CompletableFuture<Void> deleteTopicFuture2 = new CompletableFuture<>();
deleteTopicFuture2.completeExceptionally(new MetadataStoreException("test exception"));
doReturn(deleteTopicFuture2).when(brokerService).deleteTopic(anyString(), anyBoolean(), anyBoolean());
- try {
- persistentTopics.deleteTopic(testTenant, testNamespace, topicName, true, true, true);
- } catch (Exception e) {
- Assert.assertTrue(e instanceof RestException);
- }
- //
+ response = mock(AsyncResponse.class);
+ persistentTopics.deleteTopic(response, testTenant, testNamespace, topicName, true, true, true);
+ verify(response, timeout(5000).times(1)).resume(errorCaptor.capture());
+
CompletableFuture<Void> deleteTopicFuture3 = new CompletableFuture<>();
deleteTopicFuture3.completeExceptionally(new MetadataStoreException.NotFoundException());
doReturn(deleteTopicFuture3).when(brokerService).deleteTopic(anyString(), anyBoolean(), anyBoolean());
- try {
- persistentTopics.deleteTopic(testTenant, testNamespace, topicName, false, true, true);
- } catch (RestException e) {
- Assert.assertEquals(e.getResponse().getStatus(), 404);
- }
+ response = mock(AsyncResponse.class);
+ persistentTopics.deleteTopic(response, testTenant, testNamespace, topicName, false, true, true);
+ verify(response, timeout(5000).times(1)).resume(errorCaptor.capture());
Review comment:
Please check the `errorCaptor.getValue()`.
--
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] gaoran10 edited a comment on pull request #14069: [Broker] Make PersistentTopicsBase#internalDeleteTopic async
Posted by GitBox <gi...@apache.org>.
gaoran10 edited a comment on pull request #14069:
URL: https://github.com/apache/pulsar/pull/14069#issuecomment-1032190166
Could we try to merge these two delete methods?
--
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] RobertIndie commented on a change in pull request #14069: [Broker] Make PersistentTopicsBase#internalDeleteTopic async
Posted by GitBox <gi...@apache.org>.
RobertIndie commented on a change in pull request #14069:
URL: https://github.com/apache/pulsar/pull/14069#discussion_r804353368
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v1/PersistentTopics.java
##########
@@ -301,13 +301,21 @@ public void unloadTopic(@Suspended final AsyncResponse asyncResponse, @PathParam
@ApiResponse(code = 403, message = "Don't have admin permission"),
@ApiResponse(code = 404, message = "Topic does not exist"),
@ApiResponse(code = 412, message = "Topic has active producers/subscriptions")})
- public void deleteTopic(@PathParam("property") String property, @PathParam("cluster") String cluster,
- @PathParam("namespace") String namespace, @PathParam("topic") @Encoded String encodedTopic,
- @QueryParam("force") @DefaultValue("false") boolean force,
- @QueryParam("authoritative") @DefaultValue("false") boolean authoritative,
- @QueryParam("deleteSchema") @DefaultValue("false") boolean deleteSchema) {
+ public void deleteTopic(@Suspended AsyncResponse asyncResponse, @PathParam("property") String property,
+ @PathParam("cluster") String cluster,
+ @PathParam("namespace") String namespace, @PathParam("topic") @Encoded String encodedTopic,
+ @QueryParam("force") @DefaultValue("false") boolean force,
+ @QueryParam("authoritative") @DefaultValue("false") boolean authoritative,
+ @QueryParam("deleteSchema") @DefaultValue("false") boolean deleteSchema) {
validateTopicName(property, cluster, namespace, encodedTopic);
Review comment:
Same as https://github.com/apache/pulsar/pull/14064#discussion_r804353275
--
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] liudezhi2098 commented on a change in pull request #14069: [Broker] Make PersistentTopicsBase#internalDeleteTopic async
Posted by GitBox <gi...@apache.org>.
liudezhi2098 commented on a change in pull request #14069:
URL: https://github.com/apache/pulsar/pull/14069#discussion_r803468189
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v1/PersistentTopics.java
##########
@@ -301,13 +301,21 @@ public void unloadTopic(@Suspended final AsyncResponse asyncResponse, @PathParam
@ApiResponse(code = 403, message = "Don't have admin permission"),
@ApiResponse(code = 404, message = "Topic does not exist"),
@ApiResponse(code = 412, message = "Topic has active producers/subscriptions")})
- public void deleteTopic(@PathParam("property") String property, @PathParam("cluster") String cluster,
- @PathParam("namespace") String namespace, @PathParam("topic") @Encoded String encodedTopic,
- @QueryParam("force") @DefaultValue("false") boolean force,
- @QueryParam("authoritative") @DefaultValue("false") boolean authoritative,
- @QueryParam("deleteSchema") @DefaultValue("false") boolean deleteSchema) {
+ public void deleteTopic(@Suspended AsyncResponse asyncResponse, @PathParam("property") String property,
+ @PathParam("cluster") String cluster,
+ @PathParam("namespace") String namespace, @PathParam("topic") @Encoded String encodedTopic,
+ @QueryParam("force") @DefaultValue("false") boolean force,
+ @QueryParam("authoritative") @DefaultValue("false") boolean authoritative,
+ @QueryParam("deleteSchema") @DefaultValue("false") boolean deleteSchema) {
validateTopicName(property, cluster, namespace, encodedTopic);
Review comment:
need catch exception
--
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] shibd commented on pull request #14069: [Broker] Make PersistentTopicsBase#internalDeleteTopic async
Posted by GitBox <gi...@apache.org>.
shibd commented on pull request #14069:
URL: https://github.com/apache/pulsar/pull/14069#issuecomment-1025185372
LGTM
--
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] nodece commented on a change in pull request #14069: [Broker] Make PersistentTopicsBase#internalDeleteTopic async
Posted by GitBox <gi...@apache.org>.
nodece commented on a change in pull request #14069:
URL: https://github.com/apache/pulsar/pull/14069#discussion_r801247302
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java
##########
@@ -1007,32 +1013,36 @@ private void internalUnloadTransactionCoordinatorAsync(AsyncResponse asyncRespon
});
}
- protected void internalDeleteTopic(boolean authoritative, boolean force, boolean deleteSchema) {
+ protected void internalDeleteTopic(AsyncResponse asyncResponse, boolean authoritative, boolean force,
+ boolean deleteSchema) {
if (force) {
- internalDeleteTopicForcefully(authoritative, deleteSchema);
+ internalDeleteTopicForcefully(asyncResponse, authoritative, deleteSchema);
} else {
- internalDeleteTopic(authoritative, deleteSchema);
+ internalDeleteTopic(asyncResponse, authoritative, deleteSchema);
}
}
- protected void internalDeleteTopic(boolean authoritative, boolean deleteSchema) {
- validateNamespaceOperation(topicName.getNamespaceObject(), NamespaceOperation.DELETE_TOPIC);
- validateTopicOwnership(topicName, authoritative);
-
- try {
- pulsar().getBrokerService().deleteTopic(topicName.toString(), false, deleteSchema).get();
- log.info("[{}] Successfully removed topic {}", clientAppId(), topicName);
- } catch (Exception e) {
- Throwable t = e.getCause();
- log.error("[{}] Failed to delete topic {}", clientAppId(), topicName, t);
- if (t instanceof TopicBusyException) {
- throw new RestException(Status.PRECONDITION_FAILED, "Topic has active producers/subscriptions");
- } else if (isManagedLedgerNotFoundException(e)) {
- throw new RestException(Status.NOT_FOUND, "Topic not found");
- } else {
- throw new RestException(t);
- }
- }
+ protected void internalDeleteTopic(AsyncResponse asyncResponse, boolean authoritative, boolean deleteSchema) {
+ validateNamespaceOperationAsync(topicName.getNamespaceObject(), NamespaceOperation.DELETE_TOPIC)
+ .thenCompose(__ -> validateTopicOwnershipAsync(topicName, authoritative))
+ .thenCompose(__ -> pulsar().getBrokerService().deleteTopic(topicName.toString(), false, deleteSchema))
+ .thenRun(() -> {
+ log.info("[{}] Successfully removed topic {}", clientAppId(), topicName);
+ asyncResponse.resume(Response.noContent().build());
+ })
+ .exceptionally(e -> {
+ Throwable t = e.getCause();
+ log.error("[{}] Failed to delete topic {}", clientAppId(), topicName, t);
+ if (t instanceof TopicBusyException) {
+ asyncResponse.resume(new RestException(Status.PRECONDITION_FAILED,
+ "Topic has active producers/subscriptions"));
+ } else if (isManagedLedgerNotFoundException(t)) {
+ asyncResponse.resume(new RestException(Status.NOT_FOUND, "Topic not found"));
+ } else {
+ asyncResponse.resume(new RestException(t));
Review comment:
```suggestion
resumeAsyncResponseExceptionally(asyncResponse, t);
```
--
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] nodece commented on a change in pull request #14069: [Broker] Make PersistentTopicsBase#internalDeleteTopic async
Posted by GitBox <gi...@apache.org>.
nodece commented on a change in pull request #14069:
URL: https://github.com/apache/pulsar/pull/14069#discussion_r801264986
##########
File path: pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/PersistentTopicsTest.java
##########
@@ -1151,28 +1151,29 @@ public void testDeleteTopic() throws Exception {
BrokerService brokerService = spy(pulsar.getBrokerService());
doReturn(brokerService).when(pulsar).getBrokerService();
persistentTopics.createNonPartitionedTopic(testTenant, testNamespace, topicName, false, null);
+
CompletableFuture<Void> deleteTopicFuture = new CompletableFuture<>();
deleteTopicFuture.completeExceptionally(new MetadataStoreException.NotFoundException());
doReturn(deleteTopicFuture).when(brokerService).deleteTopic(anyString(), anyBoolean(), anyBoolean());
- persistentTopics.deleteTopic(testTenant, testNamespace, topicName, true, true, true);
- //
+ AsyncResponse response = mock(AsyncResponse.class);
+ ArgumentCaptor<Response> responseCaptor = ArgumentCaptor.forClass(Response.class);
+ persistentTopics.deleteTopic(response, testTenant, testNamespace, topicName, true, true, true);
+ verify(response, timeout(5000).times(1)).resume(responseCaptor.capture());
+
+ ArgumentCaptor<RestException> errorCaptor = ArgumentCaptor.forClass(RestException.class);
CompletableFuture<Void> deleteTopicFuture2 = new CompletableFuture<>();
deleteTopicFuture2.completeExceptionally(new MetadataStoreException("test exception"));
doReturn(deleteTopicFuture2).when(brokerService).deleteTopic(anyString(), anyBoolean(), anyBoolean());
- try {
- persistentTopics.deleteTopic(testTenant, testNamespace, topicName, true, true, true);
- } catch (Exception e) {
- Assert.assertTrue(e instanceof RestException);
- }
- //
+ response = mock(AsyncResponse.class);
+ persistentTopics.deleteTopic(response, testTenant, testNamespace, topicName, true, true, true);
+ verify(response, timeout(5000).times(1)).resume(errorCaptor.capture());
+
CompletableFuture<Void> deleteTopicFuture3 = new CompletableFuture<>();
deleteTopicFuture3.completeExceptionally(new MetadataStoreException.NotFoundException());
doReturn(deleteTopicFuture3).when(brokerService).deleteTopic(anyString(), anyBoolean(), anyBoolean());
- try {
- persistentTopics.deleteTopic(testTenant, testNamespace, topicName, false, true, true);
- } catch (RestException e) {
- Assert.assertEquals(e.getResponse().getStatus(), 404);
- }
+ response = mock(AsyncResponse.class);
+ persistentTopics.deleteTopic(response, testTenant, testNamespace, topicName, false, true, true);
+ verify(response, timeout(5000).times(1)).resume(errorCaptor.capture());
}
Review comment:
Please check the `errorCaptor.getValue()`.
--
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] nodece commented on a change in pull request #14069: [Broker] Make PersistentTopicsBase#internalDeleteTopic async
Posted by GitBox <gi...@apache.org>.
nodece commented on a change in pull request #14069:
URL: https://github.com/apache/pulsar/pull/14069#discussion_r801264888
##########
File path: pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/PersistentTopicsTest.java
##########
@@ -1151,28 +1151,29 @@ public void testDeleteTopic() throws Exception {
BrokerService brokerService = spy(pulsar.getBrokerService());
doReturn(brokerService).when(pulsar).getBrokerService();
persistentTopics.createNonPartitionedTopic(testTenant, testNamespace, topicName, false, null);
+
CompletableFuture<Void> deleteTopicFuture = new CompletableFuture<>();
deleteTopicFuture.completeExceptionally(new MetadataStoreException.NotFoundException());
doReturn(deleteTopicFuture).when(brokerService).deleteTopic(anyString(), anyBoolean(), anyBoolean());
- persistentTopics.deleteTopic(testTenant, testNamespace, topicName, true, true, true);
- //
+ AsyncResponse response = mock(AsyncResponse.class);
+ ArgumentCaptor<Response> responseCaptor = ArgumentCaptor.forClass(Response.class);
+ persistentTopics.deleteTopic(response, testTenant, testNamespace, topicName, true, true, true);
+ verify(response, timeout(5000).times(1)).resume(responseCaptor.capture());
+
+ ArgumentCaptor<RestException> errorCaptor = ArgumentCaptor.forClass(RestException.class);
CompletableFuture<Void> deleteTopicFuture2 = new CompletableFuture<>();
deleteTopicFuture2.completeExceptionally(new MetadataStoreException("test exception"));
doReturn(deleteTopicFuture2).when(brokerService).deleteTopic(anyString(), anyBoolean(), anyBoolean());
- try {
- persistentTopics.deleteTopic(testTenant, testNamespace, topicName, true, true, true);
- } catch (Exception e) {
- Assert.assertTrue(e instanceof RestException);
- }
- //
+ response = mock(AsyncResponse.class);
+ persistentTopics.deleteTopic(response, testTenant, testNamespace, topicName, true, true, true);
+ verify(response, timeout(5000).times(1)).resume(errorCaptor.capture());
+
Review comment:
Please check the `errorCaptor.getValue()`.
--
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] github-actions[bot] commented on pull request #14069: [Broker] Make PersistentTopicsBase#internalDeleteTopic async
Posted by GitBox <gi...@apache.org>.
github-actions[bot] commented on pull request #14069:
URL: https://github.com/apache/pulsar/pull/14069#issuecomment-1072916876
The pr had no activity for 30 days, mark with Stale 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] RobertIndie commented on a change in pull request #14069: [Broker] Make PersistentTopicsBase#internalDeleteTopic async
Posted by GitBox <gi...@apache.org>.
RobertIndie commented on a change in pull request #14069:
URL: https://github.com/apache/pulsar/pull/14069#discussion_r802293430
##########
File path: pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/PersistentTopicsTest.java
##########
@@ -1151,28 +1151,29 @@ public void testDeleteTopic() throws Exception {
BrokerService brokerService = spy(pulsar.getBrokerService());
doReturn(brokerService).when(pulsar).getBrokerService();
persistentTopics.createNonPartitionedTopic(testTenant, testNamespace, topicName, false, null);
+
CompletableFuture<Void> deleteTopicFuture = new CompletableFuture<>();
deleteTopicFuture.completeExceptionally(new MetadataStoreException.NotFoundException());
doReturn(deleteTopicFuture).when(brokerService).deleteTopic(anyString(), anyBoolean(), anyBoolean());
- persistentTopics.deleteTopic(testTenant, testNamespace, topicName, true, true, true);
- //
+ AsyncResponse response = mock(AsyncResponse.class);
+ ArgumentCaptor<Response> responseCaptor = ArgumentCaptor.forClass(Response.class);
+ persistentTopics.deleteTopic(response, testTenant, testNamespace, topicName, true, true, true);
+ verify(response, timeout(5000).times(1)).resume(responseCaptor.capture());
+
+ ArgumentCaptor<RestException> errorCaptor = ArgumentCaptor.forClass(RestException.class);
CompletableFuture<Void> deleteTopicFuture2 = new CompletableFuture<>();
deleteTopicFuture2.completeExceptionally(new MetadataStoreException("test exception"));
doReturn(deleteTopicFuture2).when(brokerService).deleteTopic(anyString(), anyBoolean(), anyBoolean());
- try {
- persistentTopics.deleteTopic(testTenant, testNamespace, topicName, true, true, true);
- } catch (Exception e) {
- Assert.assertTrue(e instanceof RestException);
- }
- //
+ response = mock(AsyncResponse.class);
+ persistentTopics.deleteTopic(response, testTenant, testNamespace, topicName, true, true, true);
+ verify(response, timeout(5000).times(1)).resume(errorCaptor.capture());
+
Review comment:
`errorCaptor` already checks that instance type.
--
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] gaoran10 commented on pull request #14069: [Broker] Make PersistentTopicsBase#internalDeleteTopic async
Posted by GitBox <gi...@apache.org>.
gaoran10 commented on pull request #14069:
URL: https://github.com/apache/pulsar/pull/14069#issuecomment-1032190166
Could we merge these two delete methods?
--
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] nodece commented on a change in pull request #14069: [Broker] Make PersistentTopicsBase#internalDeleteTopic async
Posted by GitBox <gi...@apache.org>.
nodece commented on a change in pull request #14069:
URL: https://github.com/apache/pulsar/pull/14069#discussion_r801246725
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java
##########
@@ -311,20 +311,26 @@ protected void internalGrantPermissionsOnTopic(String role, Set<AuthAction> acti
grantPermissions(topicName, role, actions);
}
- protected void internalDeleteTopicForcefully(boolean authoritative, boolean deleteSchema) {
- validateTopicOwnership(topicName, authoritative);
- validateNamespaceOperation(topicName.getNamespaceObject(), NamespaceOperation.DELETE_TOPIC);
-
- try {
- pulsar().getBrokerService().deleteTopic(topicName.toString(), true, deleteSchema).get();
- } catch (Exception e) {
- if (isManagedLedgerNotFoundException(e)) {
- log.info("[{}] Topic was already not existing {}", clientAppId(), topicName, e);
- } else {
- log.error("[{}] Failed to delete topic forcefully {}", clientAppId(), topicName, e);
- throw new RestException(e);
- }
- }
+ protected void internalDeleteTopicForcefully(AsyncResponse asyncResponse, boolean authoritative,
+ boolean deleteSchema) {
+ validateTopicOwnershipAsync(topicName, authoritative)
+ .thenCompose(__ -> validateNamespaceOperationAsync(topicName.getNamespaceObject(),
+ NamespaceOperation.DELETE_TOPIC))
+ .thenCompose(__ -> pulsar().getBrokerService().deleteTopic(topicName.toString(), true, deleteSchema))
+ .thenRun(() -> {
+ log.info("[{}] Successfully removed topic {}", clientAppId(), topicName);
+ asyncResponse.resume(Response.noContent().build());
+ })
+ .exceptionally(e -> {
+ if (isManagedLedgerNotFoundException(e.getCause())) {
+ log.info("[{}] Topic was already not existing {}", clientAppId(), topicName, e);
Review comment:
```suggestion
log.info("[{}] Topic was already not existing {}", clientAppId(), topicName, throwable);
```
--
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] nodece commented on a change in pull request #14069: [Broker] Make PersistentTopicsBase#internalDeleteTopic async
Posted by GitBox <gi...@apache.org>.
nodece commented on a change in pull request #14069:
URL: https://github.com/apache/pulsar/pull/14069#discussion_r801246623
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java
##########
@@ -311,20 +311,26 @@ protected void internalGrantPermissionsOnTopic(String role, Set<AuthAction> acti
grantPermissions(topicName, role, actions);
}
- protected void internalDeleteTopicForcefully(boolean authoritative, boolean deleteSchema) {
- validateTopicOwnership(topicName, authoritative);
- validateNamespaceOperation(topicName.getNamespaceObject(), NamespaceOperation.DELETE_TOPIC);
-
- try {
- pulsar().getBrokerService().deleteTopic(topicName.toString(), true, deleteSchema).get();
- } catch (Exception e) {
- if (isManagedLedgerNotFoundException(e)) {
- log.info("[{}] Topic was already not existing {}", clientAppId(), topicName, e);
- } else {
- log.error("[{}] Failed to delete topic forcefully {}", clientAppId(), topicName, e);
- throw new RestException(e);
- }
- }
+ protected void internalDeleteTopicForcefully(AsyncResponse asyncResponse, boolean authoritative,
+ boolean deleteSchema) {
+ validateTopicOwnershipAsync(topicName, authoritative)
+ .thenCompose(__ -> validateNamespaceOperationAsync(topicName.getNamespaceObject(),
+ NamespaceOperation.DELETE_TOPIC))
+ .thenCompose(__ -> pulsar().getBrokerService().deleteTopic(topicName.toString(), true, deleteSchema))
+ .thenRun(() -> {
+ log.info("[{}] Successfully removed topic {}", clientAppId(), topicName);
+ asyncResponse.resume(Response.noContent().build());
+ })
+ .exceptionally(e -> {
+ if (isManagedLedgerNotFoundException(e.getCause())) {
Review comment:
```suggestion
if (isManagedLedgerNotFoundException(throwable)) {
```
--
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] nodece commented on a change in pull request #14069: [Broker] Make PersistentTopicsBase#internalDeleteTopic async
Posted by GitBox <gi...@apache.org>.
nodece commented on a change in pull request #14069:
URL: https://github.com/apache/pulsar/pull/14069#discussion_r801246538
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java
##########
@@ -311,20 +311,26 @@ protected void internalGrantPermissionsOnTopic(String role, Set<AuthAction> acti
grantPermissions(topicName, role, actions);
}
- protected void internalDeleteTopicForcefully(boolean authoritative, boolean deleteSchema) {
- validateTopicOwnership(topicName, authoritative);
- validateNamespaceOperation(topicName.getNamespaceObject(), NamespaceOperation.DELETE_TOPIC);
-
- try {
- pulsar().getBrokerService().deleteTopic(topicName.toString(), true, deleteSchema).get();
- } catch (Exception e) {
- if (isManagedLedgerNotFoundException(e)) {
- log.info("[{}] Topic was already not existing {}", clientAppId(), topicName, e);
- } else {
- log.error("[{}] Failed to delete topic forcefully {}", clientAppId(), topicName, e);
- throw new RestException(e);
- }
- }
+ protected void internalDeleteTopicForcefully(AsyncResponse asyncResponse, boolean authoritative,
+ boolean deleteSchema) {
+ validateTopicOwnershipAsync(topicName, authoritative)
+ .thenCompose(__ -> validateNamespaceOperationAsync(topicName.getNamespaceObject(),
+ NamespaceOperation.DELETE_TOPIC))
+ .thenCompose(__ -> pulsar().getBrokerService().deleteTopic(topicName.toString(), true, deleteSchema))
+ .thenRun(() -> {
+ log.info("[{}] Successfully removed topic {}", clientAppId(), topicName);
+ asyncResponse.resume(Response.noContent().build());
+ })
+ .exceptionally(e -> {
Review comment:
Throwable throwable = FutureUtil.unwrapCompletionException(e);
--
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] nodece commented on a change in pull request #14069: [Broker] Make PersistentTopicsBase#internalDeleteTopic async
Posted by GitBox <gi...@apache.org>.
nodece commented on a change in pull request #14069:
URL: https://github.com/apache/pulsar/pull/14069#discussion_r801264986
##########
File path: pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/PersistentTopicsTest.java
##########
@@ -1151,28 +1151,29 @@ public void testDeleteTopic() throws Exception {
BrokerService brokerService = spy(pulsar.getBrokerService());
doReturn(brokerService).when(pulsar).getBrokerService();
persistentTopics.createNonPartitionedTopic(testTenant, testNamespace, topicName, false, null);
+
CompletableFuture<Void> deleteTopicFuture = new CompletableFuture<>();
deleteTopicFuture.completeExceptionally(new MetadataStoreException.NotFoundException());
doReturn(deleteTopicFuture).when(brokerService).deleteTopic(anyString(), anyBoolean(), anyBoolean());
- persistentTopics.deleteTopic(testTenant, testNamespace, topicName, true, true, true);
- //
+ AsyncResponse response = mock(AsyncResponse.class);
+ ArgumentCaptor<Response> responseCaptor = ArgumentCaptor.forClass(Response.class);
+ persistentTopics.deleteTopic(response, testTenant, testNamespace, topicName, true, true, true);
+ verify(response, timeout(5000).times(1)).resume(responseCaptor.capture());
+
+ ArgumentCaptor<RestException> errorCaptor = ArgumentCaptor.forClass(RestException.class);
CompletableFuture<Void> deleteTopicFuture2 = new CompletableFuture<>();
deleteTopicFuture2.completeExceptionally(new MetadataStoreException("test exception"));
doReturn(deleteTopicFuture2).when(brokerService).deleteTopic(anyString(), anyBoolean(), anyBoolean());
- try {
- persistentTopics.deleteTopic(testTenant, testNamespace, topicName, true, true, true);
- } catch (Exception e) {
- Assert.assertTrue(e instanceof RestException);
- }
- //
+ response = mock(AsyncResponse.class);
+ persistentTopics.deleteTopic(response, testTenant, testNamespace, topicName, true, true, true);
+ verify(response, timeout(5000).times(1)).resume(errorCaptor.capture());
+
CompletableFuture<Void> deleteTopicFuture3 = new CompletableFuture<>();
deleteTopicFuture3.completeExceptionally(new MetadataStoreException.NotFoundException());
doReturn(deleteTopicFuture3).when(brokerService).deleteTopic(anyString(), anyBoolean(), anyBoolean());
- try {
- persistentTopics.deleteTopic(testTenant, testNamespace, topicName, false, true, true);
- } catch (RestException e) {
- Assert.assertEquals(e.getResponse().getStatus(), 404);
- }
+ response = mock(AsyncResponse.class);
+ persistentTopics.deleteTopic(response, testTenant, testNamespace, topicName, false, true, true);
+ verify(response, timeout(5000).times(1)).resume(errorCaptor.capture());
}
Review comment:
Please check the `errorCaptor.getValue()`, the http status == 404
--
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] nodece commented on a change in pull request #14069: [Broker] Make PersistentTopicsBase#internalDeleteTopic async
Posted by GitBox <gi...@apache.org>.
nodece commented on a change in pull request #14069:
URL: https://github.com/apache/pulsar/pull/14069#discussion_r801264413
##########
File path: pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/PersistentTopicsTest.java
##########
@@ -1151,28 +1151,29 @@ public void testDeleteTopic() throws Exception {
BrokerService brokerService = spy(pulsar.getBrokerService());
doReturn(brokerService).when(pulsar).getBrokerService();
persistentTopics.createNonPartitionedTopic(testTenant, testNamespace, topicName, false, null);
+
CompletableFuture<Void> deleteTopicFuture = new CompletableFuture<>();
deleteTopicFuture.completeExceptionally(new MetadataStoreException.NotFoundException());
doReturn(deleteTopicFuture).when(brokerService).deleteTopic(anyString(), anyBoolean(), anyBoolean());
- persistentTopics.deleteTopic(testTenant, testNamespace, topicName, true, true, true);
- //
+ AsyncResponse response = mock(AsyncResponse.class);
+ ArgumentCaptor<Response> responseCaptor = ArgumentCaptor.forClass(Response.class);
+ persistentTopics.deleteTopic(response, testTenant, testNamespace, topicName, true, true, true);
+ verify(response, timeout(5000).times(1)).resume(responseCaptor.capture());
+
Review comment:
Please check the `responseCaptor.getValue().getStatus() == 204`
--
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] nodece commented on a change in pull request #14069: [Broker] Make PersistentTopicsBase#internalDeleteTopic async
Posted by GitBox <gi...@apache.org>.
nodece commented on a change in pull request #14069:
URL: https://github.com/apache/pulsar/pull/14069#discussion_r801246933
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java
##########
@@ -311,20 +311,26 @@ protected void internalGrantPermissionsOnTopic(String role, Set<AuthAction> acti
grantPermissions(topicName, role, actions);
}
- protected void internalDeleteTopicForcefully(boolean authoritative, boolean deleteSchema) {
- validateTopicOwnership(topicName, authoritative);
- validateNamespaceOperation(topicName.getNamespaceObject(), NamespaceOperation.DELETE_TOPIC);
-
- try {
- pulsar().getBrokerService().deleteTopic(topicName.toString(), true, deleteSchema).get();
- } catch (Exception e) {
- if (isManagedLedgerNotFoundException(e)) {
- log.info("[{}] Topic was already not existing {}", clientAppId(), topicName, e);
- } else {
- log.error("[{}] Failed to delete topic forcefully {}", clientAppId(), topicName, e);
- throw new RestException(e);
- }
- }
+ protected void internalDeleteTopicForcefully(AsyncResponse asyncResponse, boolean authoritative,
+ boolean deleteSchema) {
+ validateTopicOwnershipAsync(topicName, authoritative)
+ .thenCompose(__ -> validateNamespaceOperationAsync(topicName.getNamespaceObject(),
+ NamespaceOperation.DELETE_TOPIC))
+ .thenCompose(__ -> pulsar().getBrokerService().deleteTopic(topicName.toString(), true, deleteSchema))
+ .thenRun(() -> {
+ log.info("[{}] Successfully removed topic {}", clientAppId(), topicName);
+ asyncResponse.resume(Response.noContent().build());
+ })
+ .exceptionally(e -> {
+ if (isManagedLedgerNotFoundException(e.getCause())) {
+ log.info("[{}] Topic was already not existing {}", clientAppId(), topicName, e);
+ asyncResponse.resume(Response.noContent().build());
+ } else {
+ log.error("[{}] Failed to delete topic forcefully {}", clientAppId(), topicName, e);
+ asyncResponse.resume(new RestException(e));
Review comment:
```suggestion
resumeAsyncResponseExceptionally(asyncResponse, e);
```
--
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] nodece commented on a change in pull request #14069: [Broker] Make PersistentTopicsBase#internalDeleteTopic async
Posted by GitBox <gi...@apache.org>.
nodece commented on a change in pull request #14069:
URL: https://github.com/apache/pulsar/pull/14069#discussion_r801246387
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java
##########
@@ -311,20 +311,26 @@ protected void internalGrantPermissionsOnTopic(String role, Set<AuthAction> acti
grantPermissions(topicName, role, actions);
}
- protected void internalDeleteTopicForcefully(boolean authoritative, boolean deleteSchema) {
- validateTopicOwnership(topicName, authoritative);
- validateNamespaceOperation(topicName.getNamespaceObject(), NamespaceOperation.DELETE_TOPIC);
-
- try {
- pulsar().getBrokerService().deleteTopic(topicName.toString(), true, deleteSchema).get();
- } catch (Exception e) {
- if (isManagedLedgerNotFoundException(e)) {
- log.info("[{}] Topic was already not existing {}", clientAppId(), topicName, e);
- } else {
- log.error("[{}] Failed to delete topic forcefully {}", clientAppId(), topicName, e);
- throw new RestException(e);
- }
- }
+ protected void internalDeleteTopicForcefully(AsyncResponse asyncResponse, boolean authoritative,
+ boolean deleteSchema) {
+ validateTopicOwnershipAsync(topicName, authoritative)
+ .thenCompose(__ -> validateNamespaceOperationAsync(topicName.getNamespaceObject(),
+ NamespaceOperation.DELETE_TOPIC))
+ .thenCompose(__ -> pulsar().getBrokerService().deleteTopic(topicName.toString(), true, deleteSchema))
+ .thenRun(() -> {
+ log.info("[{}] Successfully removed topic {}", clientAppId(), topicName);
+ asyncResponse.resume(Response.noContent().build());
+ })
+ .exceptionally(e -> {
+ if (isManagedLedgerNotFoundException(e.getCause())) {
Review comment:
Throwable throwable = FutureUtil.unwrapCompletionException(e);
--
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] nodece commented on a change in pull request #14069: [Broker] Make PersistentTopicsBase#internalDeleteTopic async
Posted by GitBox <gi...@apache.org>.
nodece commented on a change in pull request #14069:
URL: https://github.com/apache/pulsar/pull/14069#discussion_r801264413
##########
File path: pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/PersistentTopicsTest.java
##########
@@ -1151,28 +1151,29 @@ public void testDeleteTopic() throws Exception {
BrokerService brokerService = spy(pulsar.getBrokerService());
doReturn(brokerService).when(pulsar).getBrokerService();
persistentTopics.createNonPartitionedTopic(testTenant, testNamespace, topicName, false, null);
+
CompletableFuture<Void> deleteTopicFuture = new CompletableFuture<>();
deleteTopicFuture.completeExceptionally(new MetadataStoreException.NotFoundException());
doReturn(deleteTopicFuture).when(brokerService).deleteTopic(anyString(), anyBoolean(), anyBoolean());
- persistentTopics.deleteTopic(testTenant, testNamespace, topicName, true, true, true);
- //
+ AsyncResponse response = mock(AsyncResponse.class);
+ ArgumentCaptor<Response> responseCaptor = ArgumentCaptor.forClass(Response.class);
+ persistentTopics.deleteTopic(response, testTenant, testNamespace, topicName, true, true, true);
+ verify(response, timeout(5000).times(1)).resume(responseCaptor.capture());
+
Review comment:
Please check the `responseCaptor.getValue().getStatus()`
--
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