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/02/25 07:57:03 UTC
[GitHub] [pulsar] mattisonchao opened a new pull request #14469: [Broker] Fix ``Future.join()`` causing deadlock.
mattisonchao opened a new pull request #14469:
URL: https://github.com/apache/pulsar/pull/14469
### Motivation
Invoking the ``join()`` method in the async method will cause some deadlock.
### Modifications
- Refactor ``PersistentTopic#tryToDeletePartitionedMetadata`` to pure async.
### Verifying this change
- [x] Make sure that the change passes the CI checks.
### 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] mattisonchao commented on a change in pull request #14469: [Broker] Fix ``Future.join()`` causing deadlock.
Posted by GitBox <gi...@apache.org>.
mattisonchao commented on a change in pull request #14469:
URL: https://github.com/apache/pulsar/pull/14469#discussion_r815406899
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java
##########
@@ -2256,42 +2256,43 @@ public void checkGC() {
return CompletableFuture.completedFuture(null);
}
TopicName topicName = TopicName.get(TopicName.get(topic).getPartitionedTopicName());
- try {
- PartitionedTopicResources partitionedTopicResources = getBrokerService().pulsar().getPulsarResources()
- .getNamespaceResources()
- .getPartitionedTopicResources();
- if (topicName.isPartitioned() && !partitionedTopicResources.partitionedTopicExists(topicName)) {
- return CompletableFuture.completedFuture(null);
- }
- CompletableFuture<Void> deleteMetadataFuture = new CompletableFuture<>();
- getBrokerService().fetchPartitionedTopicMetadataAsync(TopicName.get(topicName.getPartitionedTopicName()))
- .thenAccept((metadata -> {
- // make sure all sub partitions were deleted
- for (int i = 0; i < metadata.partitions; i++) {
- if (brokerService.getPulsar().getPulsarResources().getTopicResources()
- .persistentTopicExists(topicName.getPartition(i)).join()) {
- throw new UnsupportedOperationException();
- }
- }
- }))
- .thenAccept((res) -> partitionedTopicResources.deletePartitionedTopicAsync(topicName)
- .thenAccept((r) -> {
- deleteMetadataFuture.complete(null);
- }).exceptionally(ex -> {
- deleteMetadataFuture.completeExceptionally(ex.getCause());
- return null;
- }))
- .exceptionally((e) -> {
- if (!(e.getCause() instanceof UnsupportedOperationException)) {
- log.error("delete metadata fail", e);
- }
- deleteMetadataFuture.complete(null);
- return null;
- });
- return deleteMetadataFuture;
- } catch (Exception e) {
- return FutureUtil.failedFuture(e);
- }
+ PartitionedTopicResources partitionedTopicResources = getBrokerService().pulsar().getPulsarResources()
+ .getNamespaceResources()
+ .getPartitionedTopicResources();
+ return partitionedTopicResources.partitionedTopicExistsAsync(topicName)
+ .thenCompose(partitionedTopicExist -> {
+ if (!partitionedTopicExist) {
+ return CompletableFuture.completedFuture(null);
+ } else {
+ return getBrokerService()
+ .fetchPartitionedTopicMetadataAsync(TopicName.get(topicName.getPartitionedTopicName()))
+ .thenCompose((metadata -> {
+ List<CompletableFuture<Boolean>> persistentTopicExists =
+ new ArrayList<>(metadata.partitions);
+ for (int i = 0; i < metadata.partitions; i++) {
+ persistentTopicExists.add(brokerService.getPulsar()
+ .getPulsarResources().getTopicResources()
+ .persistentTopicExists(topicName.getPartition(i)));
+ }
+ return FutureUtil.waitForAll(persistentTopicExists)
+ .thenCompose(unused -> {
+ // make sure all sub partitions were deleted after all future complete
+ Optional<Boolean> anyExistPartition = persistentTopicExists.stream()
+ .map(CompletableFuture::join)
+ .filter(topicExist -> topicExist)
+ .findAny();
+ if (anyExistPartition.isPresent()) {
+ throw new UnsupportedOperationException(
Review comment:
@eolivelli
Yes, you are right. i will change it ASAP.
--
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 change in pull request #14469: [Broker] Fix ``Future.join()`` causing deadlock.
Posted by GitBox <gi...@apache.org>.
Technoboy- commented on a change in pull request #14469:
URL: https://github.com/apache/pulsar/pull/14469#discussion_r815302639
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java
##########
@@ -2256,44 +2256,52 @@ public void checkGC() {
return CompletableFuture.completedFuture(null);
}
TopicName topicName = TopicName.get(TopicName.get(topic).getPartitionedTopicName());
- try {
- PartitionedTopicResources partitionedTopicResources = getBrokerService().pulsar().getPulsarResources()
- .getNamespaceResources()
- .getPartitionedTopicResources();
- if (topicName.isPartitioned() && !partitionedTopicResources.partitionedTopicExists(topicName)) {
- return CompletableFuture.completedFuture(null);
- }
- CompletableFuture<Void> deleteMetadataFuture = new CompletableFuture<>();
- getBrokerService().fetchPartitionedTopicMetadataAsync(TopicName.get(topicName.getPartitionedTopicName()))
- .thenAccept((metadata -> {
- // make sure all sub partitions were deleted
- for (int i = 0; i < metadata.partitions; i++) {
- if (brokerService.getPulsar().getPulsarResources().getTopicResources()
- .persistentTopicExists(topicName.getPartition(i)).join()) {
- throw new UnsupportedOperationException();
- }
- }
- }))
- .thenAccept((res) -> partitionedTopicResources.deletePartitionedTopicAsync(topicName)
- .thenAccept((r) -> {
- deleteMetadataFuture.complete(null);
- }).exceptionally(ex -> {
- deleteMetadataFuture.completeExceptionally(ex.getCause());
- return null;
- }))
- .exceptionally((e) -> {
- if (!(e.getCause() instanceof UnsupportedOperationException)) {
- log.error("delete metadata fail", e);
+ PartitionedTopicResources partitionedTopicResources = getBrokerService().pulsar().getPulsarResources()
+ .getNamespaceResources()
+ .getPartitionedTopicResources();
+ if (topicName.isPartitioned()) {
+ return partitionedTopicResources.partitionedTopicExistsAsync(topicName)
+ .thenCompose(partitionedTopicExist -> {
+ if (!partitionedTopicExist) {
+ return CompletableFuture.completedFuture(null);
+ } else {
+ return innerDeletePartitionedMetadata(topicName, partitionedTopicResources);
}
- deleteMetadataFuture.complete(null);
- return null;
});
- return deleteMetadataFuture;
- } catch (Exception e) {
- return FutureUtil.failedFuture(e);
+ } else {
+ return innerDeletePartitionedMetadata(topicName, partitionedTopicResources);
}
}
+ private CompletableFuture<Void> innerDeletePartitionedMetadata(TopicName topicName,
+ PartitionedTopicResources partitionedTopicResources)
+ {
+ return getBrokerService()
+ .fetchPartitionedTopicMetadataAsync(TopicName.get(topicName.getPartitionedTopicName()))
+ .thenCompose((metadata -> {
+ List<CompletableFuture<Boolean>> persistentTopicExists = new ArrayList<>(metadata.partitions);
Review comment:
Hi @michaeljmarshall , I think this is ok here. But please let me know if I mis-understand something.
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java
##########
@@ -2256,44 +2256,52 @@ public void checkGC() {
return CompletableFuture.completedFuture(null);
}
TopicName topicName = TopicName.get(TopicName.get(topic).getPartitionedTopicName());
- try {
- PartitionedTopicResources partitionedTopicResources = getBrokerService().pulsar().getPulsarResources()
- .getNamespaceResources()
- .getPartitionedTopicResources();
- if (topicName.isPartitioned() && !partitionedTopicResources.partitionedTopicExists(topicName)) {
- return CompletableFuture.completedFuture(null);
- }
- CompletableFuture<Void> deleteMetadataFuture = new CompletableFuture<>();
- getBrokerService().fetchPartitionedTopicMetadataAsync(TopicName.get(topicName.getPartitionedTopicName()))
- .thenAccept((metadata -> {
- // make sure all sub partitions were deleted
- for (int i = 0; i < metadata.partitions; i++) {
- if (brokerService.getPulsar().getPulsarResources().getTopicResources()
- .persistentTopicExists(topicName.getPartition(i)).join()) {
- throw new UnsupportedOperationException();
- }
- }
- }))
- .thenAccept((res) -> partitionedTopicResources.deletePartitionedTopicAsync(topicName)
- .thenAccept((r) -> {
- deleteMetadataFuture.complete(null);
- }).exceptionally(ex -> {
- deleteMetadataFuture.completeExceptionally(ex.getCause());
- return null;
- }))
- .exceptionally((e) -> {
- if (!(e.getCause() instanceof UnsupportedOperationException)) {
- log.error("delete metadata fail", e);
+ PartitionedTopicResources partitionedTopicResources = getBrokerService().pulsar().getPulsarResources()
+ .getNamespaceResources()
+ .getPartitionedTopicResources();
+ if (topicName.isPartitioned()) {
+ return partitionedTopicResources.partitionedTopicExistsAsync(topicName)
+ .thenCompose(partitionedTopicExist -> {
+ if (!partitionedTopicExist) {
+ return CompletableFuture.completedFuture(null);
+ } else {
+ return innerDeletePartitionedMetadata(topicName, partitionedTopicResources);
}
- deleteMetadataFuture.complete(null);
- return null;
});
- return deleteMetadataFuture;
- } catch (Exception e) {
- return FutureUtil.failedFuture(e);
+ } else {
+ return innerDeletePartitionedMetadata(topicName, partitionedTopicResources);
}
}
+ private CompletableFuture<Void> innerDeletePartitionedMetadata(TopicName topicName,
+ PartitionedTopicResources partitionedTopicResources)
+ {
+ return getBrokerService()
+ .fetchPartitionedTopicMetadataAsync(TopicName.get(topicName.getPartitionedTopicName()))
+ .thenCompose((metadata -> {
+ List<CompletableFuture<Boolean>> persistentTopicExists = new ArrayList<>(metadata.partitions);
Review comment:
Hi @michaeljmarshall , I think this is ok here. But please let me know if I misunderstand something.
--
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] mattisonchao commented on pull request #14469: [Broker] Fix ``Future.join()`` causing deadlock.
Posted by GitBox <gi...@apache.org>.
mattisonchao commented on pull request #14469:
URL: https://github.com/apache/pulsar/pull/14469#issuecomment-1053761387
HI, @michaeljmarshall
PTAL :)
--
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 change in pull request #14469: [Broker] Fix ``Future.join()`` causing deadlock.
Posted by GitBox <gi...@apache.org>.
Technoboy- commented on a change in pull request #14469:
URL: https://github.com/apache/pulsar/pull/14469#discussion_r815303313
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java
##########
@@ -2256,42 +2256,42 @@ public void checkGC() {
return CompletableFuture.completedFuture(null);
}
TopicName topicName = TopicName.get(TopicName.get(topic).getPartitionedTopicName());
- try {
- PartitionedTopicResources partitionedTopicResources = getBrokerService().pulsar().getPulsarResources()
- .getNamespaceResources()
- .getPartitionedTopicResources();
- if (topicName.isPartitioned() && !partitionedTopicResources.partitionedTopicExists(topicName)) {
- return CompletableFuture.completedFuture(null);
- }
- CompletableFuture<Void> deleteMetadataFuture = new CompletableFuture<>();
- getBrokerService().fetchPartitionedTopicMetadataAsync(TopicName.get(topicName.getPartitionedTopicName()))
- .thenAccept((metadata -> {
- // make sure all sub partitions were deleted
- for (int i = 0; i < metadata.partitions; i++) {
- if (brokerService.getPulsar().getPulsarResources().getTopicResources()
- .persistentTopicExists(topicName.getPartition(i)).join()) {
- throw new UnsupportedOperationException();
- }
- }
- }))
- .thenAccept((res) -> partitionedTopicResources.deletePartitionedTopicAsync(topicName)
- .thenAccept((r) -> {
- deleteMetadataFuture.complete(null);
- }).exceptionally(ex -> {
- deleteMetadataFuture.completeExceptionally(ex.getCause());
- return null;
- }))
- .exceptionally((e) -> {
- if (!(e.getCause() instanceof UnsupportedOperationException)) {
- log.error("delete metadata fail", e);
- }
- deleteMetadataFuture.complete(null);
- return null;
- });
- return deleteMetadataFuture;
- } catch (Exception e) {
- return FutureUtil.failedFuture(e);
- }
+ PartitionedTopicResources partitionedTopicResources = getBrokerService().pulsar().getPulsarResources()
+ .getNamespaceResources()
+ .getPartitionedTopicResources();
+ return partitionedTopicResources.partitionedTopicExistsAsync(topicName)
+ .thenCompose(partitionedTopicExist -> {
+ if (!partitionedTopicExist) {
+ return CompletableFuture.completedFuture(null);
+ } else {
+ return getBrokerService()
+ .fetchPartitionedTopicMetadataAsync(TopicName.get(topicName.getPartitionedTopicName()))
+ .thenCompose((metadata -> {
+ List<CompletableFuture<Boolean>> persistentTopicExists =
+ new ArrayList<>(metadata.partitions);
+ for (int i = 0; i < metadata.partitions; i++) {
+ persistentTopicExists.add(brokerService.getPulsar()
+ .getPulsarResources().getTopicResources()
+ .persistentTopicExists(topicName.getPartition(i)));
+ }
+ return FutureUtil.waitForAll(persistentTopicExists)
+ .thenCompose(unused -> {
+ // make sure all sub partitions were deleted after all future complete
+ Optional<Boolean> anyExistPartition = persistentTopicExists.stream()
+ .map(CompletableFuture::join)
+ .filter(topicExist -> topicExist)
+ .findAny();
+ if (anyExistPartition.isPresent()) {
+ log.error("[{}] Delete metadata fail", topic);
+ return CompletableFuture.completedFuture(null);
Review comment:
Hi, @mattisonchao , another concern is: if delete metadata error, we should print log. but in this case, how do users delete these dirty datas? Could we delete via `pulsar-admin topics delete xxxx`? Please help test. If that doesn't work, could we add another new cmd to support this?
cc @codelipenghui
--
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] mattisonchao commented on pull request #14469: [Broker] Fix ``Future.join()`` causing deadlock.
Posted by GitBox <gi...@apache.org>.
mattisonchao commented on pull request #14469:
URL: https://github.com/apache/pulsar/pull/14469#issuecomment-1051466909
/pulsarbot rerun-failure-checks
--
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] michaeljmarshall commented on a change in pull request #14469: [Broker] Fix ``Future.join()`` causing deadlock.
Posted by GitBox <gi...@apache.org>.
michaeljmarshall commented on a change in pull request #14469:
URL: https://github.com/apache/pulsar/pull/14469#discussion_r815276905
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java
##########
@@ -2256,44 +2256,52 @@ public void checkGC() {
return CompletableFuture.completedFuture(null);
}
TopicName topicName = TopicName.get(TopicName.get(topic).getPartitionedTopicName());
- try {
- PartitionedTopicResources partitionedTopicResources = getBrokerService().pulsar().getPulsarResources()
- .getNamespaceResources()
- .getPartitionedTopicResources();
- if (topicName.isPartitioned() && !partitionedTopicResources.partitionedTopicExists(topicName)) {
- return CompletableFuture.completedFuture(null);
- }
- CompletableFuture<Void> deleteMetadataFuture = new CompletableFuture<>();
- getBrokerService().fetchPartitionedTopicMetadataAsync(TopicName.get(topicName.getPartitionedTopicName()))
- .thenAccept((metadata -> {
- // make sure all sub partitions were deleted
- for (int i = 0; i < metadata.partitions; i++) {
- if (brokerService.getPulsar().getPulsarResources().getTopicResources()
- .persistentTopicExists(topicName.getPartition(i)).join()) {
- throw new UnsupportedOperationException();
- }
- }
- }))
- .thenAccept((res) -> partitionedTopicResources.deletePartitionedTopicAsync(topicName)
- .thenAccept((r) -> {
- deleteMetadataFuture.complete(null);
- }).exceptionally(ex -> {
- deleteMetadataFuture.completeExceptionally(ex.getCause());
- return null;
- }))
- .exceptionally((e) -> {
- if (!(e.getCause() instanceof UnsupportedOperationException)) {
- log.error("delete metadata fail", e);
+ PartitionedTopicResources partitionedTopicResources = getBrokerService().pulsar().getPulsarResources()
+ .getNamespaceResources()
+ .getPartitionedTopicResources();
+ if (topicName.isPartitioned()) {
+ return partitionedTopicResources.partitionedTopicExistsAsync(topicName)
+ .thenCompose(partitionedTopicExist -> {
+ if (!partitionedTopicExist) {
+ return CompletableFuture.completedFuture(null);
+ } else {
+ return innerDeletePartitionedMetadata(topicName, partitionedTopicResources);
}
- deleteMetadataFuture.complete(null);
- return null;
});
- return deleteMetadataFuture;
- } catch (Exception e) {
- return FutureUtil.failedFuture(e);
+ } else {
+ return innerDeletePartitionedMetadata(topicName, partitionedTopicResources);
}
}
+ private CompletableFuture<Void> innerDeletePartitionedMetadata(TopicName topicName,
+ PartitionedTopicResources partitionedTopicResources)
+ {
+ return getBrokerService()
+ .fetchPartitionedTopicMetadataAsync(TopicName.get(topicName.getPartitionedTopicName()))
+ .thenCompose((metadata -> {
+ List<CompletableFuture<Boolean>> persistentTopicExists = new ArrayList<>(metadata.partitions);
Review comment:
Thanks for your question. My understanding is that `thenCompose` is run on which ever executor completes the `FutureUtil.waitForAll` future, otherwise, `FutureUtil.waitForAll` would be a blocking call. Therefore, line 2280 is potentially on another thread. Let me know what you think.
--
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] mattisonchao commented on a change in pull request #14469: [Broker] Fix ``Future.join()`` causing deadlock.
Posted by GitBox <gi...@apache.org>.
mattisonchao commented on a change in pull request #14469:
URL: https://github.com/apache/pulsar/pull/14469#discussion_r815241595
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java
##########
@@ -2256,44 +2256,52 @@ public void checkGC() {
return CompletableFuture.completedFuture(null);
}
TopicName topicName = TopicName.get(TopicName.get(topic).getPartitionedTopicName());
- try {
- PartitionedTopicResources partitionedTopicResources = getBrokerService().pulsar().getPulsarResources()
- .getNamespaceResources()
- .getPartitionedTopicResources();
- if (topicName.isPartitioned() && !partitionedTopicResources.partitionedTopicExists(topicName)) {
- return CompletableFuture.completedFuture(null);
- }
- CompletableFuture<Void> deleteMetadataFuture = new CompletableFuture<>();
- getBrokerService().fetchPartitionedTopicMetadataAsync(TopicName.get(topicName.getPartitionedTopicName()))
- .thenAccept((metadata -> {
- // make sure all sub partitions were deleted
- for (int i = 0; i < metadata.partitions; i++) {
- if (brokerService.getPulsar().getPulsarResources().getTopicResources()
- .persistentTopicExists(topicName.getPartition(i)).join()) {
- throw new UnsupportedOperationException();
- }
- }
- }))
- .thenAccept((res) -> partitionedTopicResources.deletePartitionedTopicAsync(topicName)
- .thenAccept((r) -> {
- deleteMetadataFuture.complete(null);
- }).exceptionally(ex -> {
- deleteMetadataFuture.completeExceptionally(ex.getCause());
- return null;
- }))
- .exceptionally((e) -> {
- if (!(e.getCause() instanceof UnsupportedOperationException)) {
- log.error("delete metadata fail", e);
+ PartitionedTopicResources partitionedTopicResources = getBrokerService().pulsar().getPulsarResources()
+ .getNamespaceResources()
+ .getPartitionedTopicResources();
+ if (topicName.isPartitioned()) {
Review comment:
Fixed
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java
##########
@@ -2256,44 +2256,52 @@ public void checkGC() {
return CompletableFuture.completedFuture(null);
}
TopicName topicName = TopicName.get(TopicName.get(topic).getPartitionedTopicName());
- try {
- PartitionedTopicResources partitionedTopicResources = getBrokerService().pulsar().getPulsarResources()
- .getNamespaceResources()
- .getPartitionedTopicResources();
- if (topicName.isPartitioned() && !partitionedTopicResources.partitionedTopicExists(topicName)) {
- return CompletableFuture.completedFuture(null);
- }
- CompletableFuture<Void> deleteMetadataFuture = new CompletableFuture<>();
- getBrokerService().fetchPartitionedTopicMetadataAsync(TopicName.get(topicName.getPartitionedTopicName()))
- .thenAccept((metadata -> {
- // make sure all sub partitions were deleted
- for (int i = 0; i < metadata.partitions; i++) {
- if (brokerService.getPulsar().getPulsarResources().getTopicResources()
- .persistentTopicExists(topicName.getPartition(i)).join()) {
- throw new UnsupportedOperationException();
- }
- }
- }))
- .thenAccept((res) -> partitionedTopicResources.deletePartitionedTopicAsync(topicName)
- .thenAccept((r) -> {
- deleteMetadataFuture.complete(null);
- }).exceptionally(ex -> {
- deleteMetadataFuture.completeExceptionally(ex.getCause());
- return null;
- }))
- .exceptionally((e) -> {
- if (!(e.getCause() instanceof UnsupportedOperationException)) {
- log.error("delete metadata fail", e);
+ PartitionedTopicResources partitionedTopicResources = getBrokerService().pulsar().getPulsarResources()
+ .getNamespaceResources()
+ .getPartitionedTopicResources();
+ if (topicName.isPartitioned()) {
+ return partitionedTopicResources.partitionedTopicExistsAsync(topicName)
+ .thenCompose(partitionedTopicExist -> {
+ if (!partitionedTopicExist) {
+ return CompletableFuture.completedFuture(null);
+ } else {
+ return innerDeletePartitionedMetadata(topicName, partitionedTopicResources);
}
- deleteMetadataFuture.complete(null);
- return null;
});
- return deleteMetadataFuture;
- } catch (Exception e) {
- return FutureUtil.failedFuture(e);
+ } else {
+ return innerDeletePartitionedMetadata(topicName, partitionedTopicResources);
}
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
[GitHub] [pulsar] mattisonchao commented on a change in pull request #14469: [Broker] Fix ``Future.join()`` causing deadlock.
Posted by GitBox <gi...@apache.org>.
mattisonchao commented on a change in pull request #14469:
URL: https://github.com/apache/pulsar/pull/14469#discussion_r814594667
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java
##########
@@ -2256,42 +2256,42 @@ public void checkGC() {
return CompletableFuture.completedFuture(null);
}
TopicName topicName = TopicName.get(TopicName.get(topic).getPartitionedTopicName());
- try {
- PartitionedTopicResources partitionedTopicResources = getBrokerService().pulsar().getPulsarResources()
- .getNamespaceResources()
- .getPartitionedTopicResources();
- if (topicName.isPartitioned() && !partitionedTopicResources.partitionedTopicExists(topicName)) {
- return CompletableFuture.completedFuture(null);
- }
- CompletableFuture<Void> deleteMetadataFuture = new CompletableFuture<>();
- getBrokerService().fetchPartitionedTopicMetadataAsync(TopicName.get(topicName.getPartitionedTopicName()))
- .thenAccept((metadata -> {
- // make sure all sub partitions were deleted
- for (int i = 0; i < metadata.partitions; i++) {
- if (brokerService.getPulsar().getPulsarResources().getTopicResources()
- .persistentTopicExists(topicName.getPartition(i)).join()) {
- throw new UnsupportedOperationException();
- }
- }
- }))
- .thenAccept((res) -> partitionedTopicResources.deletePartitionedTopicAsync(topicName)
- .thenAccept((r) -> {
- deleteMetadataFuture.complete(null);
- }).exceptionally(ex -> {
- deleteMetadataFuture.completeExceptionally(ex.getCause());
- return null;
- }))
- .exceptionally((e) -> {
- if (!(e.getCause() instanceof UnsupportedOperationException)) {
- log.error("delete metadata fail", e);
- }
- deleteMetadataFuture.complete(null);
- return null;
- });
- return deleteMetadataFuture;
- } catch (Exception e) {
- return FutureUtil.failedFuture(e);
+ PartitionedTopicResources partitionedTopicResources = getBrokerService().pulsar().getPulsarResources()
+ .getNamespaceResources()
+ .getPartitionedTopicResources();
+ if (!topicName.isPartitioned()) {
Review comment:
Fixed. Thanks for your review.
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java
##########
@@ -2256,42 +2256,42 @@ public void checkGC() {
return CompletableFuture.completedFuture(null);
}
TopicName topicName = TopicName.get(TopicName.get(topic).getPartitionedTopicName());
- try {
- PartitionedTopicResources partitionedTopicResources = getBrokerService().pulsar().getPulsarResources()
- .getNamespaceResources()
- .getPartitionedTopicResources();
- if (topicName.isPartitioned() && !partitionedTopicResources.partitionedTopicExists(topicName)) {
- return CompletableFuture.completedFuture(null);
- }
- CompletableFuture<Void> deleteMetadataFuture = new CompletableFuture<>();
- getBrokerService().fetchPartitionedTopicMetadataAsync(TopicName.get(topicName.getPartitionedTopicName()))
- .thenAccept((metadata -> {
- // make sure all sub partitions were deleted
- for (int i = 0; i < metadata.partitions; i++) {
- if (brokerService.getPulsar().getPulsarResources().getTopicResources()
- .persistentTopicExists(topicName.getPartition(i)).join()) {
- throw new UnsupportedOperationException();
- }
- }
- }))
- .thenAccept((res) -> partitionedTopicResources.deletePartitionedTopicAsync(topicName)
- .thenAccept((r) -> {
- deleteMetadataFuture.complete(null);
- }).exceptionally(ex -> {
- deleteMetadataFuture.completeExceptionally(ex.getCause());
- return null;
- }))
- .exceptionally((e) -> {
- if (!(e.getCause() instanceof UnsupportedOperationException)) {
- log.error("delete metadata fail", e);
- }
- deleteMetadataFuture.complete(null);
- return null;
- });
- return deleteMetadataFuture;
- } catch (Exception e) {
- return FutureUtil.failedFuture(e);
+ PartitionedTopicResources partitionedTopicResources = getBrokerService().pulsar().getPulsarResources()
+ .getNamespaceResources()
+ .getPartitionedTopicResources();
+ if (!topicName.isPartitioned()) {
+ return CompletableFuture.completedFuture(null);
}
+ return partitionedTopicResources.partitionedTopicExistsAsync(topicName)
+ .thenCompose(partitionedTopicExist -> {
+ if (!partitionedTopicExist) {
+ return CompletableFuture.completedFuture(null);
+ }
+ return getBrokerService()
+ .fetchPartitionedTopicMetadataAsync(TopicName.get(topicName.getPartitionedTopicName()))
+ .thenCompose((metadata -> {
+ List<CompletableFuture<Boolean>> persistentTopicExists =
+ new ArrayList<>(metadata.partitions);
+ for (int i = 0; i < metadata.partitions; i++) {
+ persistentTopicExists.add(brokerService.getPulsar()
+ .getPulsarResources().getTopicResources()
+ .persistentTopicExists(topicName.getPartition(i)));
+ }
+ return FutureUtil.waitForAll(persistentTopicExists)
+ .thenAccept(__ -> {
+ // make sure all sub partitions were deleted after all future complete
+ persistentTopicExists.forEach(future -> {
+ boolean topicExist = future.join();
+ if (topicExist) {
+ UnsupportedOperationException ex =
+ new UnsupportedOperationException();
+ log.error("[{}] Delete metadata fail", topic, ex);
Review comment:
Fixed. Thanks for your review.
--
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 change in pull request #14469: [Broker] Fix ``Future.join()`` causing deadlock.
Posted by GitBox <gi...@apache.org>.
Technoboy- commented on a change in pull request #14469:
URL: https://github.com/apache/pulsar/pull/14469#discussion_r815561844
##########
File path: pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PersistentTopicE2ETest.java
##########
@@ -97,6 +97,7 @@
@Override
protected void setup() throws Exception {
super.baseSetup();
+ conf.setBrokerDeleteInactivePartitionedTopicMetadataEnabled(true);
Review comment:
Add before `super.baseSetup();`
--
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] mattisonchao commented on a change in pull request #14469: [Broker] Fix ``Future.join()`` causing deadlock.
Posted by GitBox <gi...@apache.org>.
mattisonchao commented on a change in pull request #14469:
URL: https://github.com/apache/pulsar/pull/14469#discussion_r815312084
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java
##########
@@ -2256,42 +2256,42 @@ public void checkGC() {
return CompletableFuture.completedFuture(null);
}
TopicName topicName = TopicName.get(TopicName.get(topic).getPartitionedTopicName());
- try {
- PartitionedTopicResources partitionedTopicResources = getBrokerService().pulsar().getPulsarResources()
- .getNamespaceResources()
- .getPartitionedTopicResources();
- if (topicName.isPartitioned() && !partitionedTopicResources.partitionedTopicExists(topicName)) {
- return CompletableFuture.completedFuture(null);
- }
- CompletableFuture<Void> deleteMetadataFuture = new CompletableFuture<>();
- getBrokerService().fetchPartitionedTopicMetadataAsync(TopicName.get(topicName.getPartitionedTopicName()))
- .thenAccept((metadata -> {
- // make sure all sub partitions were deleted
- for (int i = 0; i < metadata.partitions; i++) {
- if (brokerService.getPulsar().getPulsarResources().getTopicResources()
- .persistentTopicExists(topicName.getPartition(i)).join()) {
- throw new UnsupportedOperationException();
- }
- }
- }))
- .thenAccept((res) -> partitionedTopicResources.deletePartitionedTopicAsync(topicName)
- .thenAccept((r) -> {
- deleteMetadataFuture.complete(null);
- }).exceptionally(ex -> {
- deleteMetadataFuture.completeExceptionally(ex.getCause());
- return null;
- }))
- .exceptionally((e) -> {
- if (!(e.getCause() instanceof UnsupportedOperationException)) {
- log.error("delete metadata fail", e);
- }
- deleteMetadataFuture.complete(null);
- return null;
- });
- return deleteMetadataFuture;
- } catch (Exception e) {
- return FutureUtil.failedFuture(e);
- }
+ PartitionedTopicResources partitionedTopicResources = getBrokerService().pulsar().getPulsarResources()
+ .getNamespaceResources()
+ .getPartitionedTopicResources();
+ return partitionedTopicResources.partitionedTopicExistsAsync(topicName)
+ .thenCompose(partitionedTopicExist -> {
+ if (!partitionedTopicExist) {
+ return CompletableFuture.completedFuture(null);
+ } else {
+ return getBrokerService()
+ .fetchPartitionedTopicMetadataAsync(TopicName.get(topicName.getPartitionedTopicName()))
+ .thenCompose((metadata -> {
+ List<CompletableFuture<Boolean>> persistentTopicExists =
+ new ArrayList<>(metadata.partitions);
+ for (int i = 0; i < metadata.partitions; i++) {
+ persistentTopicExists.add(brokerService.getPulsar()
+ .getPulsarResources().getTopicResources()
+ .persistentTopicExists(topicName.getPartition(i)));
+ }
+ return FutureUtil.waitForAll(persistentTopicExists)
+ .thenCompose(unused -> {
+ // make sure all sub partitions were deleted after all future complete
+ Optional<Boolean> anyExistPartition = persistentTopicExists.stream()
+ .map(CompletableFuture::join)
+ .filter(topicExist -> topicExist)
+ .findAny();
+ if (anyExistPartition.isPresent()) {
+ log.error("[{}] Delete metadata fail", topic);
+ return CompletableFuture.completedFuture(null);
Review comment:
@Technoboy
Yes, that's exactly what I was concerned with before, but before we do this work, I think we need to check if we have other logic to clean it up.
@codelipenghui @eolivelli @michaeljmarshall I'm not sure about that, I will do the test and give it feedback tomorrow, please feel free to give us more suggestions. very thanks~
--
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] mattisonchao removed a comment on pull request #14469: [Broker] Fix ``Future.join()`` causing deadlock.
Posted by GitBox <gi...@apache.org>.
mattisonchao removed a comment on pull request #14469:
URL: https://github.com/apache/pulsar/pull/14469#issuecomment-1054814133
@michaeljmarshall
use ``unmodifiableList``looks good for me. but it is just a preventive measure, right?
--
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] mattisonchao commented on a change in pull request #14469: [Broker] Fix ``Future.join()`` causing deadlock.
Posted by GitBox <gi...@apache.org>.
mattisonchao commented on a change in pull request #14469:
URL: https://github.com/apache/pulsar/pull/14469#discussion_r815406729
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java
##########
@@ -2256,42 +2256,43 @@ public void checkGC() {
return CompletableFuture.completedFuture(null);
}
TopicName topicName = TopicName.get(TopicName.get(topic).getPartitionedTopicName());
- try {
- PartitionedTopicResources partitionedTopicResources = getBrokerService().pulsar().getPulsarResources()
- .getNamespaceResources()
- .getPartitionedTopicResources();
- if (topicName.isPartitioned() && !partitionedTopicResources.partitionedTopicExists(topicName)) {
- return CompletableFuture.completedFuture(null);
- }
- CompletableFuture<Void> deleteMetadataFuture = new CompletableFuture<>();
- getBrokerService().fetchPartitionedTopicMetadataAsync(TopicName.get(topicName.getPartitionedTopicName()))
- .thenAccept((metadata -> {
- // make sure all sub partitions were deleted
- for (int i = 0; i < metadata.partitions; i++) {
- if (brokerService.getPulsar().getPulsarResources().getTopicResources()
- .persistentTopicExists(topicName.getPartition(i)).join()) {
- throw new UnsupportedOperationException();
- }
- }
- }))
- .thenAccept((res) -> partitionedTopicResources.deletePartitionedTopicAsync(topicName)
- .thenAccept((r) -> {
- deleteMetadataFuture.complete(null);
- }).exceptionally(ex -> {
- deleteMetadataFuture.completeExceptionally(ex.getCause());
- return null;
- }))
- .exceptionally((e) -> {
- if (!(e.getCause() instanceof UnsupportedOperationException)) {
- log.error("delete metadata fail", e);
- }
- deleteMetadataFuture.complete(null);
- return null;
- });
- return deleteMetadataFuture;
- } catch (Exception e) {
- return FutureUtil.failedFuture(e);
- }
+ PartitionedTopicResources partitionedTopicResources = getBrokerService().pulsar().getPulsarResources()
+ .getNamespaceResources()
+ .getPartitionedTopicResources();
+ return partitionedTopicResources.partitionedTopicExistsAsync(topicName)
+ .thenCompose(partitionedTopicExist -> {
+ if (!partitionedTopicExist) {
+ return CompletableFuture.completedFuture(null);
+ } else {
+ return getBrokerService()
+ .fetchPartitionedTopicMetadataAsync(TopicName.get(topicName.getPartitionedTopicName()))
+ .thenCompose((metadata -> {
+ List<CompletableFuture<Boolean>> persistentTopicExists =
+ new ArrayList<>(metadata.partitions);
+ for (int i = 0; i < metadata.partitions; i++) {
+ persistentTopicExists.add(brokerService.getPulsar()
+ .getPulsarResources().getTopicResources()
+ .persistentTopicExists(topicName.getPartition(i)));
+ }
+ return FutureUtil.waitForAll(persistentTopicExists)
+ .thenCompose(unused -> {
+ // make sure all sub partitions were deleted after all future complete
+ Optional<Boolean> anyExistPartition = persistentTopicExists.stream()
+ .map(CompletableFuture::join)
+ .filter(topicExist -> topicExist)
+ .findAny();
+ if (anyExistPartition.isPresent()) {
+ throw new UnsupportedOperationException(
Review comment:
Hi, @Jason918
The ``PersistentTopic#checkGC`` method already handles this error.
--
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] mattisonchao commented on pull request #14469: [Broker] Fix ``Future.join()`` causing deadlock.
Posted by GitBox <gi...@apache.org>.
mattisonchao commented on pull request #14469:
URL: https://github.com/apache/pulsar/pull/14469#issuecomment-1053937387
/pulsarbot rerun-failure-checks
--
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] mattisonchao commented on pull request #14469: [Broker] Fix ``Future.join()`` causing deadlock.
Posted by GitBox <gi...@apache.org>.
mattisonchao commented on pull request #14469:
URL: https://github.com/apache/pulsar/pull/14469#issuecomment-1054814133
@michaeljmarshall
use ``unmodifiableList``looks good for me. but it is just a preventive measure, right?
--
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] codelipenghui merged pull request #14469: [Broker] Fix ``Future.join()`` causing deadlock.
Posted by GitBox <gi...@apache.org>.
codelipenghui merged pull request #14469:
URL: https://github.com/apache/pulsar/pull/14469
--
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] mattisonchao edited a comment on pull request #14469: [Broker] Fix ``Future.join()`` causing deadlock.
Posted by GitBox <gi...@apache.org>.
mattisonchao edited a comment on pull request #14469:
URL: https://github.com/apache/pulsar/pull/14469#issuecomment-1053761387
Hi, @michaeljmarshall
PTAL :)
--
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] michaeljmarshall commented on pull request #14469: [Broker] Fix ``Future.join()`` causing deadlock.
Posted by GitBox <gi...@apache.org>.
michaeljmarshall commented on pull request #14469:
URL: https://github.com/apache/pulsar/pull/14469#issuecomment-1055041262
Just in case anyone is looking for more context, one of my concerns with the initial implementation had to do with unsafe publication of mutable state, which can behave in unexpected ways. I explain this more in this comment https://github.com/apache/pulsar/pull/14469#discussion_r816083093
--
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] codelipenghui commented on a change in pull request #14469: [Broker] Fix ``Future.join()`` causing deadlock.
Posted by GitBox <gi...@apache.org>.
codelipenghui commented on a change in pull request #14469:
URL: https://github.com/apache/pulsar/pull/14469#discussion_r814567591
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java
##########
@@ -2256,42 +2256,42 @@ public void checkGC() {
return CompletableFuture.completedFuture(null);
}
TopicName topicName = TopicName.get(TopicName.get(topic).getPartitionedTopicName());
- try {
- PartitionedTopicResources partitionedTopicResources = getBrokerService().pulsar().getPulsarResources()
- .getNamespaceResources()
- .getPartitionedTopicResources();
- if (topicName.isPartitioned() && !partitionedTopicResources.partitionedTopicExists(topicName)) {
- return CompletableFuture.completedFuture(null);
- }
- CompletableFuture<Void> deleteMetadataFuture = new CompletableFuture<>();
- getBrokerService().fetchPartitionedTopicMetadataAsync(TopicName.get(topicName.getPartitionedTopicName()))
- .thenAccept((metadata -> {
- // make sure all sub partitions were deleted
- for (int i = 0; i < metadata.partitions; i++) {
- if (brokerService.getPulsar().getPulsarResources().getTopicResources()
- .persistentTopicExists(topicName.getPartition(i)).join()) {
- throw new UnsupportedOperationException();
- }
- }
- }))
- .thenAccept((res) -> partitionedTopicResources.deletePartitionedTopicAsync(topicName)
- .thenAccept((r) -> {
- deleteMetadataFuture.complete(null);
- }).exceptionally(ex -> {
- deleteMetadataFuture.completeExceptionally(ex.getCause());
- return null;
- }))
- .exceptionally((e) -> {
- if (!(e.getCause() instanceof UnsupportedOperationException)) {
- log.error("delete metadata fail", e);
- }
- deleteMetadataFuture.complete(null);
- return null;
- });
- return deleteMetadataFuture;
- } catch (Exception e) {
- return FutureUtil.failedFuture(e);
+ PartitionedTopicResources partitionedTopicResources = getBrokerService().pulsar().getPulsarResources()
+ .getNamespaceResources()
+ .getPartitionedTopicResources();
+ if (!topicName.isPartitioned()) {
+ return CompletableFuture.completedFuture(null);
}
+ return partitionedTopicResources.partitionedTopicExistsAsync(topicName)
+ .thenCompose(partitionedTopicExist -> {
+ if (!partitionedTopicExist) {
+ return CompletableFuture.completedFuture(null);
+ }
+ return getBrokerService()
+ .fetchPartitionedTopicMetadataAsync(TopicName.get(topicName.getPartitionedTopicName()))
+ .thenCompose((metadata -> {
+ List<CompletableFuture<Boolean>> persistentTopicExists =
+ new ArrayList<>(metadata.partitions);
+ for (int i = 0; i < metadata.partitions; i++) {
+ persistentTopicExists.add(brokerService.getPulsar()
+ .getPulsarResources().getTopicResources()
+ .persistentTopicExists(topicName.getPartition(i)));
+ }
+ return FutureUtil.waitForAll(persistentTopicExists)
+ .thenAccept(__ -> {
+ // make sure all sub partitions were deleted after all future complete
+ persistentTopicExists.forEach(future -> {
+ boolean topicExist = future.join();
+ if (topicExist) {
+ UnsupportedOperationException ex =
+ new UnsupportedOperationException();
+ log.error("[{}] Delete metadata fail", topic, ex);
Review comment:
Don't need to print the log multiple times,
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java
##########
@@ -2256,42 +2256,42 @@ public void checkGC() {
return CompletableFuture.completedFuture(null);
}
TopicName topicName = TopicName.get(TopicName.get(topic).getPartitionedTopicName());
- try {
- PartitionedTopicResources partitionedTopicResources = getBrokerService().pulsar().getPulsarResources()
- .getNamespaceResources()
- .getPartitionedTopicResources();
- if (topicName.isPartitioned() && !partitionedTopicResources.partitionedTopicExists(topicName)) {
- return CompletableFuture.completedFuture(null);
- }
- CompletableFuture<Void> deleteMetadataFuture = new CompletableFuture<>();
- getBrokerService().fetchPartitionedTopicMetadataAsync(TopicName.get(topicName.getPartitionedTopicName()))
- .thenAccept((metadata -> {
- // make sure all sub partitions were deleted
- for (int i = 0; i < metadata.partitions; i++) {
- if (brokerService.getPulsar().getPulsarResources().getTopicResources()
- .persistentTopicExists(topicName.getPartition(i)).join()) {
- throw new UnsupportedOperationException();
- }
- }
- }))
- .thenAccept((res) -> partitionedTopicResources.deletePartitionedTopicAsync(topicName)
- .thenAccept((r) -> {
- deleteMetadataFuture.complete(null);
- }).exceptionally(ex -> {
- deleteMetadataFuture.completeExceptionally(ex.getCause());
- return null;
- }))
- .exceptionally((e) -> {
- if (!(e.getCause() instanceof UnsupportedOperationException)) {
- log.error("delete metadata fail", e);
- }
- deleteMetadataFuture.complete(null);
- return null;
- });
- return deleteMetadataFuture;
- } catch (Exception e) {
- return FutureUtil.failedFuture(e);
+ PartitionedTopicResources partitionedTopicResources = getBrokerService().pulsar().getPulsarResources()
+ .getNamespaceResources()
+ .getPartitionedTopicResources();
+ if (!topicName.isPartitioned()) {
Review comment:
```suggestion
if (topicName.isPartitioned()) {
```
--
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] mattisonchao commented on a change in pull request #14469: [Broker] Fix ``Future.join()`` causing deadlock.
Posted by GitBox <gi...@apache.org>.
mattisonchao commented on a change in pull request #14469:
URL: https://github.com/apache/pulsar/pull/14469#discussion_r815248232
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java
##########
@@ -2256,44 +2256,52 @@ public void checkGC() {
return CompletableFuture.completedFuture(null);
}
TopicName topicName = TopicName.get(TopicName.get(topic).getPartitionedTopicName());
- try {
- PartitionedTopicResources partitionedTopicResources = getBrokerService().pulsar().getPulsarResources()
- .getNamespaceResources()
- .getPartitionedTopicResources();
- if (topicName.isPartitioned() && !partitionedTopicResources.partitionedTopicExists(topicName)) {
- return CompletableFuture.completedFuture(null);
- }
- CompletableFuture<Void> deleteMetadataFuture = new CompletableFuture<>();
- getBrokerService().fetchPartitionedTopicMetadataAsync(TopicName.get(topicName.getPartitionedTopicName()))
- .thenAccept((metadata -> {
- // make sure all sub partitions were deleted
- for (int i = 0; i < metadata.partitions; i++) {
- if (brokerService.getPulsar().getPulsarResources().getTopicResources()
- .persistentTopicExists(topicName.getPartition(i)).join()) {
- throw new UnsupportedOperationException();
- }
- }
- }))
- .thenAccept((res) -> partitionedTopicResources.deletePartitionedTopicAsync(topicName)
- .thenAccept((r) -> {
- deleteMetadataFuture.complete(null);
- }).exceptionally(ex -> {
- deleteMetadataFuture.completeExceptionally(ex.getCause());
- return null;
- }))
- .exceptionally((e) -> {
- if (!(e.getCause() instanceof UnsupportedOperationException)) {
- log.error("delete metadata fail", e);
+ PartitionedTopicResources partitionedTopicResources = getBrokerService().pulsar().getPulsarResources()
+ .getNamespaceResources()
+ .getPartitionedTopicResources();
+ if (topicName.isPartitioned()) {
+ return partitionedTopicResources.partitionedTopicExistsAsync(topicName)
+ .thenCompose(partitionedTopicExist -> {
+ if (!partitionedTopicExist) {
+ return CompletableFuture.completedFuture(null);
+ } else {
+ return innerDeletePartitionedMetadata(topicName, partitionedTopicResources);
}
- deleteMetadataFuture.complete(null);
- return null;
});
- return deleteMetadataFuture;
- } catch (Exception e) {
- return FutureUtil.failedFuture(e);
+ } else {
+ return innerDeletePartitionedMetadata(topicName, partitionedTopicResources);
}
}
+ private CompletableFuture<Void> innerDeletePartitionedMetadata(TopicName topicName,
+ PartitionedTopicResources partitionedTopicResources)
+ {
+ return getBrokerService()
+ .fetchPartitionedTopicMetadataAsync(TopicName.get(topicName.getPartitionedTopicName()))
+ .thenCompose((metadata -> {
+ List<CompletableFuture<Boolean>> persistentTopicExists = new ArrayList<>(metadata.partitions);
+ for (int i = 0; i < metadata.partitions; i++) {
+ persistentTopicExists.add(brokerService.getPulsar()
+ .getPulsarResources().getTopicResources()
+ .persistentTopicExists(topicName.getPartition(i)));
+ }
+ return FutureUtil.waitForAll(persistentTopicExists)
+ .thenCompose(unused -> {
+ // make sure all sub partitions were deleted after all future complete
+ Optional<Boolean> anyExistPartition = persistentTopicExists.stream()
+ .map(CompletableFuture::join)
+ .filter(topicExist -> topicExist)
+ .findAny();
+ if (anyExistPartition.isPresent()) {
+ log.error("[{}] Delete metadata fail", topic);
Review comment:
The logic here is to re-check to ensure the last partition deletes metadata. If the current partition is not the last alive partition, we need to return and do nothing.
I'm not sure if I'm explaining it clearly, please let me know what you guys think.
--
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 a change in pull request #14469: [Broker] Fix ``Future.join()`` causing deadlock.
Posted by GitBox <gi...@apache.org>.
eolivelli commented on a change in pull request #14469:
URL: https://github.com/apache/pulsar/pull/14469#discussion_r815401861
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java
##########
@@ -2256,42 +2256,43 @@ public void checkGC() {
return CompletableFuture.completedFuture(null);
}
TopicName topicName = TopicName.get(TopicName.get(topic).getPartitionedTopicName());
- try {
- PartitionedTopicResources partitionedTopicResources = getBrokerService().pulsar().getPulsarResources()
- .getNamespaceResources()
- .getPartitionedTopicResources();
- if (topicName.isPartitioned() && !partitionedTopicResources.partitionedTopicExists(topicName)) {
- return CompletableFuture.completedFuture(null);
- }
- CompletableFuture<Void> deleteMetadataFuture = new CompletableFuture<>();
- getBrokerService().fetchPartitionedTopicMetadataAsync(TopicName.get(topicName.getPartitionedTopicName()))
- .thenAccept((metadata -> {
- // make sure all sub partitions were deleted
- for (int i = 0; i < metadata.partitions; i++) {
- if (brokerService.getPulsar().getPulsarResources().getTopicResources()
- .persistentTopicExists(topicName.getPartition(i)).join()) {
- throw new UnsupportedOperationException();
- }
- }
- }))
- .thenAccept((res) -> partitionedTopicResources.deletePartitionedTopicAsync(topicName)
- .thenAccept((r) -> {
- deleteMetadataFuture.complete(null);
- }).exceptionally(ex -> {
- deleteMetadataFuture.completeExceptionally(ex.getCause());
- return null;
- }))
- .exceptionally((e) -> {
- if (!(e.getCause() instanceof UnsupportedOperationException)) {
- log.error("delete metadata fail", e);
- }
- deleteMetadataFuture.complete(null);
- return null;
- });
- return deleteMetadataFuture;
- } catch (Exception e) {
- return FutureUtil.failedFuture(e);
- }
+ PartitionedTopicResources partitionedTopicResources = getBrokerService().pulsar().getPulsarResources()
+ .getNamespaceResources()
+ .getPartitionedTopicResources();
+ return partitionedTopicResources.partitionedTopicExistsAsync(topicName)
+ .thenCompose(partitionedTopicExist -> {
+ if (!partitionedTopicExist) {
+ return CompletableFuture.completedFuture(null);
+ } else {
+ return getBrokerService()
+ .fetchPartitionedTopicMetadataAsync(TopicName.get(topicName.getPartitionedTopicName()))
+ .thenCompose((metadata -> {
+ List<CompletableFuture<Boolean>> persistentTopicExists =
+ new ArrayList<>(metadata.partitions);
+ for (int i = 0; i < metadata.partitions; i++) {
+ persistentTopicExists.add(brokerService.getPulsar()
+ .getPulsarResources().getTopicResources()
+ .persistentTopicExists(topicName.getPartition(i)));
+ }
+ return FutureUtil.waitForAll(persistentTopicExists)
+ .thenCompose(unused -> {
+ // make sure all sub partitions were deleted after all future complete
+ Optional<Boolean> anyExistPartition = persistentTopicExists.stream()
+ .map(CompletableFuture::join)
+ .filter(topicExist -> topicExist)
+ .findAny();
+ if (anyExistPartition.isPresent()) {
+ throw new UnsupportedOperationException(
Review comment:
Also it is better to log the error here, unfortunately sometimes unchecked exceptions are not logged if they escape the usual system that log unhandled errors
--
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] mattisonchao commented on a change in pull request #14469: [Broker] Fix ``Future.join()`` causing deadlock.
Posted by GitBox <gi...@apache.org>.
mattisonchao commented on a change in pull request #14469:
URL: https://github.com/apache/pulsar/pull/14469#discussion_r815371937
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java
##########
@@ -2256,42 +2256,42 @@ public void checkGC() {
return CompletableFuture.completedFuture(null);
}
TopicName topicName = TopicName.get(TopicName.get(topic).getPartitionedTopicName());
- try {
- PartitionedTopicResources partitionedTopicResources = getBrokerService().pulsar().getPulsarResources()
- .getNamespaceResources()
- .getPartitionedTopicResources();
- if (topicName.isPartitioned() && !partitionedTopicResources.partitionedTopicExists(topicName)) {
- return CompletableFuture.completedFuture(null);
- }
- CompletableFuture<Void> deleteMetadataFuture = new CompletableFuture<>();
- getBrokerService().fetchPartitionedTopicMetadataAsync(TopicName.get(topicName.getPartitionedTopicName()))
- .thenAccept((metadata -> {
- // make sure all sub partitions were deleted
- for (int i = 0; i < metadata.partitions; i++) {
- if (brokerService.getPulsar().getPulsarResources().getTopicResources()
- .persistentTopicExists(topicName.getPartition(i)).join()) {
- throw new UnsupportedOperationException();
- }
- }
- }))
- .thenAccept((res) -> partitionedTopicResources.deletePartitionedTopicAsync(topicName)
- .thenAccept((r) -> {
- deleteMetadataFuture.complete(null);
- }).exceptionally(ex -> {
- deleteMetadataFuture.completeExceptionally(ex.getCause());
- return null;
- }))
- .exceptionally((e) -> {
- if (!(e.getCause() instanceof UnsupportedOperationException)) {
- log.error("delete metadata fail", e);
- }
- deleteMetadataFuture.complete(null);
- return null;
- });
- return deleteMetadataFuture;
- } catch (Exception e) {
- return FutureUtil.failedFuture(e);
- }
+ PartitionedTopicResources partitionedTopicResources = getBrokerService().pulsar().getPulsarResources()
+ .getNamespaceResources()
+ .getPartitionedTopicResources();
+ return partitionedTopicResources.partitionedTopicExistsAsync(topicName)
+ .thenCompose(partitionedTopicExist -> {
+ if (!partitionedTopicExist) {
+ return CompletableFuture.completedFuture(null);
+ } else {
+ return getBrokerService()
+ .fetchPartitionedTopicMetadataAsync(TopicName.get(topicName.getPartitionedTopicName()))
+ .thenCompose((metadata -> {
+ List<CompletableFuture<Boolean>> persistentTopicExists =
+ new ArrayList<>(metadata.partitions);
+ for (int i = 0; i < metadata.partitions; i++) {
+ persistentTopicExists.add(brokerService.getPulsar()
+ .getPulsarResources().getTopicResources()
+ .persistentTopicExists(topicName.getPartition(i)));
+ }
+ return FutureUtil.waitForAll(persistentTopicExists)
+ .thenCompose(unused -> {
+ // make sure all sub partitions were deleted after all future complete
+ Optional<Boolean> anyExistPartition = persistentTopicExists.stream()
+ .map(CompletableFuture::join)
+ .filter(topicExist -> topicExist)
+ .findAny();
+ if (anyExistPartition.isPresent()) {
+ log.error("[{}] Delete metadata fail", topic);
+ return CompletableFuture.completedFuture(null);
Review comment:
Hi, @Technoboy-
I think this concern is not this PR focused, maybe I can create another issue to discuss 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] mattisonchao commented on a change in pull request #14469: [Broker] Fix ``Future.join()`` causing deadlock.
Posted by GitBox <gi...@apache.org>.
mattisonchao commented on a change in pull request #14469:
URL: https://github.com/apache/pulsar/pull/14469#discussion_r815407426
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java
##########
@@ -2256,42 +2256,43 @@ public void checkGC() {
return CompletableFuture.completedFuture(null);
}
TopicName topicName = TopicName.get(TopicName.get(topic).getPartitionedTopicName());
- try {
- PartitionedTopicResources partitionedTopicResources = getBrokerService().pulsar().getPulsarResources()
- .getNamespaceResources()
- .getPartitionedTopicResources();
- if (topicName.isPartitioned() && !partitionedTopicResources.partitionedTopicExists(topicName)) {
- return CompletableFuture.completedFuture(null);
- }
- CompletableFuture<Void> deleteMetadataFuture = new CompletableFuture<>();
- getBrokerService().fetchPartitionedTopicMetadataAsync(TopicName.get(topicName.getPartitionedTopicName()))
- .thenAccept((metadata -> {
- // make sure all sub partitions were deleted
- for (int i = 0; i < metadata.partitions; i++) {
- if (brokerService.getPulsar().getPulsarResources().getTopicResources()
- .persistentTopicExists(topicName.getPartition(i)).join()) {
- throw new UnsupportedOperationException();
- }
- }
- }))
- .thenAccept((res) -> partitionedTopicResources.deletePartitionedTopicAsync(topicName)
- .thenAccept((r) -> {
- deleteMetadataFuture.complete(null);
- }).exceptionally(ex -> {
- deleteMetadataFuture.completeExceptionally(ex.getCause());
- return null;
- }))
- .exceptionally((e) -> {
- if (!(e.getCause() instanceof UnsupportedOperationException)) {
- log.error("delete metadata fail", e);
- }
- deleteMetadataFuture.complete(null);
- return null;
- });
- return deleteMetadataFuture;
- } catch (Exception e) {
- return FutureUtil.failedFuture(e);
- }
+ PartitionedTopicResources partitionedTopicResources = getBrokerService().pulsar().getPulsarResources()
+ .getNamespaceResources()
+ .getPartitionedTopicResources();
+ return partitionedTopicResources.partitionedTopicExistsAsync(topicName)
+ .thenCompose(partitionedTopicExist -> {
+ if (!partitionedTopicExist) {
+ return CompletableFuture.completedFuture(null);
+ } else {
+ return getBrokerService()
+ .fetchPartitionedTopicMetadataAsync(TopicName.get(topicName.getPartitionedTopicName()))
+ .thenCompose((metadata -> {
+ List<CompletableFuture<Boolean>> persistentTopicExists =
+ new ArrayList<>(metadata.partitions);
+ for (int i = 0; i < metadata.partitions; i++) {
+ persistentTopicExists.add(brokerService.getPulsar()
+ .getPulsarResources().getTopicResources()
+ .persistentTopicExists(topicName.getPartition(i)));
+ }
+ return FutureUtil.waitForAll(persistentTopicExists)
+ .thenCompose(unused -> {
+ // make sure all sub partitions were deleted after all future complete
+ Optional<Boolean> anyExistPartition = persistentTopicExists.stream()
+ .map(CompletableFuture::join)
+ .filter(topicExist -> topicExist)
+ .findAny();
+ if (anyExistPartition.isPresent()) {
+ throw new UnsupportedOperationException(
Review comment:
@Jason918 @eolivelli
Fixed.
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java
##########
@@ -2256,42 +2256,43 @@ public void checkGC() {
return CompletableFuture.completedFuture(null);
}
TopicName topicName = TopicName.get(TopicName.get(topic).getPartitionedTopicName());
- try {
- PartitionedTopicResources partitionedTopicResources = getBrokerService().pulsar().getPulsarResources()
- .getNamespaceResources()
- .getPartitionedTopicResources();
- if (topicName.isPartitioned() && !partitionedTopicResources.partitionedTopicExists(topicName)) {
- return CompletableFuture.completedFuture(null);
- }
- CompletableFuture<Void> deleteMetadataFuture = new CompletableFuture<>();
- getBrokerService().fetchPartitionedTopicMetadataAsync(TopicName.get(topicName.getPartitionedTopicName()))
- .thenAccept((metadata -> {
- // make sure all sub partitions were deleted
- for (int i = 0; i < metadata.partitions; i++) {
- if (brokerService.getPulsar().getPulsarResources().getTopicResources()
- .persistentTopicExists(topicName.getPartition(i)).join()) {
- throw new UnsupportedOperationException();
- }
- }
- }))
- .thenAccept((res) -> partitionedTopicResources.deletePartitionedTopicAsync(topicName)
- .thenAccept((r) -> {
- deleteMetadataFuture.complete(null);
- }).exceptionally(ex -> {
- deleteMetadataFuture.completeExceptionally(ex.getCause());
- return null;
- }))
- .exceptionally((e) -> {
- if (!(e.getCause() instanceof UnsupportedOperationException)) {
- log.error("delete metadata fail", e);
- }
- deleteMetadataFuture.complete(null);
- return null;
- });
- return deleteMetadataFuture;
- } catch (Exception e) {
- return FutureUtil.failedFuture(e);
- }
+ PartitionedTopicResources partitionedTopicResources = getBrokerService().pulsar().getPulsarResources()
+ .getNamespaceResources()
+ .getPartitionedTopicResources();
+ return partitionedTopicResources.partitionedTopicExistsAsync(topicName)
+ .thenCompose(partitionedTopicExist -> {
+ if (!partitionedTopicExist) {
+ return CompletableFuture.completedFuture(null);
+ } else {
+ return getBrokerService()
+ .fetchPartitionedTopicMetadataAsync(TopicName.get(topicName.getPartitionedTopicName()))
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
[GitHub] [pulsar] mattisonchao commented on a change in pull request #14469: [Broker] Fix ``Future.join()`` causing deadlock.
Posted by GitBox <gi...@apache.org>.
mattisonchao commented on a change in pull request #14469:
URL: https://github.com/apache/pulsar/pull/14469#discussion_r815250565
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java
##########
@@ -2256,44 +2256,52 @@ public void checkGC() {
return CompletableFuture.completedFuture(null);
}
TopicName topicName = TopicName.get(TopicName.get(topic).getPartitionedTopicName());
- try {
- PartitionedTopicResources partitionedTopicResources = getBrokerService().pulsar().getPulsarResources()
- .getNamespaceResources()
- .getPartitionedTopicResources();
- if (topicName.isPartitioned() && !partitionedTopicResources.partitionedTopicExists(topicName)) {
- return CompletableFuture.completedFuture(null);
- }
- CompletableFuture<Void> deleteMetadataFuture = new CompletableFuture<>();
- getBrokerService().fetchPartitionedTopicMetadataAsync(TopicName.get(topicName.getPartitionedTopicName()))
- .thenAccept((metadata -> {
- // make sure all sub partitions were deleted
- for (int i = 0; i < metadata.partitions; i++) {
- if (brokerService.getPulsar().getPulsarResources().getTopicResources()
- .persistentTopicExists(topicName.getPartition(i)).join()) {
- throw new UnsupportedOperationException();
- }
- }
- }))
- .thenAccept((res) -> partitionedTopicResources.deletePartitionedTopicAsync(topicName)
- .thenAccept((r) -> {
- deleteMetadataFuture.complete(null);
- }).exceptionally(ex -> {
- deleteMetadataFuture.completeExceptionally(ex.getCause());
- return null;
- }))
- .exceptionally((e) -> {
- if (!(e.getCause() instanceof UnsupportedOperationException)) {
- log.error("delete metadata fail", e);
+ PartitionedTopicResources partitionedTopicResources = getBrokerService().pulsar().getPulsarResources()
+ .getNamespaceResources()
+ .getPartitionedTopicResources();
+ if (topicName.isPartitioned()) {
+ return partitionedTopicResources.partitionedTopicExistsAsync(topicName)
+ .thenCompose(partitionedTopicExist -> {
+ if (!partitionedTopicExist) {
+ return CompletableFuture.completedFuture(null);
+ } else {
+ return innerDeletePartitionedMetadata(topicName, partitionedTopicResources);
}
- deleteMetadataFuture.complete(null);
- return null;
});
- return deleteMetadataFuture;
- } catch (Exception e) {
- return FutureUtil.failedFuture(e);
+ } else {
+ return innerDeletePartitionedMetadata(topicName, partitionedTopicResources);
}
}
+ private CompletableFuture<Void> innerDeletePartitionedMetadata(TopicName topicName,
+ PartitionedTopicResources partitionedTopicResources)
+ {
+ return getBrokerService()
+ .fetchPartitionedTopicMetadataAsync(TopicName.get(topicName.getPartitionedTopicName()))
+ .thenCompose((metadata -> {
+ List<CompletableFuture<Boolean>> persistentTopicExists = new ArrayList<>(metadata.partitions);
+ for (int i = 0; i < metadata.partitions; i++) {
+ persistentTopicExists.add(brokerService.getPulsar()
+ .getPulsarResources().getTopicResources()
+ .persistentTopicExists(topicName.getPartition(i)));
+ }
+ return FutureUtil.waitForAll(persistentTopicExists)
+ .thenCompose(unused -> {
+ // make sure all sub partitions were deleted after all future complete
+ Optional<Boolean> anyExistPartition = persistentTopicExists.stream()
+ .map(CompletableFuture::join)
+ .filter(topicExist -> topicExist)
+ .findAny();
+ if (anyExistPartition.isPresent()) {
+ log.error("[{}] Delete metadata fail", topic);
Review comment:
However, I'm a little concerned about this, is there some case where partition metadata deletion fails and always exists in ZK, I'm not sure if we have another mechanism to handle this. If not, I'm happy to work on another PR.
@eolivelli @michaeljmarshall Please give me more suggestions, very thanks~
--
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] mattisonchao commented on a change in pull request #14469: [Broker] Fix ``Future.join()`` causing deadlock.
Posted by GitBox <gi...@apache.org>.
mattisonchao commented on a change in pull request #14469:
URL: https://github.com/apache/pulsar/pull/14469#discussion_r815296561
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java
##########
@@ -2256,42 +2256,42 @@ public void checkGC() {
return CompletableFuture.completedFuture(null);
}
TopicName topicName = TopicName.get(TopicName.get(topic).getPartitionedTopicName());
- try {
- PartitionedTopicResources partitionedTopicResources = getBrokerService().pulsar().getPulsarResources()
- .getNamespaceResources()
- .getPartitionedTopicResources();
- if (topicName.isPartitioned() && !partitionedTopicResources.partitionedTopicExists(topicName)) {
- return CompletableFuture.completedFuture(null);
- }
- CompletableFuture<Void> deleteMetadataFuture = new CompletableFuture<>();
- getBrokerService().fetchPartitionedTopicMetadataAsync(TopicName.get(topicName.getPartitionedTopicName()))
- .thenAccept((metadata -> {
- // make sure all sub partitions were deleted
- for (int i = 0; i < metadata.partitions; i++) {
- if (brokerService.getPulsar().getPulsarResources().getTopicResources()
- .persistentTopicExists(topicName.getPartition(i)).join()) {
- throw new UnsupportedOperationException();
- }
- }
- }))
- .thenAccept((res) -> partitionedTopicResources.deletePartitionedTopicAsync(topicName)
- .thenAccept((r) -> {
- deleteMetadataFuture.complete(null);
- }).exceptionally(ex -> {
- deleteMetadataFuture.completeExceptionally(ex.getCause());
- return null;
- }))
- .exceptionally((e) -> {
- if (!(e.getCause() instanceof UnsupportedOperationException)) {
- log.error("delete metadata fail", e);
- }
- deleteMetadataFuture.complete(null);
- return null;
- });
- return deleteMetadataFuture;
- } catch (Exception e) {
- return FutureUtil.failedFuture(e);
- }
+ PartitionedTopicResources partitionedTopicResources = getBrokerService().pulsar().getPulsarResources()
+ .getNamespaceResources()
+ .getPartitionedTopicResources();
+ return partitionedTopicResources.partitionedTopicExistsAsync(topicName)
+ .thenCompose(partitionedTopicExist -> {
+ if (!partitionedTopicExist) {
+ return CompletableFuture.completedFuture(null);
+ } else {
+ return getBrokerService()
+ .fetchPartitionedTopicMetadataAsync(TopicName.get(topicName.getPartitionedTopicName()))
+ .thenCompose((metadata -> {
+ List<CompletableFuture<Boolean>> persistentTopicExists =
+ new ArrayList<>(metadata.partitions);
+ for (int i = 0; i < metadata.partitions; i++) {
+ persistentTopicExists.add(brokerService.getPulsar()
+ .getPulsarResources().getTopicResources()
+ .persistentTopicExists(topicName.getPartition(i)));
+ }
+ return FutureUtil.waitForAll(persistentTopicExists)
+ .thenCompose(unused -> {
+ // make sure all sub partitions were deleted after all future complete
+ Optional<Boolean> anyExistPartition = persistentTopicExists.stream()
+ .map(CompletableFuture::join)
+ .filter(topicExist -> topicExist)
+ .findAny();
+ if (anyExistPartition.isPresent()) {
+ log.error("[{}] Delete metadata fail", topic);
+ return CompletableFuture.completedFuture(null);
Review comment:
@codelipenghui @michaeljmarshall
Sure, I will fix it ASAP.
--
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] mattisonchao commented on a change in pull request #14469: [Broker] Fix ``Future.join()`` causing deadlock.
Posted by GitBox <gi...@apache.org>.
mattisonchao commented on a change in pull request #14469:
URL: https://github.com/apache/pulsar/pull/14469#discussion_r815311386
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java
##########
@@ -2256,44 +2256,52 @@ public void checkGC() {
return CompletableFuture.completedFuture(null);
}
TopicName topicName = TopicName.get(TopicName.get(topic).getPartitionedTopicName());
- try {
- PartitionedTopicResources partitionedTopicResources = getBrokerService().pulsar().getPulsarResources()
- .getNamespaceResources()
- .getPartitionedTopicResources();
- if (topicName.isPartitioned() && !partitionedTopicResources.partitionedTopicExists(topicName)) {
- return CompletableFuture.completedFuture(null);
- }
- CompletableFuture<Void> deleteMetadataFuture = new CompletableFuture<>();
- getBrokerService().fetchPartitionedTopicMetadataAsync(TopicName.get(topicName.getPartitionedTopicName()))
- .thenAccept((metadata -> {
- // make sure all sub partitions were deleted
- for (int i = 0; i < metadata.partitions; i++) {
- if (brokerService.getPulsar().getPulsarResources().getTopicResources()
- .persistentTopicExists(topicName.getPartition(i)).join()) {
- throw new UnsupportedOperationException();
- }
- }
- }))
- .thenAccept((res) -> partitionedTopicResources.deletePartitionedTopicAsync(topicName)
- .thenAccept((r) -> {
- deleteMetadataFuture.complete(null);
- }).exceptionally(ex -> {
- deleteMetadataFuture.completeExceptionally(ex.getCause());
- return null;
- }))
- .exceptionally((e) -> {
- if (!(e.getCause() instanceof UnsupportedOperationException)) {
- log.error("delete metadata fail", e);
+ PartitionedTopicResources partitionedTopicResources = getBrokerService().pulsar().getPulsarResources()
+ .getNamespaceResources()
+ .getPartitionedTopicResources();
+ if (topicName.isPartitioned()) {
+ return partitionedTopicResources.partitionedTopicExistsAsync(topicName)
+ .thenCompose(partitionedTopicExist -> {
+ if (!partitionedTopicExist) {
+ return CompletableFuture.completedFuture(null);
+ } else {
+ return innerDeletePartitionedMetadata(topicName, partitionedTopicResources);
}
- deleteMetadataFuture.complete(null);
- return null;
});
- return deleteMetadataFuture;
- } catch (Exception e) {
- return FutureUtil.failedFuture(e);
+ } else {
+ return innerDeletePartitionedMetadata(topicName, partitionedTopicResources);
}
}
+ private CompletableFuture<Void> innerDeletePartitionedMetadata(TopicName topicName,
+ PartitionedTopicResources partitionedTopicResources)
+ {
+ return getBrokerService()
+ .fetchPartitionedTopicMetadataAsync(TopicName.get(topicName.getPartitionedTopicName()))
+ .thenCompose((metadata -> {
+ List<CompletableFuture<Boolean>> persistentTopicExists = new ArrayList<>(metadata.partitions);
+ for (int i = 0; i < metadata.partitions; i++) {
+ persistentTopicExists.add(brokerService.getPulsar()
+ .getPulsarResources().getTopicResources()
+ .persistentTopicExists(topicName.getPartition(i)));
+ }
+ return FutureUtil.waitForAll(persistentTopicExists)
+ .thenCompose(unused -> {
+ // make sure all sub partitions were deleted after all future complete
+ Optional<Boolean> anyExistPartition = persistentTopicExists.stream()
+ .map(CompletableFuture::join)
+ .filter(topicExist -> topicExist)
+ .findAny();
+ if (anyExistPartition.isPresent()) {
+ log.error("[{}] Delete metadata fail", topic);
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
[GitHub] [pulsar] Jason918 commented on a change in pull request #14469: [Broker] Fix ``Future.join()`` causing deadlock.
Posted by GitBox <gi...@apache.org>.
Jason918 commented on a change in pull request #14469:
URL: https://github.com/apache/pulsar/pull/14469#discussion_r815381896
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java
##########
@@ -2256,42 +2256,43 @@ public void checkGC() {
return CompletableFuture.completedFuture(null);
}
TopicName topicName = TopicName.get(TopicName.get(topic).getPartitionedTopicName());
- try {
- PartitionedTopicResources partitionedTopicResources = getBrokerService().pulsar().getPulsarResources()
- .getNamespaceResources()
- .getPartitionedTopicResources();
- if (topicName.isPartitioned() && !partitionedTopicResources.partitionedTopicExists(topicName)) {
- return CompletableFuture.completedFuture(null);
- }
- CompletableFuture<Void> deleteMetadataFuture = new CompletableFuture<>();
- getBrokerService().fetchPartitionedTopicMetadataAsync(TopicName.get(topicName.getPartitionedTopicName()))
- .thenAccept((metadata -> {
- // make sure all sub partitions were deleted
- for (int i = 0; i < metadata.partitions; i++) {
- if (brokerService.getPulsar().getPulsarResources().getTopicResources()
- .persistentTopicExists(topicName.getPartition(i)).join()) {
- throw new UnsupportedOperationException();
- }
- }
- }))
- .thenAccept((res) -> partitionedTopicResources.deletePartitionedTopicAsync(topicName)
- .thenAccept((r) -> {
- deleteMetadataFuture.complete(null);
- }).exceptionally(ex -> {
- deleteMetadataFuture.completeExceptionally(ex.getCause());
- return null;
- }))
- .exceptionally((e) -> {
- if (!(e.getCause() instanceof UnsupportedOperationException)) {
- log.error("delete metadata fail", e);
- }
- deleteMetadataFuture.complete(null);
- return null;
- });
- return deleteMetadataFuture;
- } catch (Exception e) {
- return FutureUtil.failedFuture(e);
- }
+ PartitionedTopicResources partitionedTopicResources = getBrokerService().pulsar().getPulsarResources()
+ .getNamespaceResources()
+ .getPartitionedTopicResources();
+ return partitionedTopicResources.partitionedTopicExistsAsync(topicName)
+ .thenCompose(partitionedTopicExist -> {
+ if (!partitionedTopicExist) {
+ return CompletableFuture.completedFuture(null);
+ } else {
+ return getBrokerService()
+ .fetchPartitionedTopicMetadataAsync(TopicName.get(topicName.getPartitionedTopicName()))
Review comment:
Is this `topicName` already changed to partitioned topic name ? So we can skip `TopicName.get(topicName.getPartitionedTopicName()` here.
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java
##########
@@ -2256,42 +2256,43 @@ public void checkGC() {
return CompletableFuture.completedFuture(null);
}
TopicName topicName = TopicName.get(TopicName.get(topic).getPartitionedTopicName());
- try {
- PartitionedTopicResources partitionedTopicResources = getBrokerService().pulsar().getPulsarResources()
- .getNamespaceResources()
- .getPartitionedTopicResources();
- if (topicName.isPartitioned() && !partitionedTopicResources.partitionedTopicExists(topicName)) {
- return CompletableFuture.completedFuture(null);
- }
- CompletableFuture<Void> deleteMetadataFuture = new CompletableFuture<>();
- getBrokerService().fetchPartitionedTopicMetadataAsync(TopicName.get(topicName.getPartitionedTopicName()))
- .thenAccept((metadata -> {
- // make sure all sub partitions were deleted
- for (int i = 0; i < metadata.partitions; i++) {
- if (brokerService.getPulsar().getPulsarResources().getTopicResources()
- .persistentTopicExists(topicName.getPartition(i)).join()) {
- throw new UnsupportedOperationException();
- }
- }
- }))
- .thenAccept((res) -> partitionedTopicResources.deletePartitionedTopicAsync(topicName)
- .thenAccept((r) -> {
- deleteMetadataFuture.complete(null);
- }).exceptionally(ex -> {
- deleteMetadataFuture.completeExceptionally(ex.getCause());
- return null;
- }))
- .exceptionally((e) -> {
- if (!(e.getCause() instanceof UnsupportedOperationException)) {
- log.error("delete metadata fail", e);
- }
- deleteMetadataFuture.complete(null);
- return null;
- });
- return deleteMetadataFuture;
- } catch (Exception e) {
- return FutureUtil.failedFuture(e);
- }
+ PartitionedTopicResources partitionedTopicResources = getBrokerService().pulsar().getPulsarResources()
+ .getNamespaceResources()
+ .getPartitionedTopicResources();
+ return partitionedTopicResources.partitionedTopicExistsAsync(topicName)
+ .thenCompose(partitionedTopicExist -> {
+ if (!partitionedTopicExist) {
+ return CompletableFuture.completedFuture(null);
+ } else {
+ return getBrokerService()
+ .fetchPartitionedTopicMetadataAsync(TopicName.get(topicName.getPartitionedTopicName()))
+ .thenCompose((metadata -> {
+ List<CompletableFuture<Boolean>> persistentTopicExists =
+ new ArrayList<>(metadata.partitions);
+ for (int i = 0; i < metadata.partitions; i++) {
+ persistentTopicExists.add(brokerService.getPulsar()
+ .getPulsarResources().getTopicResources()
+ .persistentTopicExists(topicName.getPartition(i)));
+ }
+ return FutureUtil.waitForAll(persistentTopicExists)
+ .thenCompose(unused -> {
+ // make sure all sub partitions were deleted after all future complete
+ Optional<Boolean> anyExistPartition = persistentTopicExists.stream()
+ .map(CompletableFuture::join)
+ .filter(topicExist -> topicExist)
+ .findAny();
+ if (anyExistPartition.isPresent()) {
+ throw new UnsupportedOperationException(
Review comment:
It seems that this exception will be returned in the result future. Please make sure it's will be handled well in outside caller.
--
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 #14469: [Broker] Fix ``Future.join()`` causing deadlock.
Posted by GitBox <gi...@apache.org>.
eolivelli commented on pull request #14469:
URL: https://github.com/apache/pulsar/pull/14469#issuecomment-1053946246
@michaeljmarshall could you please take a final look ?
--
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] mattisonchao removed a comment on pull request #14469: [Broker] Fix ``Future.join()`` causing deadlock.
Posted by GitBox <gi...@apache.org>.
mattisonchao removed a comment on pull request #14469:
URL: https://github.com/apache/pulsar/pull/14469#issuecomment-1053740404
/pulsarbot rerun-failure-checks
--
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] mattisonchao commented on a change in pull request #14469: [Broker] Fix ``Future.join()`` causing deadlock.
Posted by GitBox <gi...@apache.org>.
mattisonchao commented on a change in pull request #14469:
URL: https://github.com/apache/pulsar/pull/14469#discussion_r815247189
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java
##########
@@ -2256,44 +2256,52 @@ public void checkGC() {
return CompletableFuture.completedFuture(null);
}
TopicName topicName = TopicName.get(TopicName.get(topic).getPartitionedTopicName());
- try {
- PartitionedTopicResources partitionedTopicResources = getBrokerService().pulsar().getPulsarResources()
- .getNamespaceResources()
- .getPartitionedTopicResources();
- if (topicName.isPartitioned() && !partitionedTopicResources.partitionedTopicExists(topicName)) {
- return CompletableFuture.completedFuture(null);
- }
- CompletableFuture<Void> deleteMetadataFuture = new CompletableFuture<>();
- getBrokerService().fetchPartitionedTopicMetadataAsync(TopicName.get(topicName.getPartitionedTopicName()))
- .thenAccept((metadata -> {
- // make sure all sub partitions were deleted
- for (int i = 0; i < metadata.partitions; i++) {
- if (brokerService.getPulsar().getPulsarResources().getTopicResources()
- .persistentTopicExists(topicName.getPartition(i)).join()) {
- throw new UnsupportedOperationException();
- }
- }
- }))
- .thenAccept((res) -> partitionedTopicResources.deletePartitionedTopicAsync(topicName)
- .thenAccept((r) -> {
- deleteMetadataFuture.complete(null);
- }).exceptionally(ex -> {
- deleteMetadataFuture.completeExceptionally(ex.getCause());
- return null;
- }))
- .exceptionally((e) -> {
- if (!(e.getCause() instanceof UnsupportedOperationException)) {
- log.error("delete metadata fail", e);
+ PartitionedTopicResources partitionedTopicResources = getBrokerService().pulsar().getPulsarResources()
+ .getNamespaceResources()
+ .getPartitionedTopicResources();
+ if (topicName.isPartitioned()) {
+ return partitionedTopicResources.partitionedTopicExistsAsync(topicName)
+ .thenCompose(partitionedTopicExist -> {
+ if (!partitionedTopicExist) {
+ return CompletableFuture.completedFuture(null);
+ } else {
+ return innerDeletePartitionedMetadata(topicName, partitionedTopicResources);
}
- deleteMetadataFuture.complete(null);
- return null;
});
- return deleteMetadataFuture;
- } catch (Exception e) {
- return FutureUtil.failedFuture(e);
+ } else {
+ return innerDeletePartitionedMetadata(topicName, partitionedTopicResources);
}
}
+ private CompletableFuture<Void> innerDeletePartitionedMetadata(TopicName topicName,
+ PartitionedTopicResources partitionedTopicResources)
+ {
+ return getBrokerService()
+ .fetchPartitionedTopicMetadataAsync(TopicName.get(topicName.getPartitionedTopicName()))
+ .thenCompose((metadata -> {
+ List<CompletableFuture<Boolean>> persistentTopicExists = new ArrayList<>(metadata.partitions);
+ for (int i = 0; i < metadata.partitions; i++) {
+ persistentTopicExists.add(brokerService.getPulsar()
+ .getPulsarResources().getTopicResources()
+ .persistentTopicExists(topicName.getPartition(i)));
+ }
+ return FutureUtil.waitForAll(persistentTopicExists)
+ .thenCompose(unused -> {
+ // make sure all sub partitions were deleted after all future complete
+ Optional<Boolean> anyExistPartition = persistentTopicExists.stream()
+ .map(CompletableFuture::join)
+ .filter(topicExist -> topicExist)
+ .findAny();
+ if (anyExistPartition.isPresent()) {
+ log.error("[{}] Delete metadata fail", topic);
Review comment:
@eolivelli @michaeljmarshall
Yes, the previous behaviour is to throw ``UnsupportedOperationException ``, but this exception will be caught here.
https://github.com/apache/pulsar/blob/5235ff919f68ef83690eca465e852f74d56e3899/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java#L2296-L2303
--
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] michaeljmarshall commented on a change in pull request #14469: [Broker] Fix ``Future.join()`` causing deadlock.
Posted by GitBox <gi...@apache.org>.
michaeljmarshall commented on a change in pull request #14469:
URL: https://github.com/apache/pulsar/pull/14469#discussion_r815276360
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java
##########
@@ -2256,44 +2256,52 @@ public void checkGC() {
return CompletableFuture.completedFuture(null);
}
TopicName topicName = TopicName.get(TopicName.get(topic).getPartitionedTopicName());
- try {
- PartitionedTopicResources partitionedTopicResources = getBrokerService().pulsar().getPulsarResources()
- .getNamespaceResources()
- .getPartitionedTopicResources();
- if (topicName.isPartitioned() && !partitionedTopicResources.partitionedTopicExists(topicName)) {
- return CompletableFuture.completedFuture(null);
- }
- CompletableFuture<Void> deleteMetadataFuture = new CompletableFuture<>();
- getBrokerService().fetchPartitionedTopicMetadataAsync(TopicName.get(topicName.getPartitionedTopicName()))
- .thenAccept((metadata -> {
- // make sure all sub partitions were deleted
- for (int i = 0; i < metadata.partitions; i++) {
- if (brokerService.getPulsar().getPulsarResources().getTopicResources()
- .persistentTopicExists(topicName.getPartition(i)).join()) {
- throw new UnsupportedOperationException();
- }
- }
- }))
- .thenAccept((res) -> partitionedTopicResources.deletePartitionedTopicAsync(topicName)
- .thenAccept((r) -> {
- deleteMetadataFuture.complete(null);
- }).exceptionally(ex -> {
- deleteMetadataFuture.completeExceptionally(ex.getCause());
- return null;
- }))
- .exceptionally((e) -> {
- if (!(e.getCause() instanceof UnsupportedOperationException)) {
- log.error("delete metadata fail", e);
+ PartitionedTopicResources partitionedTopicResources = getBrokerService().pulsar().getPulsarResources()
+ .getNamespaceResources()
+ .getPartitionedTopicResources();
+ if (topicName.isPartitioned()) {
+ return partitionedTopicResources.partitionedTopicExistsAsync(topicName)
+ .thenCompose(partitionedTopicExist -> {
+ if (!partitionedTopicExist) {
+ return CompletableFuture.completedFuture(null);
+ } else {
+ return innerDeletePartitionedMetadata(topicName, partitionedTopicResources);
}
- deleteMetadataFuture.complete(null);
- return null;
});
- return deleteMetadataFuture;
- } catch (Exception e) {
- return FutureUtil.failedFuture(e);
+ } else {
+ return innerDeletePartitionedMetadata(topicName, partitionedTopicResources);
}
}
+ private CompletableFuture<Void> innerDeletePartitionedMetadata(TopicName topicName,
+ PartitionedTopicResources partitionedTopicResources)
+ {
+ return getBrokerService()
+ .fetchPartitionedTopicMetadataAsync(TopicName.get(topicName.getPartitionedTopicName()))
+ .thenCompose((metadata -> {
+ List<CompletableFuture<Boolean>> persistentTopicExists = new ArrayList<>(metadata.partitions);
+ for (int i = 0; i < metadata.partitions; i++) {
+ persistentTopicExists.add(brokerService.getPulsar()
+ .getPulsarResources().getTopicResources()
+ .persistentTopicExists(topicName.getPartition(i)));
+ }
+ return FutureUtil.waitForAll(persistentTopicExists)
+ .thenCompose(unused -> {
+ // make sure all sub partitions were deleted after all future complete
+ Optional<Boolean> anyExistPartition = persistentTopicExists.stream()
+ .map(CompletableFuture::join)
+ .filter(topicExist -> topicExist)
+ .findAny();
+ if (anyExistPartition.isPresent()) {
+ log.error("[{}] Delete metadata fail", topic);
Review comment:
@mattisonchao - thank you for the explanation. You are correct about the current behavior. I think this logic is probably not what I expected because I forgot that it is an internal cleanup loop in the behavior.
--
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] mattisonchao commented on a change in pull request #14469: [Broker] Fix ``Future.join()`` causing deadlock.
Posted by GitBox <gi...@apache.org>.
mattisonchao commented on a change in pull request #14469:
URL: https://github.com/apache/pulsar/pull/14469#discussion_r815406729
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java
##########
@@ -2256,42 +2256,43 @@ public void checkGC() {
return CompletableFuture.completedFuture(null);
}
TopicName topicName = TopicName.get(TopicName.get(topic).getPartitionedTopicName());
- try {
- PartitionedTopicResources partitionedTopicResources = getBrokerService().pulsar().getPulsarResources()
- .getNamespaceResources()
- .getPartitionedTopicResources();
- if (topicName.isPartitioned() && !partitionedTopicResources.partitionedTopicExists(topicName)) {
- return CompletableFuture.completedFuture(null);
- }
- CompletableFuture<Void> deleteMetadataFuture = new CompletableFuture<>();
- getBrokerService().fetchPartitionedTopicMetadataAsync(TopicName.get(topicName.getPartitionedTopicName()))
- .thenAccept((metadata -> {
- // make sure all sub partitions were deleted
- for (int i = 0; i < metadata.partitions; i++) {
- if (brokerService.getPulsar().getPulsarResources().getTopicResources()
- .persistentTopicExists(topicName.getPartition(i)).join()) {
- throw new UnsupportedOperationException();
- }
- }
- }))
- .thenAccept((res) -> partitionedTopicResources.deletePartitionedTopicAsync(topicName)
- .thenAccept((r) -> {
- deleteMetadataFuture.complete(null);
- }).exceptionally(ex -> {
- deleteMetadataFuture.completeExceptionally(ex.getCause());
- return null;
- }))
- .exceptionally((e) -> {
- if (!(e.getCause() instanceof UnsupportedOperationException)) {
- log.error("delete metadata fail", e);
- }
- deleteMetadataFuture.complete(null);
- return null;
- });
- return deleteMetadataFuture;
- } catch (Exception e) {
- return FutureUtil.failedFuture(e);
- }
+ PartitionedTopicResources partitionedTopicResources = getBrokerService().pulsar().getPulsarResources()
+ .getNamespaceResources()
+ .getPartitionedTopicResources();
+ return partitionedTopicResources.partitionedTopicExistsAsync(topicName)
+ .thenCompose(partitionedTopicExist -> {
+ if (!partitionedTopicExist) {
+ return CompletableFuture.completedFuture(null);
+ } else {
+ return getBrokerService()
+ .fetchPartitionedTopicMetadataAsync(TopicName.get(topicName.getPartitionedTopicName()))
+ .thenCompose((metadata -> {
+ List<CompletableFuture<Boolean>> persistentTopicExists =
+ new ArrayList<>(metadata.partitions);
+ for (int i = 0; i < metadata.partitions; i++) {
+ persistentTopicExists.add(brokerService.getPulsar()
+ .getPulsarResources().getTopicResources()
+ .persistentTopicExists(topicName.getPartition(i)));
+ }
+ return FutureUtil.waitForAll(persistentTopicExists)
+ .thenCompose(unused -> {
+ // make sure all sub partitions were deleted after all future complete
+ Optional<Boolean> anyExistPartition = persistentTopicExists.stream()
+ .map(CompletableFuture::join)
+ .filter(topicExist -> topicExist)
+ .findAny();
+ if (anyExistPartition.isPresent()) {
+ throw new UnsupportedOperationException(
Review comment:
Hi, @Jason918
The ``PersistentTopic#checkGC`` method already handles this error.
Also, I think a good code style is to handle all exceptions on CompletableFuture in the outer layer. Unless we need to do some special handling for exceptions.
Good
```java
class Scratch {
public static void main(String[] args) {
CompletableFuture.completedFuture(null)
.thenCompose(__ -> futureMethodA())
.thenCompose(__ -> futureMethodB())
.thenCompose(__ -> futureMethodC())
.exceptionally(ex -> {
Throwable realCause = ex.getCause();
if (realCause instanceof IllegalArgumentException) {
// do sth
} else if (realCause instanceof IllegalStateException) {
// do sth
} else if (realCause instanceof RuntimeException) {
// do sth
}
return null;
});
}
public static CompletableFuture<Void> futureMethodA() {
throw new IllegalArgumentException();
}
public static CompletableFuture<Void> futureMethodB() {
throw new IllegalStateException();
}
public static CompletableFuture<Void> futureMethodC() {
throw new RuntimeException();
}
}
```
Bad
```java
class Scratch {
public static void main(String[] args) {
CompletableFuture.completedFuture(null)
.thenAccept(__ ->
futureMethodA().exceptionally(ex -> {
Throwable realCause = ex.getCause();
// do sth
return null;
})
)
.thenAccept(__ -> futureMethodB()
.exceptionally(ex -> {
Throwable realCause = ex.getCause();
// do sth
return null;
})
)
.thenCompose(__ -> futureMethodC()
.exceptionally(ex -> {
Throwable realCause = ex.getCause();
// do sth
return null;
}));
}
public static CompletableFuture<Void> futureMethodA() {
throw new IllegalArgumentException();
}
public static CompletableFuture<Void> futureMethodB() {
throw new IllegalStateException();
}
public static CompletableFuture<Void> futureMethodC() {
throw new RuntimeException();
}
}
```
I'm not sure if you agree with me, please let me know if you have any suggestions.
very thanks.
--
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] michaeljmarshall commented on a change in pull request #14469: [Broker] Fix ``Future.join()`` causing deadlock.
Posted by GitBox <gi...@apache.org>.
michaeljmarshall commented on a change in pull request #14469:
URL: https://github.com/apache/pulsar/pull/14469#discussion_r815221561
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java
##########
@@ -2256,44 +2256,52 @@ public void checkGC() {
return CompletableFuture.completedFuture(null);
}
TopicName topicName = TopicName.get(TopicName.get(topic).getPartitionedTopicName());
- try {
- PartitionedTopicResources partitionedTopicResources = getBrokerService().pulsar().getPulsarResources()
- .getNamespaceResources()
- .getPartitionedTopicResources();
- if (topicName.isPartitioned() && !partitionedTopicResources.partitionedTopicExists(topicName)) {
- return CompletableFuture.completedFuture(null);
- }
- CompletableFuture<Void> deleteMetadataFuture = new CompletableFuture<>();
- getBrokerService().fetchPartitionedTopicMetadataAsync(TopicName.get(topicName.getPartitionedTopicName()))
- .thenAccept((metadata -> {
- // make sure all sub partitions were deleted
- for (int i = 0; i < metadata.partitions; i++) {
- if (brokerService.getPulsar().getPulsarResources().getTopicResources()
- .persistentTopicExists(topicName.getPartition(i)).join()) {
- throw new UnsupportedOperationException();
- }
- }
- }))
- .thenAccept((res) -> partitionedTopicResources.deletePartitionedTopicAsync(topicName)
- .thenAccept((r) -> {
- deleteMetadataFuture.complete(null);
- }).exceptionally(ex -> {
- deleteMetadataFuture.completeExceptionally(ex.getCause());
- return null;
- }))
- .exceptionally((e) -> {
- if (!(e.getCause() instanceof UnsupportedOperationException)) {
- log.error("delete metadata fail", e);
+ PartitionedTopicResources partitionedTopicResources = getBrokerService().pulsar().getPulsarResources()
+ .getNamespaceResources()
+ .getPartitionedTopicResources();
+ if (topicName.isPartitioned()) {
Review comment:
We've already verified that the topic is partitioned on line 2255. I don't think we need this check.
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java
##########
@@ -2256,44 +2256,52 @@ public void checkGC() {
return CompletableFuture.completedFuture(null);
}
TopicName topicName = TopicName.get(TopicName.get(topic).getPartitionedTopicName());
- try {
- PartitionedTopicResources partitionedTopicResources = getBrokerService().pulsar().getPulsarResources()
- .getNamespaceResources()
- .getPartitionedTopicResources();
- if (topicName.isPartitioned() && !partitionedTopicResources.partitionedTopicExists(topicName)) {
- return CompletableFuture.completedFuture(null);
- }
- CompletableFuture<Void> deleteMetadataFuture = new CompletableFuture<>();
- getBrokerService().fetchPartitionedTopicMetadataAsync(TopicName.get(topicName.getPartitionedTopicName()))
- .thenAccept((metadata -> {
- // make sure all sub partitions were deleted
- for (int i = 0; i < metadata.partitions; i++) {
- if (brokerService.getPulsar().getPulsarResources().getTopicResources()
- .persistentTopicExists(topicName.getPartition(i)).join()) {
- throw new UnsupportedOperationException();
- }
- }
- }))
- .thenAccept((res) -> partitionedTopicResources.deletePartitionedTopicAsync(topicName)
- .thenAccept((r) -> {
- deleteMetadataFuture.complete(null);
- }).exceptionally(ex -> {
- deleteMetadataFuture.completeExceptionally(ex.getCause());
- return null;
- }))
- .exceptionally((e) -> {
- if (!(e.getCause() instanceof UnsupportedOperationException)) {
- log.error("delete metadata fail", e);
+ PartitionedTopicResources partitionedTopicResources = getBrokerService().pulsar().getPulsarResources()
+ .getNamespaceResources()
+ .getPartitionedTopicResources();
+ if (topicName.isPartitioned()) {
+ return partitionedTopicResources.partitionedTopicExistsAsync(topicName)
+ .thenCompose(partitionedTopicExist -> {
+ if (!partitionedTopicExist) {
+ return CompletableFuture.completedFuture(null);
+ } else {
+ return innerDeletePartitionedMetadata(topicName, partitionedTopicResources);
}
- deleteMetadataFuture.complete(null);
- return null;
});
- return deleteMetadataFuture;
- } catch (Exception e) {
- return FutureUtil.failedFuture(e);
+ } else {
+ return innerDeletePartitionedMetadata(topicName, partitionedTopicResources);
}
Review comment:
This method is only for partitioned topics, so we can drop this code.
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java
##########
@@ -2256,44 +2256,52 @@ public void checkGC() {
return CompletableFuture.completedFuture(null);
}
TopicName topicName = TopicName.get(TopicName.get(topic).getPartitionedTopicName());
- try {
- PartitionedTopicResources partitionedTopicResources = getBrokerService().pulsar().getPulsarResources()
- .getNamespaceResources()
- .getPartitionedTopicResources();
- if (topicName.isPartitioned() && !partitionedTopicResources.partitionedTopicExists(topicName)) {
- return CompletableFuture.completedFuture(null);
- }
- CompletableFuture<Void> deleteMetadataFuture = new CompletableFuture<>();
- getBrokerService().fetchPartitionedTopicMetadataAsync(TopicName.get(topicName.getPartitionedTopicName()))
- .thenAccept((metadata -> {
- // make sure all sub partitions were deleted
- for (int i = 0; i < metadata.partitions; i++) {
- if (brokerService.getPulsar().getPulsarResources().getTopicResources()
- .persistentTopicExists(topicName.getPartition(i)).join()) {
- throw new UnsupportedOperationException();
- }
- }
- }))
- .thenAccept((res) -> partitionedTopicResources.deletePartitionedTopicAsync(topicName)
- .thenAccept((r) -> {
- deleteMetadataFuture.complete(null);
- }).exceptionally(ex -> {
- deleteMetadataFuture.completeExceptionally(ex.getCause());
- return null;
- }))
- .exceptionally((e) -> {
- if (!(e.getCause() instanceof UnsupportedOperationException)) {
- log.error("delete metadata fail", e);
+ PartitionedTopicResources partitionedTopicResources = getBrokerService().pulsar().getPulsarResources()
+ .getNamespaceResources()
+ .getPartitionedTopicResources();
+ if (topicName.isPartitioned()) {
+ return partitionedTopicResources.partitionedTopicExistsAsync(topicName)
+ .thenCompose(partitionedTopicExist -> {
+ if (!partitionedTopicExist) {
+ return CompletableFuture.completedFuture(null);
+ } else {
+ return innerDeletePartitionedMetadata(topicName, partitionedTopicResources);
}
- deleteMetadataFuture.complete(null);
- return null;
});
- return deleteMetadataFuture;
- } catch (Exception e) {
- return FutureUtil.failedFuture(e);
+ } else {
+ return innerDeletePartitionedMetadata(topicName, partitionedTopicResources);
}
}
+ private CompletableFuture<Void> innerDeletePartitionedMetadata(TopicName topicName,
+ PartitionedTopicResources partitionedTopicResources)
+ {
+ return getBrokerService()
+ .fetchPartitionedTopicMetadataAsync(TopicName.get(topicName.getPartitionedTopicName()))
+ .thenCompose((metadata -> {
+ List<CompletableFuture<Boolean>> persistentTopicExists = new ArrayList<>(metadata.partitions);
+ for (int i = 0; i < metadata.partitions; i++) {
+ persistentTopicExists.add(brokerService.getPulsar()
+ .getPulsarResources().getTopicResources()
+ .persistentTopicExists(topicName.getPartition(i)));
+ }
+ return FutureUtil.waitForAll(persistentTopicExists)
+ .thenCompose(unused -> {
+ // make sure all sub partitions were deleted after all future complete
+ Optional<Boolean> anyExistPartition = persistentTopicExists.stream()
+ .map(CompletableFuture::join)
+ .filter(topicExist -> topicExist)
+ .findAny();
+ if (anyExistPartition.isPresent()) {
+ log.error("[{}] Delete metadata fail", topic);
Review comment:
Good question @eolivelli. The previous behavior was to throw `UnsupportedOperationException`.
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java
##########
@@ -2256,44 +2256,52 @@ public void checkGC() {
return CompletableFuture.completedFuture(null);
}
TopicName topicName = TopicName.get(TopicName.get(topic).getPartitionedTopicName());
- try {
- PartitionedTopicResources partitionedTopicResources = getBrokerService().pulsar().getPulsarResources()
- .getNamespaceResources()
- .getPartitionedTopicResources();
- if (topicName.isPartitioned() && !partitionedTopicResources.partitionedTopicExists(topicName)) {
- return CompletableFuture.completedFuture(null);
- }
- CompletableFuture<Void> deleteMetadataFuture = new CompletableFuture<>();
- getBrokerService().fetchPartitionedTopicMetadataAsync(TopicName.get(topicName.getPartitionedTopicName()))
- .thenAccept((metadata -> {
- // make sure all sub partitions were deleted
- for (int i = 0; i < metadata.partitions; i++) {
- if (brokerService.getPulsar().getPulsarResources().getTopicResources()
- .persistentTopicExists(topicName.getPartition(i)).join()) {
- throw new UnsupportedOperationException();
- }
- }
- }))
- .thenAccept((res) -> partitionedTopicResources.deletePartitionedTopicAsync(topicName)
- .thenAccept((r) -> {
- deleteMetadataFuture.complete(null);
- }).exceptionally(ex -> {
- deleteMetadataFuture.completeExceptionally(ex.getCause());
- return null;
- }))
- .exceptionally((e) -> {
- if (!(e.getCause() instanceof UnsupportedOperationException)) {
- log.error("delete metadata fail", e);
+ PartitionedTopicResources partitionedTopicResources = getBrokerService().pulsar().getPulsarResources()
+ .getNamespaceResources()
+ .getPartitionedTopicResources();
+ if (topicName.isPartitioned()) {
+ return partitionedTopicResources.partitionedTopicExistsAsync(topicName)
+ .thenCompose(partitionedTopicExist -> {
+ if (!partitionedTopicExist) {
+ return CompletableFuture.completedFuture(null);
+ } else {
+ return innerDeletePartitionedMetadata(topicName, partitionedTopicResources);
}
- deleteMetadataFuture.complete(null);
- return null;
});
- return deleteMetadataFuture;
- } catch (Exception e) {
- return FutureUtil.failedFuture(e);
+ } else {
+ return innerDeletePartitionedMetadata(topicName, partitionedTopicResources);
}
}
+ private CompletableFuture<Void> innerDeletePartitionedMetadata(TopicName topicName,
+ PartitionedTopicResources partitionedTopicResources)
+ {
+ return getBrokerService()
+ .fetchPartitionedTopicMetadataAsync(TopicName.get(topicName.getPartitionedTopicName()))
+ .thenCompose((metadata -> {
+ List<CompletableFuture<Boolean>> persistentTopicExists = new ArrayList<>(metadata.partitions);
Review comment:
We need a thread safe data structure here.
--
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] mattisonchao commented on pull request #14469: [Broker] Fix ``Future.join()`` causing deadlock.
Posted by GitBox <gi...@apache.org>.
mattisonchao commented on pull request #14469:
URL: https://github.com/apache/pulsar/pull/14469#issuecomment-1053740404
/pulsarbot rerun-failure-checks
--
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] mattisonchao commented on a change in pull request #14469: [Broker] Fix ``Future.join()`` causing deadlock.
Posted by GitBox <gi...@apache.org>.
mattisonchao commented on a change in pull request #14469:
URL: https://github.com/apache/pulsar/pull/14469#discussion_r816348536
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java
##########
@@ -2256,44 +2256,52 @@ public void checkGC() {
return CompletableFuture.completedFuture(null);
}
TopicName topicName = TopicName.get(TopicName.get(topic).getPartitionedTopicName());
- try {
- PartitionedTopicResources partitionedTopicResources = getBrokerService().pulsar().getPulsarResources()
- .getNamespaceResources()
- .getPartitionedTopicResources();
- if (topicName.isPartitioned() && !partitionedTopicResources.partitionedTopicExists(topicName)) {
- return CompletableFuture.completedFuture(null);
- }
- CompletableFuture<Void> deleteMetadataFuture = new CompletableFuture<>();
- getBrokerService().fetchPartitionedTopicMetadataAsync(TopicName.get(topicName.getPartitionedTopicName()))
- .thenAccept((metadata -> {
- // make sure all sub partitions were deleted
- for (int i = 0; i < metadata.partitions; i++) {
- if (brokerService.getPulsar().getPulsarResources().getTopicResources()
- .persistentTopicExists(topicName.getPartition(i)).join()) {
- throw new UnsupportedOperationException();
- }
- }
- }))
- .thenAccept((res) -> partitionedTopicResources.deletePartitionedTopicAsync(topicName)
- .thenAccept((r) -> {
- deleteMetadataFuture.complete(null);
- }).exceptionally(ex -> {
- deleteMetadataFuture.completeExceptionally(ex.getCause());
- return null;
- }))
- .exceptionally((e) -> {
- if (!(e.getCause() instanceof UnsupportedOperationException)) {
- log.error("delete metadata fail", e);
+ PartitionedTopicResources partitionedTopicResources = getBrokerService().pulsar().getPulsarResources()
+ .getNamespaceResources()
+ .getPartitionedTopicResources();
+ if (topicName.isPartitioned()) {
+ return partitionedTopicResources.partitionedTopicExistsAsync(topicName)
+ .thenCompose(partitionedTopicExist -> {
+ if (!partitionedTopicExist) {
+ return CompletableFuture.completedFuture(null);
+ } else {
+ return innerDeletePartitionedMetadata(topicName, partitionedTopicResources);
}
- deleteMetadataFuture.complete(null);
- return null;
});
- return deleteMetadataFuture;
- } catch (Exception e) {
- return FutureUtil.failedFuture(e);
+ } else {
+ return innerDeletePartitionedMetadata(topicName, partitionedTopicResources);
}
}
+ private CompletableFuture<Void> innerDeletePartitionedMetadata(TopicName topicName,
+ PartitionedTopicResources partitionedTopicResources)
+ {
+ return getBrokerService()
+ .fetchPartitionedTopicMetadataAsync(TopicName.get(topicName.getPartitionedTopicName()))
+ .thenCompose((metadata -> {
+ List<CompletableFuture<Boolean>> persistentTopicExists = new ArrayList<>(metadata.partitions);
Review comment:
@michaeljmarshall
Sorry, I missed this comment here.
Very thanks for you.
--
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] mattisonchao commented on pull request #14469: [Broker] Fix ``Future.join()`` causing deadlock.
Posted by GitBox <gi...@apache.org>.
mattisonchao commented on pull request #14469:
URL: https://github.com/apache/pulsar/pull/14469#issuecomment-1052850079
/pulsarbot rerun-failure-checks
--
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] mattisonchao commented on a change in pull request #14469: [Broker] Fix ``Future.join()`` causing deadlock.
Posted by GitBox <gi...@apache.org>.
mattisonchao commented on a change in pull request #14469:
URL: https://github.com/apache/pulsar/pull/14469#discussion_r815562752
##########
File path: pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PersistentTopicE2ETest.java
##########
@@ -97,6 +97,7 @@
@Override
protected void setup() throws Exception {
super.baseSetup();
+ conf.setBrokerDeleteInactivePartitionedTopicMetadataEnabled(true);
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
[GitHub] [pulsar] michaeljmarshall commented on a change in pull request #14469: [Broker] Fix ``Future.join()`` causing deadlock.
Posted by GitBox <gi...@apache.org>.
michaeljmarshall commented on a change in pull request #14469:
URL: https://github.com/apache/pulsar/pull/14469#discussion_r815276509
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java
##########
@@ -2256,42 +2256,42 @@ public void checkGC() {
return CompletableFuture.completedFuture(null);
}
TopicName topicName = TopicName.get(TopicName.get(topic).getPartitionedTopicName());
- try {
- PartitionedTopicResources partitionedTopicResources = getBrokerService().pulsar().getPulsarResources()
- .getNamespaceResources()
- .getPartitionedTopicResources();
- if (topicName.isPartitioned() && !partitionedTopicResources.partitionedTopicExists(topicName)) {
- return CompletableFuture.completedFuture(null);
- }
- CompletableFuture<Void> deleteMetadataFuture = new CompletableFuture<>();
- getBrokerService().fetchPartitionedTopicMetadataAsync(TopicName.get(topicName.getPartitionedTopicName()))
- .thenAccept((metadata -> {
- // make sure all sub partitions were deleted
- for (int i = 0; i < metadata.partitions; i++) {
- if (brokerService.getPulsar().getPulsarResources().getTopicResources()
- .persistentTopicExists(topicName.getPartition(i)).join()) {
- throw new UnsupportedOperationException();
- }
- }
- }))
- .thenAccept((res) -> partitionedTopicResources.deletePartitionedTopicAsync(topicName)
- .thenAccept((r) -> {
- deleteMetadataFuture.complete(null);
- }).exceptionally(ex -> {
- deleteMetadataFuture.completeExceptionally(ex.getCause());
- return null;
- }))
- .exceptionally((e) -> {
- if (!(e.getCause() instanceof UnsupportedOperationException)) {
- log.error("delete metadata fail", e);
- }
- deleteMetadataFuture.complete(null);
- return null;
- });
- return deleteMetadataFuture;
- } catch (Exception e) {
- return FutureUtil.failedFuture(e);
- }
+ PartitionedTopicResources partitionedTopicResources = getBrokerService().pulsar().getPulsarResources()
+ .getNamespaceResources()
+ .getPartitionedTopicResources();
+ return partitionedTopicResources.partitionedTopicExistsAsync(topicName)
+ .thenCompose(partitionedTopicExist -> {
+ if (!partitionedTopicExist) {
+ return CompletableFuture.completedFuture(null);
+ } else {
+ return getBrokerService()
+ .fetchPartitionedTopicMetadataAsync(TopicName.get(topicName.getPartitionedTopicName()))
+ .thenCompose((metadata -> {
+ List<CompletableFuture<Boolean>> persistentTopicExists =
+ new ArrayList<>(metadata.partitions);
+ for (int i = 0; i < metadata.partitions; i++) {
+ persistentTopicExists.add(brokerService.getPulsar()
+ .getPulsarResources().getTopicResources()
+ .persistentTopicExists(topicName.getPartition(i)));
+ }
+ return FutureUtil.waitForAll(persistentTopicExists)
+ .thenCompose(unused -> {
+ // make sure all sub partitions were deleted after all future complete
+ Optional<Boolean> anyExistPartition = persistentTopicExists.stream()
+ .map(CompletableFuture::join)
+ .filter(topicExist -> topicExist)
+ .findAny();
+ if (anyExistPartition.isPresent()) {
+ log.error("[{}] Delete metadata fail", topic);
+ return CompletableFuture.completedFuture(null);
Review comment:
Great catch. We should also try to log failure once (not twice) for a single failure. It looks like the current code might log a failure and then a success, which is just confusing.
--
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] mattisonchao commented on pull request #14469: [Broker] Fix ``Future.join()`` causing deadlock.
Posted by GitBox <gi...@apache.org>.
mattisonchao commented on pull request #14469:
URL: https://github.com/apache/pulsar/pull/14469#issuecomment-1053760860
@Jason918 @eolivelli
I have supplemented the metadata test on the original basis.
--
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] codelipenghui commented on a change in pull request #14469: [Broker] Fix ``Future.join()`` causing deadlock.
Posted by GitBox <gi...@apache.org>.
codelipenghui commented on a change in pull request #14469:
URL: https://github.com/apache/pulsar/pull/14469#discussion_r815254411
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java
##########
@@ -2256,42 +2256,42 @@ public void checkGC() {
return CompletableFuture.completedFuture(null);
}
TopicName topicName = TopicName.get(TopicName.get(topic).getPartitionedTopicName());
- try {
- PartitionedTopicResources partitionedTopicResources = getBrokerService().pulsar().getPulsarResources()
- .getNamespaceResources()
- .getPartitionedTopicResources();
- if (topicName.isPartitioned() && !partitionedTopicResources.partitionedTopicExists(topicName)) {
- return CompletableFuture.completedFuture(null);
- }
- CompletableFuture<Void> deleteMetadataFuture = new CompletableFuture<>();
- getBrokerService().fetchPartitionedTopicMetadataAsync(TopicName.get(topicName.getPartitionedTopicName()))
- .thenAccept((metadata -> {
- // make sure all sub partitions were deleted
- for (int i = 0; i < metadata.partitions; i++) {
- if (brokerService.getPulsar().getPulsarResources().getTopicResources()
- .persistentTopicExists(topicName.getPartition(i)).join()) {
- throw new UnsupportedOperationException();
- }
- }
- }))
- .thenAccept((res) -> partitionedTopicResources.deletePartitionedTopicAsync(topicName)
- .thenAccept((r) -> {
- deleteMetadataFuture.complete(null);
- }).exceptionally(ex -> {
- deleteMetadataFuture.completeExceptionally(ex.getCause());
- return null;
- }))
- .exceptionally((e) -> {
- if (!(e.getCause() instanceof UnsupportedOperationException)) {
- log.error("delete metadata fail", e);
- }
- deleteMetadataFuture.complete(null);
- return null;
- });
- return deleteMetadataFuture;
- } catch (Exception e) {
- return FutureUtil.failedFuture(e);
- }
+ PartitionedTopicResources partitionedTopicResources = getBrokerService().pulsar().getPulsarResources()
+ .getNamespaceResources()
+ .getPartitionedTopicResources();
+ return partitionedTopicResources.partitionedTopicExistsAsync(topicName)
+ .thenCompose(partitionedTopicExist -> {
+ if (!partitionedTopicExist) {
+ return CompletableFuture.completedFuture(null);
+ } else {
+ return getBrokerService()
+ .fetchPartitionedTopicMetadataAsync(TopicName.get(topicName.getPartitionedTopicName()))
+ .thenCompose((metadata -> {
+ List<CompletableFuture<Boolean>> persistentTopicExists =
+ new ArrayList<>(metadata.partitions);
+ for (int i = 0; i < metadata.partitions; i++) {
+ persistentTopicExists.add(brokerService.getPulsar()
+ .getPulsarResources().getTopicResources()
+ .persistentTopicExists(topicName.getPartition(i)));
+ }
+ return FutureUtil.waitForAll(persistentTopicExists)
+ .thenCompose(unused -> {
+ // make sure all sub partitions were deleted after all future complete
+ Optional<Boolean> anyExistPartition = persistentTopicExists.stream()
+ .map(CompletableFuture::join)
+ .filter(topicExist -> topicExist)
+ .findAny();
+ if (anyExistPartition.isPresent()) {
+ log.error("[{}] Delete metadata fail", topic);
+ return CompletableFuture.completedFuture(null);
Review comment:
We should return the exception here, otherwise, the `checkGC()` method will print log "Topic deleted successfully due to inactivity" but doesn't actually delete.
--
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] mattisonchao removed a comment on pull request #14469: [Broker] Fix ``Future.join()`` causing deadlock.
Posted by GitBox <gi...@apache.org>.
mattisonchao removed a comment on pull request #14469:
URL: https://github.com/apache/pulsar/pull/14469#issuecomment-1053937387
/pulsarbot rerun-failure-checks
--
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] michaeljmarshall commented on a change in pull request #14469: [Broker] Fix ``Future.join()`` causing deadlock.
Posted by GitBox <gi...@apache.org>.
michaeljmarshall commented on a change in pull request #14469:
URL: https://github.com/apache/pulsar/pull/14469#discussion_r816083093
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java
##########
@@ -2256,44 +2256,52 @@ public void checkGC() {
return CompletableFuture.completedFuture(null);
}
TopicName topicName = TopicName.get(TopicName.get(topic).getPartitionedTopicName());
- try {
- PartitionedTopicResources partitionedTopicResources = getBrokerService().pulsar().getPulsarResources()
- .getNamespaceResources()
- .getPartitionedTopicResources();
- if (topicName.isPartitioned() && !partitionedTopicResources.partitionedTopicExists(topicName)) {
- return CompletableFuture.completedFuture(null);
- }
- CompletableFuture<Void> deleteMetadataFuture = new CompletableFuture<>();
- getBrokerService().fetchPartitionedTopicMetadataAsync(TopicName.get(topicName.getPartitionedTopicName()))
- .thenAccept((metadata -> {
- // make sure all sub partitions were deleted
- for (int i = 0; i < metadata.partitions; i++) {
- if (brokerService.getPulsar().getPulsarResources().getTopicResources()
- .persistentTopicExists(topicName.getPartition(i)).join()) {
- throw new UnsupportedOperationException();
- }
- }
- }))
- .thenAccept((res) -> partitionedTopicResources.deletePartitionedTopicAsync(topicName)
- .thenAccept((r) -> {
- deleteMetadataFuture.complete(null);
- }).exceptionally(ex -> {
- deleteMetadataFuture.completeExceptionally(ex.getCause());
- return null;
- }))
- .exceptionally((e) -> {
- if (!(e.getCause() instanceof UnsupportedOperationException)) {
- log.error("delete metadata fail", e);
+ PartitionedTopicResources partitionedTopicResources = getBrokerService().pulsar().getPulsarResources()
+ .getNamespaceResources()
+ .getPartitionedTopicResources();
+ if (topicName.isPartitioned()) {
+ return partitionedTopicResources.partitionedTopicExistsAsync(topicName)
+ .thenCompose(partitionedTopicExist -> {
+ if (!partitionedTopicExist) {
+ return CompletableFuture.completedFuture(null);
+ } else {
+ return innerDeletePartitionedMetadata(topicName, partitionedTopicResources);
}
- deleteMetadataFuture.complete(null);
- return null;
});
- return deleteMetadataFuture;
- } catch (Exception e) {
- return FutureUtil.failedFuture(e);
+ } else {
+ return innerDeletePartitionedMetadata(topicName, partitionedTopicResources);
}
}
+ private CompletableFuture<Void> innerDeletePartitionedMetadata(TopicName topicName,
+ PartitionedTopicResources partitionedTopicResources)
+ {
+ return getBrokerService()
+ .fetchPartitionedTopicMetadataAsync(TopicName.get(topicName.getPartitionedTopicName()))
+ .thenCompose((metadata -> {
+ List<CompletableFuture<Boolean>> persistentTopicExists = new ArrayList<>(metadata.partitions);
Review comment:
I should have said we need to make sure the `ArrayList` is safely published from the thread where it is initialized to the thread where it is used. I think we should create the `List` by wrapping `ArrayList` with `Collections.unmodifiableList`. Then, we can ensure that the list is safely published from one thread to the next.
Here is a reference for understanding safe publication: https://shipilev.net/blog/2014/safe-public-construction/#_safe_publication.
--
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] michaeljmarshall commented on a change in pull request #14469: [Broker] Fix ``Future.join()`` causing deadlock.
Posted by GitBox <gi...@apache.org>.
michaeljmarshall commented on a change in pull request #14469:
URL: https://github.com/apache/pulsar/pull/14469#discussion_r816261247
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java
##########
@@ -2256,42 +2256,45 @@ public void checkGC() {
return CompletableFuture.completedFuture(null);
}
TopicName topicName = TopicName.get(TopicName.get(topic).getPartitionedTopicName());
- try {
- PartitionedTopicResources partitionedTopicResources = getBrokerService().pulsar().getPulsarResources()
- .getNamespaceResources()
- .getPartitionedTopicResources();
- if (topicName.isPartitioned() && !partitionedTopicResources.partitionedTopicExists(topicName)) {
- return CompletableFuture.completedFuture(null);
- }
- CompletableFuture<Void> deleteMetadataFuture = new CompletableFuture<>();
- getBrokerService().fetchPartitionedTopicMetadataAsync(TopicName.get(topicName.getPartitionedTopicName()))
- .thenAccept((metadata -> {
- // make sure all sub partitions were deleted
- for (int i = 0; i < metadata.partitions; i++) {
- if (brokerService.getPulsar().getPulsarResources().getTopicResources()
- .persistentTopicExists(topicName.getPartition(i)).join()) {
- throw new UnsupportedOperationException();
- }
- }
- }))
- .thenAccept((res) -> partitionedTopicResources.deletePartitionedTopicAsync(topicName)
- .thenAccept((r) -> {
- deleteMetadataFuture.complete(null);
- }).exceptionally(ex -> {
- deleteMetadataFuture.completeExceptionally(ex.getCause());
- return null;
- }))
- .exceptionally((e) -> {
- if (!(e.getCause() instanceof UnsupportedOperationException)) {
- log.error("delete metadata fail", e);
- }
- deleteMetadataFuture.complete(null);
- return null;
- });
- return deleteMetadataFuture;
- } catch (Exception e) {
- return FutureUtil.failedFuture(e);
- }
+ PartitionedTopicResources partitionedTopicResources = getBrokerService().pulsar().getPulsarResources()
+ .getNamespaceResources()
+ .getPartitionedTopicResources();
+ return partitionedTopicResources.partitionedTopicExistsAsync(topicName)
+ .thenCompose(partitionedTopicExist -> {
+ if (!partitionedTopicExist) {
+ return CompletableFuture.completedFuture(null);
+ } else {
+ return getBrokerService()
+ .fetchPartitionedTopicMetadataAsync(topicName)
+ .thenCompose((metadata -> {
+ List<CompletableFuture<Boolean>> persistentTopicExists =
+ new ArrayList<>(metadata.partitions);
+ for (int i = 0; i < metadata.partitions; i++) {
+ persistentTopicExists.add(brokerService.getPulsar()
+ .getPulsarResources().getTopicResources()
+ .persistentTopicExists(topicName.getPartition(i)));
+ }
+ return FutureUtil.waitForAll(persistentTopicExists)
+ .thenCompose(unused -> {
+ // make sure all sub partitions were deleted after all future complete
+ Optional<Boolean> anyExistPartition = persistentTopicExists.stream()
Review comment:
@mattisonchao - wrapping the `ArrayList` as I do here is sufficient. The basic premise is that we need to make sure the `ArrayList` is safely published to the `thenCompose` thread. There are places in Pulsar where we don't do this, and I think we need to address those as well in order to ensure correct access to data structures.
```suggestion
List<CompletableFuture<Boolean>> unmodifiablePersistentTopicExists =
Collections.unmodifiableList(persistentTopicExists);
return FutureUtil.waitForAll(unmodifiablePersistentTopicExists)
.thenCompose(unused -> {
// make sure all sub partitions were deleted after all future complete
Optional<Boolean> anyExistPartition = unmodifiablePersistentTopicExists
.stream()
```
--
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 a change in pull request #14469: [Broker] Fix ``Future.join()`` causing deadlock.
Posted by GitBox <gi...@apache.org>.
eolivelli commented on a change in pull request #14469:
URL: https://github.com/apache/pulsar/pull/14469#discussion_r814671876
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java
##########
@@ -2256,44 +2256,52 @@ public void checkGC() {
return CompletableFuture.completedFuture(null);
}
TopicName topicName = TopicName.get(TopicName.get(topic).getPartitionedTopicName());
- try {
- PartitionedTopicResources partitionedTopicResources = getBrokerService().pulsar().getPulsarResources()
- .getNamespaceResources()
- .getPartitionedTopicResources();
- if (topicName.isPartitioned() && !partitionedTopicResources.partitionedTopicExists(topicName)) {
- return CompletableFuture.completedFuture(null);
- }
- CompletableFuture<Void> deleteMetadataFuture = new CompletableFuture<>();
- getBrokerService().fetchPartitionedTopicMetadataAsync(TopicName.get(topicName.getPartitionedTopicName()))
- .thenAccept((metadata -> {
- // make sure all sub partitions were deleted
- for (int i = 0; i < metadata.partitions; i++) {
- if (brokerService.getPulsar().getPulsarResources().getTopicResources()
- .persistentTopicExists(topicName.getPartition(i)).join()) {
- throw new UnsupportedOperationException();
- }
- }
- }))
- .thenAccept((res) -> partitionedTopicResources.deletePartitionedTopicAsync(topicName)
- .thenAccept((r) -> {
- deleteMetadataFuture.complete(null);
- }).exceptionally(ex -> {
- deleteMetadataFuture.completeExceptionally(ex.getCause());
- return null;
- }))
- .exceptionally((e) -> {
- if (!(e.getCause() instanceof UnsupportedOperationException)) {
- log.error("delete metadata fail", e);
+ PartitionedTopicResources partitionedTopicResources = getBrokerService().pulsar().getPulsarResources()
+ .getNamespaceResources()
+ .getPartitionedTopicResources();
+ if (topicName.isPartitioned()) {
+ return partitionedTopicResources.partitionedTopicExistsAsync(topicName)
+ .thenCompose(partitionedTopicExist -> {
+ if (!partitionedTopicExist) {
+ return CompletableFuture.completedFuture(null);
+ } else {
+ return innerDeletePartitionedMetadata(topicName, partitionedTopicResources);
}
- deleteMetadataFuture.complete(null);
- return null;
});
- return deleteMetadataFuture;
- } catch (Exception e) {
- return FutureUtil.failedFuture(e);
+ } else {
+ return innerDeletePartitionedMetadata(topicName, partitionedTopicResources);
}
}
+ private CompletableFuture<Void> innerDeletePartitionedMetadata(TopicName topicName,
+ PartitionedTopicResources partitionedTopicResources)
+ {
+ return getBrokerService()
+ .fetchPartitionedTopicMetadataAsync(TopicName.get(topicName.getPartitionedTopicName()))
+ .thenCompose((metadata -> {
+ List<CompletableFuture<Boolean>> persistentTopicExists = new ArrayList<>(metadata.partitions);
+ for (int i = 0; i < metadata.partitions; i++) {
+ persistentTopicExists.add(brokerService.getPulsar()
+ .getPulsarResources().getTopicResources()
+ .persistentTopicExists(topicName.getPartition(i)));
+ }
+ return FutureUtil.waitForAll(persistentTopicExists)
+ .thenCompose(unused -> {
+ // make sure all sub partitions were deleted after all future complete
+ Optional<Boolean> anyExistPartition = persistentTopicExists.stream()
+ .map(CompletableFuture::join)
+ .filter(topicExist -> topicExist)
+ .findAny();
+ if (anyExistPartition.isPresent()) {
+ log.error("[{}] Delete metadata fail", topic);
Review comment:
why this is a error and we are returning a successful future ?
--
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] mattisonchao commented on a change in pull request #14469: [Broker] Fix ``Future.join()`` causing deadlock.
Posted by GitBox <gi...@apache.org>.
mattisonchao commented on a change in pull request #14469:
URL: https://github.com/apache/pulsar/pull/14469#discussion_r815241470
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java
##########
@@ -2256,44 +2256,52 @@ public void checkGC() {
return CompletableFuture.completedFuture(null);
}
TopicName topicName = TopicName.get(TopicName.get(topic).getPartitionedTopicName());
- try {
- PartitionedTopicResources partitionedTopicResources = getBrokerService().pulsar().getPulsarResources()
- .getNamespaceResources()
- .getPartitionedTopicResources();
- if (topicName.isPartitioned() && !partitionedTopicResources.partitionedTopicExists(topicName)) {
- return CompletableFuture.completedFuture(null);
- }
- CompletableFuture<Void> deleteMetadataFuture = new CompletableFuture<>();
- getBrokerService().fetchPartitionedTopicMetadataAsync(TopicName.get(topicName.getPartitionedTopicName()))
- .thenAccept((metadata -> {
- // make sure all sub partitions were deleted
- for (int i = 0; i < metadata.partitions; i++) {
- if (brokerService.getPulsar().getPulsarResources().getTopicResources()
- .persistentTopicExists(topicName.getPartition(i)).join()) {
- throw new UnsupportedOperationException();
- }
- }
- }))
- .thenAccept((res) -> partitionedTopicResources.deletePartitionedTopicAsync(topicName)
- .thenAccept((r) -> {
- deleteMetadataFuture.complete(null);
- }).exceptionally(ex -> {
- deleteMetadataFuture.completeExceptionally(ex.getCause());
- return null;
- }))
- .exceptionally((e) -> {
- if (!(e.getCause() instanceof UnsupportedOperationException)) {
- log.error("delete metadata fail", e);
+ PartitionedTopicResources partitionedTopicResources = getBrokerService().pulsar().getPulsarResources()
+ .getNamespaceResources()
+ .getPartitionedTopicResources();
+ if (topicName.isPartitioned()) {
+ return partitionedTopicResources.partitionedTopicExistsAsync(topicName)
+ .thenCompose(partitionedTopicExist -> {
+ if (!partitionedTopicExist) {
+ return CompletableFuture.completedFuture(null);
+ } else {
+ return innerDeletePartitionedMetadata(topicName, partitionedTopicResources);
}
- deleteMetadataFuture.complete(null);
- return null;
});
- return deleteMetadataFuture;
- } catch (Exception e) {
- return FutureUtil.failedFuture(e);
+ } else {
+ return innerDeletePartitionedMetadata(topicName, partitionedTopicResources);
}
}
+ private CompletableFuture<Void> innerDeletePartitionedMetadata(TopicName topicName,
+ PartitionedTopicResources partitionedTopicResources)
+ {
+ return getBrokerService()
+ .fetchPartitionedTopicMetadataAsync(TopicName.get(topicName.getPartitionedTopicName()))
+ .thenCompose((metadata -> {
+ List<CompletableFuture<Boolean>> persistentTopicExists = new ArrayList<>(metadata.partitions);
Review comment:
@michaeljmarshall
This variable doesn't seem to be shared with other threads, ( We are calling ``thenCompose`` not "thenComposeAsync") why does it need to be thread-safe?
Please give some explanation, Very thanks ~
--
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