poorbarcode commented on code in PR #18283:
URL: https://github.com/apache/pulsar/pull/18283#discussion_r1011214669
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java:
##########
@@ -1179,30 +1191,24 @@ private CompletableFuture<Void> delete(boolean
failIfHasSubscriptions,
CompletableFuture<Void> deleteFuture = new
CompletableFuture<>();
CompletableFuture<Void> closeClientFuture = new
CompletableFuture<>();
+ List<CompletableFuture<Void>> futures = new ArrayList<>();
+ subscriptions.forEach((s, sub) ->
futures.add(sub.disconnect()));
if (closeIfClientsConnected) {
- List<CompletableFuture<Void>> futures = new ArrayList<>();
replicators.forEach((cluster, replicator) ->
futures.add(replicator.disconnect()));
shadowReplicators.forEach((__, replicator) ->
futures.add(replicator.disconnect()));
producers.values().forEach(producer ->
futures.add(producer.disconnect()));
- subscriptions.forEach((s, sub) ->
futures.add(sub.disconnect()));
- FutureUtil.waitForAll(futures).thenRun(() -> {
- closeClientFuture.complete(null);
- }).exceptionally(ex -> {
- log.error("[{}] Error closing clients", topic, ex);
- unfenceTopicToResume();
- closeClientFuture.completeExceptionally(ex);
- return null;
- });
- } else {
- closeClientFuture.complete(null);
}
+ FutureUtil.waitForAll(futures).thenRun(() -> {
+ closeClientFuture.complete(null);
+ }).exceptionally(ex -> {
+ log.error("[{}] Error closing clients", topic, ex);
+ unfenceTopicToResume();
+ closeClientFuture.completeExceptionally(ex);
+ return null;
+ });
closeClientFuture.thenAccept(delete -> {
- // We can proceed with the deletion if either:
- // 1. No one is connected
- // 2. We want to kick out everyone and forcefully delete
the topic.
- // In this case, we shouldn't care if the usageCount
is 0 or not, just proceed
- if (currentUsageCount() == 0 || (closeIfClientsConnected
&& !failIfHasSubscriptions)) {
+ if (currentUsageCount() == 0) {
Review Comment:
I feel the current code is correct. Because we have called disconnection
operation before. If there are still any connections at the current time, it
means that a new connection has come in, we should consider that there is a
concurrency problem and give up deleting the topic.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
