Skip to content

[fix][branch-2.11] Fix delete system topic clean topic policy (#18823) #19835

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 5 commits into from
Mar 20, 2023
Merged

[fix][branch-2.11] Fix delete system topic clean topic policy (#18823) #19835

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
dd4ce57
[fix][broker] Fix delete system topic clean topic policy (#18823)
liangyepianzhou Dec 9, 2022
4471ece
fix
liangyepianzhou Mar 16, 2023
2faafe6
unused import
liangyepianzhou Mar 16, 2023
fa2d660
fix
liangyepianzhou Mar 17, 2023
c1bf08a
Fix test.
Technoboy- Mar 20, 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
47 changes: 38 additions & 9 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 @@ -73,6 +73,7 @@
import org.apache.pulsar.common.naming.NamespaceBundleSplitAlgorithm;
import org.apache.pulsar.common.naming.NamespaceBundles;
import org.apache.pulsar.common.naming.NamespaceName;
import org.apache.pulsar.common.naming.SystemTopicNames;
import org.apache.pulsar.common.naming.TopicDomain;
import org.apache.pulsar.common.naming.TopicName;
import org.apache.pulsar.common.policies.data.AuthAction;
Expand Down Expand Up @@ -310,22 +311,36 @@ protected void internalDeleteNamespace(AsyncResponse asyncResponse, boolean auth
// remove from owned namespace map and ephemeral node from ZK
final List<CompletableFuture<Void>> futures = Lists.newArrayList();
// remove system topics first.
Set<String> noPartitionedTopicPolicySystemTopic = new HashSet<>();
Set<String> partitionedTopicPolicySystemTopic = new HashSet<>();
if (!topics.isEmpty()) {
for (String topic : topics) {
try {
futures.add(pulsar().getAdminClient().topics().deleteAsync(topic, true, true));
if (SystemTopicNames.isTopicPoliciesSystemTopic(topic)) {
TopicName topicName = TopicName.get(topic);
if (topicName.isPartitioned()) {
partitionedTopicPolicySystemTopic.add(topicName.getPartitionedTopicName());
} else {
noPartitionedTopicPolicySystemTopic.add(topic);
}
} else {
futures.add(pulsar().getAdminClient().topics().deleteAsync(topic, true, true));
}
} catch (Exception ex) {
log.error("[{}] Failed to delete system topic {}", clientAppId(), topic, ex);
asyncResponse.resume(new RestException(Status.INTERNAL_SERVER_ERROR, ex));
return;
}
}
}
FutureUtil.waitForAll(futures).thenCompose(__ -> {
List<CompletableFuture<Void>> deleteBundleFutures = Lists.newArrayList();
NamespaceBundles bundles = pulsar().getNamespaceService().getNamespaceBundleFactory()
.getBundles(namespaceName);
for (NamespaceBundle bundle : bundles.getBundles()) {
FutureUtil.waitForAll(futures)
.thenCompose(ignore -> internalDeleteTopicsAsync(noPartitionedTopicPolicySystemTopic))
.thenCompose(ignore -> internalDeletePartitionedTopicsAsync(partitionedTopicPolicySystemTopic))
.thenCompose(__ -> {
List<CompletableFuture<Void>> deleteBundleFutures = Lists.newArrayList();
NamespaceBundles bundles = pulsar().getNamespaceService().getNamespaceBundleFactory()
.getBundles(namespaceName);
for (NamespaceBundle bundle : bundles.getBundles()) {
// check if the bundle is owned by any broker, if not then we do not need to delete the bundle
deleteBundleFutures.add(pulsar().getNamespaceService().getOwnerAsync(bundle).thenCompose(ownership -> {
if (ownership.isPresent()) {
Expand Down Expand Up @@ -475,13 +490,19 @@ protected void internalDeleteNamespaceForcefully(AsyncResponse asyncResponse, bo
Set<String> nonPartitionedTopics = new HashSet<>();
Set<String> allSystemTopics = new HashSet<>();
Set<String> allPartitionedSystemTopics = new HashSet<>();
Set<String> noPartitionedTopicPolicySystemTopic = new HashSet<>();
Set<String> partitionedTopicPolicySystemTopic = new HashSet<>();

for (String topic : topics) {
try {
TopicName topicName = TopicName.get(topic);
if (topicName.isPartitioned()) {
if (pulsar().getBrokerService().isSystemTopic(topicName)) {
allPartitionedSystemTopics.add(topicName.getPartitionedTopicName());
if (SystemTopicNames.isTopicPoliciesSystemTopic(topic)) {
partitionedTopicPolicySystemTopic.add(topicName.getPartitionedTopicName());
} else {
allPartitionedSystemTopics.add(topicName.getPartitionedTopicName());
}
continue;
}
String partitionedTopic = topicName.getPartitionedTopicName();
Expand All @@ -490,12 +511,17 @@ protected void internalDeleteNamespaceForcefully(AsyncResponse asyncResponse, bo
}
} else {
if (pulsar().getBrokerService().isSystemTopic(topicName)) {
allSystemTopics.add(topic);
if (SystemTopicNames.isTopicPoliciesSystemTopic(topic)) {
noPartitionedTopicPolicySystemTopic.add(topic);
} else {
allSystemTopics.add(topic);
}
continue;
}
nonPartitionedTopics.add(topic);
}
topicFutures.add(pulsar().getAdminClient().topics().deleteAsync(topic, true));
topicFutures.add(pulsar().getAdminClient().topics().deleteAsync(
topic, true, true));
} catch (Exception e) {
String errorMessage = String.format("Failed to force delete topic %s, "
+ "but the previous deletion command of partitioned-topics:%s "
Expand Down Expand Up @@ -524,6 +550,9 @@ protected void internalDeleteNamespaceForcefully(AsyncResponse asyncResponse, bo
.thenCompose((ignore) -> internalDeleteTopicsAsync(allSystemTopics))
.thenCompose((ignore) ->
internalDeletePartitionedTopicsAsync(allPartitionedSystemTopics))
.thenCompose(ignore ->
internalDeletePartitionedTopicsAsync(partitionedTopicPolicySystemTopic))
.thenCompose(ignore -> internalDeleteTopicsAsync(noPartitionedTopicPolicySystemTopic))
.handle((result, exception) -> {
if (exception != null) {
if (exception.getCause() instanceof PulsarAdminException) {
Expand Down
2 changes: 1 addition & 1 deletion pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1186,7 +1186,7 @@ private CompletableFuture<Void> delete(boolean failIfHasSubscriptions,
deleteTopicAuthenticationFuture.thenCompose(ignore -> deleteSchema())
.thenCompose(ignore -> {
if (!SystemTopicNames.isTopicPoliciesSystemTopic(topic)
&& brokerService.getPulsar().getConfiguration().isSystemTopicEnabled()) {
&& brokerService.getPulsar().getConfiguration().isSystemTopicEnabled()) {
return deleteTopicPolicies();
} else {
return CompletableFuture.completedFuture(null);
Expand Down
36 changes: 35 additions & 1 deletion pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/NamespacesTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,7 @@
*/
package org.apache.pulsar.broker.admin;

import static org.apache.pulsar.common.naming.NamespaceName.SYSTEM_NAMESPACE;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.Mockito.doNothing;
import static org.mockito.Mockito.doReturn;
Expand Down Expand Up @@ -57,11 +58,13 @@
import javax.ws.rs.core.Response.Status;
import javax.ws.rs.core.UriBuilder;
import javax.ws.rs.core.UriInfo;
import lombok.Cleanup;
import org.apache.bookkeeper.client.api.ReadHandle;
import org.apache.bookkeeper.mledger.LedgerOffloader;
import org.apache.bookkeeper.mledger.ManagedLedgerConfig;
import org.apache.bookkeeper.util.ZkUtils;
import org.apache.pulsar.broker.BrokerTestUtil;
import org.apache.pulsar.broker.PulsarService;
import org.apache.pulsar.broker.admin.v1.Namespaces;
import org.apache.pulsar.broker.admin.v1.PersistentTopics;
import org.apache.pulsar.broker.auth.MockedPulsarServiceBaseTest;
Expand All @@ -81,6 +84,7 @@
import org.apache.pulsar.client.api.ConsumerBuilder;
import org.apache.pulsar.client.api.Producer;
import org.apache.pulsar.client.api.PulsarClientException;
import org.apache.pulsar.client.api.Schema;
import org.apache.pulsar.client.api.SubscriptionType;
import org.apache.pulsar.common.api.proto.CommandSubscribe;
import org.apache.pulsar.common.naming.NamespaceBundle;
Expand All @@ -103,6 +107,7 @@
import org.apache.pulsar.common.policies.data.SubscribeRate;
import org.apache.pulsar.common.policies.data.TenantInfo;
import org.apache.pulsar.common.policies.data.TenantInfoImpl;
import org.apache.pulsar.common.policies.data.TopicPolicies;
import org.apache.pulsar.common.policies.data.impl.DispatchRateImpl;
import org.apache.pulsar.metadata.cache.impl.MetadataCacheImpl;
import org.apache.pulsar.metadata.impl.AbstractMetadataStore;
Expand Down Expand Up @@ -1938,7 +1943,7 @@ public void testFinallyDeleteSystemTopicWhenDeleteNamespace() throws Exception {
}

@Test
public void testNotClearTopicPolicesWhenDeleteSystemTopic() throws Exception {
public void testNotClearTopicPolicesWhenDeleteTopicPolicyTopic() throws Exception {
String namespace = this.testTenant + "/delete-systemTopic";
String topic = TopicName.get(TopicDomain.persistent.toString(), this.testTenant, "delete-systemTopic",
"testNotClearTopicPolicesWhenDeleteSystemTopic").toString();
Expand All @@ -1958,4 +1963,33 @@ public void testNotClearTopicPolicesWhenDeleteSystemTopic() throws Exception {
// 4. delete the policies topic and the topic wil not to clear topic polices
admin.topics().delete(namespace + "/" + SystemTopicNames.NAMESPACE_EVENTS_LOCAL_NAME, true);
}
@Test
public void testDeleteTopicPolicyWhenDeleteSystemTopic() throws Exception {
conf.setTopicLevelPoliciesEnabled(true);
conf.setSystemTopicEnabled(true);
Field field = PulsarService.class.getDeclaredField("topicPoliciesService");
field.setAccessible(true);
field.set(pulsar, new SystemTopicBasedTopicPoliciesService(pulsar));

String systemTopic = SYSTEM_NAMESPACE.toString() + "/" + "testDeleteTopicPolicyWhenDeleteSystemTopic";
admin.tenants().createTenant(SYSTEM_NAMESPACE.getTenant(),
new TenantInfoImpl(Set.of("role1", "role2"), Set.of("use", "usc", "usw")));

admin.namespaces().createNamespace(SYSTEM_NAMESPACE.toString());
@Cleanup
Producer<String> producer = pulsarClient.newProducer(Schema.STRING)
.topic(systemTopic).create();
admin.topicPolicies().setMaxConsumers(systemTopic, 5);

Integer maxConsumerPerTopic = pulsar
.getTopicPoliciesService()
.getTopicPoliciesBypassCacheAsync(TopicName.get(systemTopic)).get()
.getMaxConsumerPerTopic();

assertEquals(maxConsumerPerTopic, Integer.valueOf(5));
admin.topics().delete(systemTopic, true);
TopicPolicies topicPolicies = pulsar.getTopicPoliciesService()
.getTopicPoliciesBypassCacheAsync(TopicName.get(systemTopic)).get(5, TimeUnit.SECONDS);
assertNull(topicPolicies);
}
}