Skip to content

[fix][broker] Fix delete namespace fail by a In-flight topic #19374

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
Technoboy- merged 15 commits into from
Feb 17, 2023
Merged

[fix][broker] Fix delete namespace fail by a In-flight topic #19374

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
15 commits
Select commit Hold shift + click to select a range
1b6af5a
Using retry to fix problem
mattisonchao Feb 1, 2023
2ab44b9
Remove useless test code
mattisonchao Feb 1, 2023
67758ca
Update pulsar-broker/src/main/java/org/apache/pulsar/broker/service/p…
mattisonchao Feb 1, 2023
1d0eb64
Fix checkstyle
mattisonchao Feb 1, 2023
ef7b10f
Using retry to instead backoff.
mattisonchao Feb 15, 2023
24ed229
Fix checkstyle
mattisonchao Feb 15, 2023
2b810fe
Remove reduntant logic
mattisonchao Feb 15, 2023
3124e58
Fix checkstyle
mattisonchao Feb 15, 2023
4681d5b
Remove test invocations
mattisonchao Feb 15, 2023
522aa6e
Add the log
mattisonchao Feb 15, 2023
f8db021
Add forgotten -1
mattisonchao Feb 15, 2023
288374f
Apply comments
mattisonchao Feb 15, 2023
4d55e1e
Apply comment
mattisonchao Feb 15, 2023
3fa7af4
Change the logic to fix the test
mattisonchao Feb 17, 2023
9fff22f
Revert useless code
mattisonchao Feb 17, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
183 changes: 113 additions & 70 deletions pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/NamespacesBase.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -44,6 +44,7 @@
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Function;
import java.util.stream.Collectors;
import javax.annotation.Nonnull;
import javax.ws.rs.WebApplicationException;
import javax.ws.rs.container.AsyncResponse;
import javax.ws.rs.core.Response;
Expand Down Expand Up @@ -115,6 +116,7 @@
import org.apache.pulsar.metadata.api.MetadataStoreException;
import org.apache.pulsar.metadata.api.MetadataStoreException.BadVersionException;
import org.apache.pulsar.metadata.api.MetadataStoreException.NotFoundException;
import org.apache.zookeeper.KeeperException;

@Slf4j
public abstract class NamespacesBase extends AdminResource {
Expand Down Expand Up @@ -204,78 +206,94 @@ protected CompletableFuture<List<String>> internalGetNonPersistentTopics(Policie
});
}

@SuppressWarnings("unchecked")
protected CompletableFuture<Void> internalDeleteNamespaceAsync(boolean force) {
CompletableFuture<Policies> preconditionCheck = precheckWhenDeleteNamespace(namespaceName, force);
return preconditionCheck
/**
* Delete the namespace and retry to resolve some topics that were not created successfully(in metadata)
* during the deletion.
*/
protected @Nonnull CompletableFuture<Void> internalDeleteNamespaceAsync(boolean force) {
final CompletableFuture<Void> future = new CompletableFuture<>();
internalRetryableDeleteNamespaceAsync0(force, 5, future);
return future;
}
private void internalRetryableDeleteNamespaceAsync0(boolean force, int retryTimes,
@Nonnull CompletableFuture<Void> callback) {
precheckWhenDeleteNamespace(namespaceName, force)
.thenCompose(policies -> {
final CompletableFuture<List<String>> topicsFuture;
if (policies == null || CollectionUtils.isEmpty(policies.replication_clusters)){
return pulsar().getNamespaceService().getListOfPersistentTopics(namespaceName);
}
return pulsar().getNamespaceService().getFullListOfTopics(namespaceName);
})
.thenCompose(allTopics -> pulsar().getNamespaceService().getFullListOfPartitionedTopic(namespaceName)
.thenCompose(allPartitionedTopics -> {
List<List<String>> topicsSum = new ArrayList<>(2);
topicsSum.add(allTopics);
topicsSum.add(allPartitionedTopics);
return CompletableFuture.completedFuture(topicsSum);
}))
.thenCompose(topics -> {
List<String> allTopics = topics.get(0);
ArrayList<String> allUserCreatedTopics = new ArrayList<>();
List<String> allPartitionedTopics = topics.get(1);
ArrayList<String> allUserCreatedPartitionTopics = new ArrayList<>();
boolean hasNonSystemTopic = false;
List<String> allSystemTopics = new ArrayList<>();
List<String> allPartitionedSystemTopics = new ArrayList<>();
List<String> topicPolicy = new ArrayList<>();
List<String> partitionedTopicPolicy = new ArrayList<>();
for (String topic : allTopics) {
if (!pulsar().getBrokerService().isSystemTopic(TopicName.get(topic))) {
hasNonSystemTopic = true;
allUserCreatedTopics.add(topic);
} else {
if (SystemTopicNames.isTopicPoliciesSystemTopic(topic)) {
topicPolicy.add(topic);
} else {
allSystemTopics.add(topic);
}
}
}
for (String topic : allPartitionedTopics) {
if (!pulsar().getBrokerService().isSystemTopic(TopicName.get(topic))) {
hasNonSystemTopic = true;
allUserCreatedPartitionTopics.add(topic);
} else {
if (SystemTopicNames.isTopicPoliciesSystemTopic(topic)) {
partitionedTopicPolicy.add(topic);
} else {
allPartitionedSystemTopics.add(topic);
}
}
}
if (!force) {
if (hasNonSystemTopic) {
throw new RestException(Status.CONFLICT, "Cannot delete non empty namespace");
}
topicsFuture = pulsar().getNamespaceService().getListOfPersistentTopics(namespaceName);
} else {
topicsFuture = pulsar().getNamespaceService().getFullListOfTopics(namespaceName);
}
return namespaceResources().setPoliciesAsync(namespaceName, old -> {
old.deleted = true;
return old;
}).thenCompose(ignore -> {
return internalDeleteTopicsAsync(allUserCreatedTopics);
}).thenCompose(ignore -> {
return internalDeletePartitionedTopicsAsync(allUserCreatedPartitionTopics);
}).thenCompose(ignore -> {
return internalDeleteTopicsAsync(allSystemTopics);
}).thenCompose(ignore__ -> {
return internalDeletePartitionedTopicsAsync(allPartitionedSystemTopics);
}).thenCompose(ignore -> {
return internalDeleteTopicsAsync(topicPolicy);
}).thenCompose(ignore__ -> {
return internalDeletePartitionedTopicsAsync(partitionedTopicPolicy);
});
return topicsFuture.thenCompose(allTopics ->
pulsar().getNamespaceService().getFullListOfPartitionedTopic(namespaceName)
.thenCompose(allPartitionedTopics -> {
List<List<String>> topicsSum = new ArrayList<>(2);
topicsSum.add(allTopics);
topicsSum.add(allPartitionedTopics);
return CompletableFuture.completedFuture(topicsSum);
}))
.thenCompose(topics -> {
List<String> allTopics = topics.get(0);
ArrayList<String> allUserCreatedTopics = new ArrayList<>();
List<String> allPartitionedTopics = topics.get(1);
ArrayList<String> allUserCreatedPartitionTopics = new ArrayList<>();
boolean hasNonSystemTopic = false;
List<String> allSystemTopics = new ArrayList<>();
List<String> allPartitionedSystemTopics = new ArrayList<>();
List<String> topicPolicy = new ArrayList<>();
List<String> partitionedTopicPolicy = new ArrayList<>();
for (String topic : allTopics) {
if (!pulsar().getBrokerService().isSystemTopic(TopicName.get(topic))) {
hasNonSystemTopic = true;
allUserCreatedTopics.add(topic);
} else {
if (SystemTopicNames.isTopicPoliciesSystemTopic(topic)) {
topicPolicy.add(topic);
} else {
allSystemTopics.add(topic);
}
}
}
for (String topic : allPartitionedTopics) {
if (!pulsar().getBrokerService().isSystemTopic(TopicName.get(topic))) {
hasNonSystemTopic = true;
allUserCreatedPartitionTopics.add(topic);
} else {
if (SystemTopicNames.isTopicPoliciesSystemTopic(topic)) {
partitionedTopicPolicy.add(topic);
} else {
allPartitionedSystemTopics.add(topic);
}
}
}
if (!force) {
if (hasNonSystemTopic) {
throw new RestException(Status.CONFLICT, "Cannot delete non empty namespace");
}
}
final CompletableFuture<Void> markDeleteFuture;
if (policies != null && policies.deleted) {
markDeleteFuture = CompletableFuture.completedFuture(null);
} else {
markDeleteFuture = namespaceResources().setPoliciesAsync(namespaceName, old -> {
old.deleted = true;
return old;
});
}
return markDeleteFuture.thenCompose(__ ->
internalDeleteTopicsAsync(allUserCreatedTopics))
.thenCompose(ignore ->
internalDeletePartitionedTopicsAsync(allUserCreatedPartitionTopics))
.thenCompose(ignore ->
internalDeleteTopicsAsync(allSystemTopics))
.thenCompose(ignore ->
internalDeletePartitionedTopicsAsync(allPartitionedSystemTopics))
.thenCompose(ignore ->
internalDeleteTopicsAsync(topicPolicy))
.thenCompose(ignore ->
internalDeletePartitionedTopicsAsync(partitionedTopicPolicy));
});
})
.thenCompose(ignore -> pulsar().getNamespaceService()
.getNamespaceBundleFactory().getBundlesAsync(namespaceName))
Expand All @@ -299,7 +317,32 @@ protected CompletableFuture<Void> internalDeleteNamespaceAsync(boolean force) {
return CompletableFuture.completedFuture(null);
})
).collect(Collectors.toList())))
.thenCompose(ignore -> internalClearZkSources());
.thenCompose(ignore -> internalClearZkSources())
.whenComplete((result, error) -> {
if (error != null) {
final Throwable rc = FutureUtil.unwrapCompletionException(error);
if (rc instanceof MetadataStoreException) {
if (rc.getCause() != null && rc.getCause() instanceof KeeperException.NotEmptyException) {
log.info("[{}] There are in-flight topics created during the namespace deletion, "
+ "retry to delete the namespace again.", namespaceName);
final int next = retryTimes - 1;
if (next > 0) {
// async recursive
internalRetryableDeleteNamespaceAsync0(force, next, callback);
} else {
callback.completeExceptionally(
new RestException(Status.CONFLICT, "The broker still have in-flight topics"
+ " created during namespace deletion, please try again."));
// drop out recursive
}
return;
}
}
callback.completeExceptionally(error);
return;
}
callback.complete(result);
});
}

private CompletableFuture<Void> internalDeletePartitionedTopicsAsync(List<String> topicNames) {
Expand Down
7 changes: 0 additions & 7 deletions pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v1/Namespaces.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -48,7 +48,6 @@
import org.apache.pulsar.broker.admin.impl.NamespacesBase;
import org.apache.pulsar.broker.service.BrokerServiceException;
import org.apache.pulsar.broker.web.RestException;
import org.apache.pulsar.client.admin.PulsarAdminException;
import org.apache.pulsar.common.api.proto.CommandGetTopicsOfNamespace.Mode;
import org.apache.pulsar.common.naming.NamespaceBundleSplitAlgorithm;
import org.apache.pulsar.common.naming.NamespaceName;
Expand Down Expand Up @@ -250,12 +249,6 @@ public void deleteNamespace(@Suspended final AsyncResponse asyncResponse, @PathP
asyncResponse.resume(Response.noContent().build());
})
.exceptionally(ex -> {
Throwable cause = FutureUtil.unwrapCompletionException(ex);
if (cause instanceof PulsarAdminException.ConflictException) {
log.info("[{}] There are new topics created during the namespace deletion, "
+ "retry to delete the namespace again.", namespaceName);
pulsar().getExecutor().execute(() -> internalDeleteNamespaceAsync(force));
}
if (!isRedirectException(ex)) {
log.error("[{}] Failed to delete namespace {}", clientAppId(), namespaceName, ex);
}
Expand Down
7 changes: 0 additions & 7 deletions pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v2/Namespaces.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -51,7 +51,6 @@
import org.apache.pulsar.broker.admin.impl.OffloaderObjectsScannerUtils;
import org.apache.pulsar.broker.service.BrokerServiceException;
import org.apache.pulsar.broker.web.RestException;
import org.apache.pulsar.client.admin.PulsarAdminException;
import org.apache.pulsar.client.api.SubscriptionType;
import org.apache.pulsar.common.api.proto.CommandGetTopicsOfNamespace.Mode;
import org.apache.pulsar.common.naming.NamespaceName;
Expand Down Expand Up @@ -199,12 +198,6 @@ public void deleteNamespace(@Suspended final AsyncResponse asyncResponse, @PathP
asyncResponse.resume(Response.noContent().build());
})
.exceptionally(ex -> {
Throwable cause = FutureUtil.unwrapCompletionException(ex);
if (cause instanceof PulsarAdminException.ConflictException) {
log.info("[{}] There are new topics created during the namespace deletion, "
+ "retry to delete the namespace again.", namespaceName);
pulsar().getExecutor().execute(() -> internalDeleteNamespaceAsync(force));
}
if (!isRedirectException(ex)) {
log.error("[{}] Failed to delete namespace {}", clientAppId(), namespaceName, ex);
}
Expand Down
3 changes: 0 additions & 3 deletions pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractTopic.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -246,9 +246,6 @@ protected void updateTopicPolicyByNamespacePolicy(Policies namespacePolicies) {
if (log.isDebugEnabled()) {
log.debug("[{}]updateTopicPolicyByNamespacePolicy,data={}", topic, namespacePolicies);
}
if (namespacePolicies.deleted) {
return;
}
topicPolicies.getRetentionPolicies().updateNamespaceValue(namespacePolicies.retention_policies);
topicPolicies.getCompactionThreshold().updateNamespaceValue(namespacePolicies.compaction_threshold);
topicPolicies.getReplicationClusters().updateNamespaceValue(
Expand Down
6 changes: 6 additions & 0 deletions pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApi2Test.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -46,6 +46,7 @@
import java.util.Set;
import java.util.TreeSet;
import java.util.UUID;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;
import javax.ws.rs.NotAcceptableException;
import javax.ws.rs.core.Response.Status;
Expand Down Expand Up @@ -1654,6 +1655,11 @@ public void testDeleteNamespaceWithTopicPolicies() throws Exception {
// verify namespace can be deleted even without topic policy events
admin.namespaces().deleteNamespace(namespace, true);

Awaitility.await().untilAsserted(() -> {
final CompletableFuture<Optional<Topic>> eventTopicFuture =
pulsar.getBrokerService().getTopics().get("persistent://test-tenant/test-ns2/__change_events");
assertNull(eventTopicFuture);
});
admin.namespaces().createNamespace(namespace, Set.of("test"));
// create topic
String topic = namespace + "/test-topic2";
Expand Down