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/20 05:35:24 UTC
[GitHub] [pulsar] mattisonchao commented on a change in pull request #13845: [Broker] Fix call sync method in async rest api for internalUnloadTopic
mattisonchao commented on a change in pull request #13845:
URL: https://github.com/apache/pulsar/pull/13845#discussion_r788363349
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java
##########
@@ -714,70 +714,73 @@ protected void internalDeletePartitionedTopic(AsyncResponse asyncResponse, boole
protected void internalUnloadTopic(AsyncResponse asyncResponse, boolean authoritative) {
log.info("[{}] Unloading topic {}", clientAppId(), topicName);
- try {
- if (topicName.isGlobal()) {
- validateGlobalNamespaceOwnership(namespaceName);
- }
- } catch (Exception e) {
- log.error("[{}] Failed to unload topic {}", clientAppId(), topicName, e);
- resumeAsyncResponseExceptionally(asyncResponse, e);
- return;
- }
- // If the topic name is a partition name, no need to get partition topic metadata again
- if (topicName.isPartitioned()) {
- if (checkTopicIsTransactionCoordinatorAssign(topicName)) {
- internalUnloadTransactionCoordinator(asyncResponse, authoritative);
- } else {
- internalUnloadNonPartitionedTopic(asyncResponse, authoritative);
- }
- } else {
- getPartitionedTopicMetadataAsync(topicName, authoritative, false)
- .thenAccept(meta -> {
- if (meta.partitions > 0) {
- final List<CompletableFuture<Void>> futures = Lists.newArrayList();
-
- for (int i = 0; i < meta.partitions; i++) {
- TopicName topicNamePartition = topicName.getPartition(i);
- try {
- futures.add(pulsar().getAdminClient().topics().unloadAsync(
- topicNamePartition.toString()));
- } catch (Exception e) {
- log.error("[{}] Failed to unload topic {}", clientAppId(), topicNamePartition, e);
- asyncResponse.resume(new RestException(e));
- return;
- }
- }
-
- FutureUtil.waitForAll(futures).handle((result, exception) -> {
- if (exception != null) {
- Throwable th = exception.getCause();
- if (th instanceof NotFoundException) {
- asyncResponse.resume(new RestException(Status.NOT_FOUND, th.getMessage()));
- } else if (th instanceof WebApplicationException) {
- asyncResponse.resume(th);
- } else {
- log.error("[{}] Failed to unload topic {}", clientAppId(), topicName,
- exception);
- asyncResponse.resume(new RestException(exception));
- }
- } else {
- asyncResponse.resume(Response.noContent().build());
- }
- return null;
- });
- } else {
- internalUnloadNonPartitionedTopic(asyncResponse, authoritative);
- }
- }).exceptionally(t -> {
- log.error("[{}] Failed to unload topic {}", clientAppId(), topicName, t);
- if (t instanceof WebApplicationException) {
- asyncResponse.resume(t);
- } else {
- asyncResponse.resume(new RestException(t));
- }
- return null;
- });
- }
+ CompletableFuture<Void> future = CompletableFuture.completedFuture(null);
+ if (topicName.isGlobal()) {
+ future = validateGlobalNamespaceOwnershipAsync(namespaceName);
+ }
+ future.thenAccept(__ ->{
+ // If the topic name is a partition name, no need to get partition topic metadata again
+ if (topicName.isPartitioned()) {
+ if (checkTopicIsTransactionCoordinatorAssign(topicName)) {
+ internalUnloadTransactionCoordinatorAsync(asyncResponse, authoritative);
+ } else {
+ internalUnloadNonPartitionedTopicAsync(asyncResponse, authoritative);
+ }
+ } else {
+ getPartitionedTopicMetadataAsync(topicName, authoritative, false)
+ .thenAccept(meta -> {
+ if (meta.partitions > 0) {
+ final List<CompletableFuture<Void>> futures = Lists.newArrayList();
+
+ for (int i = 0; i < meta.partitions; i++) {
+ TopicName topicNamePartition = topicName.getPartition(i);
+ try {
+ futures.add(pulsar().getAdminClient().topics().unloadAsync(
+ topicNamePartition.toString()));
+ } catch (Exception e) {
+ log.error("[{}] Failed to unload topic {}", clientAppId(),
+ topicNamePartition, e);
+ asyncResponse.resume(new RestException(e));
+ return;
+ }
+ }
+
+ FutureUtil.waitForAll(futures).handle((result, exception) -> {
+ if (exception != null) {
+ Throwable th = exception.getCause();
+ if (th instanceof NotFoundException) {
+ asyncResponse.resume(new RestException(Status.NOT_FOUND, th.getMessage()));
+ } else if (th instanceof WebApplicationException) {
+ asyncResponse.resume(th);
+ } else {
+ log.error("[{}] Failed to unload topic {}", clientAppId(), topicName,
+ exception);
+ asyncResponse.resume(new RestException(exception));
+ }
+ } else {
+ asyncResponse.resume(Response.noContent().build());
+ }
+ return null;
+ });
+ } else {
+ internalUnloadNonPartitionedTopicAsync(asyncResponse, authoritative);
+ }
+ }).exceptionally(t -> {
+ log.error("[{}] Failed to unload topic {}", clientAppId(), topicName, t);
Review comment:
fixed
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java
##########
@@ -714,70 +714,73 @@ protected void internalDeletePartitionedTopic(AsyncResponse asyncResponse, boole
protected void internalUnloadTopic(AsyncResponse asyncResponse, boolean authoritative) {
log.info("[{}] Unloading topic {}", clientAppId(), topicName);
- try {
- if (topicName.isGlobal()) {
- validateGlobalNamespaceOwnership(namespaceName);
- }
- } catch (Exception e) {
- log.error("[{}] Failed to unload topic {}", clientAppId(), topicName, e);
- resumeAsyncResponseExceptionally(asyncResponse, e);
- return;
- }
- // If the topic name is a partition name, no need to get partition topic metadata again
- if (topicName.isPartitioned()) {
- if (checkTopicIsTransactionCoordinatorAssign(topicName)) {
- internalUnloadTransactionCoordinator(asyncResponse, authoritative);
- } else {
- internalUnloadNonPartitionedTopic(asyncResponse, authoritative);
- }
- } else {
- getPartitionedTopicMetadataAsync(topicName, authoritative, false)
- .thenAccept(meta -> {
- if (meta.partitions > 0) {
- final List<CompletableFuture<Void>> futures = Lists.newArrayList();
-
- for (int i = 0; i < meta.partitions; i++) {
- TopicName topicNamePartition = topicName.getPartition(i);
- try {
- futures.add(pulsar().getAdminClient().topics().unloadAsync(
- topicNamePartition.toString()));
- } catch (Exception e) {
- log.error("[{}] Failed to unload topic {}", clientAppId(), topicNamePartition, e);
- asyncResponse.resume(new RestException(e));
- return;
- }
- }
-
- FutureUtil.waitForAll(futures).handle((result, exception) -> {
- if (exception != null) {
- Throwable th = exception.getCause();
- if (th instanceof NotFoundException) {
- asyncResponse.resume(new RestException(Status.NOT_FOUND, th.getMessage()));
- } else if (th instanceof WebApplicationException) {
- asyncResponse.resume(th);
- } else {
- log.error("[{}] Failed to unload topic {}", clientAppId(), topicName,
- exception);
- asyncResponse.resume(new RestException(exception));
- }
- } else {
- asyncResponse.resume(Response.noContent().build());
- }
- return null;
- });
- } else {
- internalUnloadNonPartitionedTopic(asyncResponse, authoritative);
- }
- }).exceptionally(t -> {
- log.error("[{}] Failed to unload topic {}", clientAppId(), topicName, t);
- if (t instanceof WebApplicationException) {
- asyncResponse.resume(t);
- } else {
- asyncResponse.resume(new RestException(t));
- }
- return null;
- });
- }
+ CompletableFuture<Void> future = CompletableFuture.completedFuture(null);
+ if (topicName.isGlobal()) {
+ future = validateGlobalNamespaceOwnershipAsync(namespaceName);
+ }
+ future.thenAccept(__ ->{
+ // If the topic name is a partition name, no need to get partition topic metadata again
+ if (topicName.isPartitioned()) {
+ if (checkTopicIsTransactionCoordinatorAssign(topicName)) {
+ internalUnloadTransactionCoordinatorAsync(asyncResponse, authoritative);
+ } else {
+ internalUnloadNonPartitionedTopicAsync(asyncResponse, authoritative);
+ }
+ } else {
+ getPartitionedTopicMetadataAsync(topicName, authoritative, false)
+ .thenAccept(meta -> {
+ if (meta.partitions > 0) {
+ final List<CompletableFuture<Void>> futures = Lists.newArrayList();
Review comment:
fixed
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java
##########
@@ -714,70 +714,73 @@ protected void internalDeletePartitionedTopic(AsyncResponse asyncResponse, boole
protected void internalUnloadTopic(AsyncResponse asyncResponse, boolean authoritative) {
log.info("[{}] Unloading topic {}", clientAppId(), topicName);
- try {
- if (topicName.isGlobal()) {
- validateGlobalNamespaceOwnership(namespaceName);
- }
- } catch (Exception e) {
- log.error("[{}] Failed to unload topic {}", clientAppId(), topicName, e);
- resumeAsyncResponseExceptionally(asyncResponse, e);
- return;
- }
- // If the topic name is a partition name, no need to get partition topic metadata again
- if (topicName.isPartitioned()) {
- if (checkTopicIsTransactionCoordinatorAssign(topicName)) {
- internalUnloadTransactionCoordinator(asyncResponse, authoritative);
- } else {
- internalUnloadNonPartitionedTopic(asyncResponse, authoritative);
- }
- } else {
- getPartitionedTopicMetadataAsync(topicName, authoritative, false)
- .thenAccept(meta -> {
- if (meta.partitions > 0) {
- final List<CompletableFuture<Void>> futures = Lists.newArrayList();
-
- for (int i = 0; i < meta.partitions; i++) {
- TopicName topicNamePartition = topicName.getPartition(i);
- try {
- futures.add(pulsar().getAdminClient().topics().unloadAsync(
- topicNamePartition.toString()));
- } catch (Exception e) {
- log.error("[{}] Failed to unload topic {}", clientAppId(), topicNamePartition, e);
- asyncResponse.resume(new RestException(e));
- return;
- }
- }
-
- FutureUtil.waitForAll(futures).handle((result, exception) -> {
- if (exception != null) {
- Throwable th = exception.getCause();
- if (th instanceof NotFoundException) {
- asyncResponse.resume(new RestException(Status.NOT_FOUND, th.getMessage()));
- } else if (th instanceof WebApplicationException) {
- asyncResponse.resume(th);
- } else {
- log.error("[{}] Failed to unload topic {}", clientAppId(), topicName,
- exception);
- asyncResponse.resume(new RestException(exception));
- }
- } else {
- asyncResponse.resume(Response.noContent().build());
- }
- return null;
- });
- } else {
- internalUnloadNonPartitionedTopic(asyncResponse, authoritative);
- }
- }).exceptionally(t -> {
- log.error("[{}] Failed to unload topic {}", clientAppId(), topicName, t);
- if (t instanceof WebApplicationException) {
- asyncResponse.resume(t);
- } else {
- asyncResponse.resume(new RestException(t));
- }
- return null;
- });
- }
+ CompletableFuture<Void> future = CompletableFuture.completedFuture(null);
+ if (topicName.isGlobal()) {
+ future = validateGlobalNamespaceOwnershipAsync(namespaceName);
+ }
+ future.thenAccept(__ ->{
+ // If the topic name is a partition name, no need to get partition topic metadata again
+ if (topicName.isPartitioned()) {
+ if (checkTopicIsTransactionCoordinatorAssign(topicName)) {
+ internalUnloadTransactionCoordinatorAsync(asyncResponse, authoritative);
+ } else {
+ internalUnloadNonPartitionedTopicAsync(asyncResponse, authoritative);
+ }
+ } else {
+ getPartitionedTopicMetadataAsync(topicName, authoritative, false)
+ .thenAccept(meta -> {
+ if (meta.partitions > 0) {
+ final List<CompletableFuture<Void>> futures = Lists.newArrayList();
+
+ for (int i = 0; i < meta.partitions; i++) {
+ TopicName topicNamePartition = topicName.getPartition(i);
+ try {
+ futures.add(pulsar().getAdminClient().topics().unloadAsync(
+ topicNamePartition.toString()));
+ } catch (Exception e) {
+ log.error("[{}] Failed to unload topic {}", clientAppId(),
+ topicNamePartition, e);
+ asyncResponse.resume(new RestException(e));
+ return;
+ }
+ }
+
+ FutureUtil.waitForAll(futures).handle((result, exception) -> {
+ if (exception != null) {
+ Throwable th = exception.getCause();
+ if (th instanceof NotFoundException) {
+ asyncResponse.resume(new RestException(Status.NOT_FOUND, th.getMessage()));
+ } else if (th instanceof WebApplicationException) {
+ asyncResponse.resume(th);
+ } else {
+ log.error("[{}] Failed to unload topic {}", clientAppId(), topicName,
+ exception);
+ asyncResponse.resume(new RestException(exception));
+ }
+ } else {
+ asyncResponse.resume(Response.noContent().build());
+ }
+ return null;
+ });
+ } else {
+ internalUnloadNonPartitionedTopicAsync(asyncResponse, authoritative);
+ }
+ }).exceptionally(t -> {
+ log.error("[{}] Failed to unload topic {}", clientAppId(), topicName, t);
+ if (t instanceof WebApplicationException) {
+ asyncResponse.resume(t);
+ } else {
+ asyncResponse.resume(new RestException(t));
+ }
+ return null;
+ });
+ }
+ }).exceptionally(ex -> {
+ Throwable cause = ex.getCause();
+ log.error("[{}] Failed to unload topic {}", clientAppId(), topicName, cause);
Review comment:
fixed
--
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