You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by pe...@apache.org on 2022/03/25 05:48:51 UTC
[pulsar] 01/14: Fix normal topic named ends with `healthcheck` becomes system topic issue. (#14671)
This is an automated email from the ASF dual-hosted git repository.
penghui pushed a commit to branch branch-2.9
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit aa66cb8c129d71fbb610f2a8528c7d81dffbbadb
Author: Jiwei Guo <te...@apache.org>
AuthorDate: Mon Mar 14 21:31:43 2022 +0800
Fix normal topic named ends with `healthcheck` becomes system topic issue. (#14671)
(cherry picked from commit 3afd1e673304bb2dfb2c42f74a5895a0aa12cfc1)
---
.../apache/pulsar/broker/admin/AdminResource.java | 7 +--
.../pulsar/broker/admin/impl/NamespacesBase.java | 7 +--
.../pulsar/broker/service/AbstractTopic.java | 3 +-
.../pulsar/broker/service/BrokerService.java | 33 +++++++++--
.../pulsar/broker/systopic/SystemTopicClient.java | 25 ---------
.../pulsar/broker/service/BrokerServiceTest.java | 34 +++++++++++
.../broker/service/InactiveTopicDeleteTest.java | 65 +++++++++-------------
.../NamespaceEventsSystemTopicServiceTest.java | 7 ++-
.../broker/systopic/SystemTopicClientTest.java | 47 ----------------
9 files changed, 98 insertions(+), 130 deletions(-)
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/AdminResource.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/AdminResource.java
index 8d7de94..3598e03 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/AdminResource.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/AdminResource.java
@@ -39,7 +39,6 @@ import org.apache.bookkeeper.mledger.ManagedLedgerException;
import org.apache.pulsar.broker.PulsarService;
import org.apache.pulsar.broker.ServiceConfiguration;
import org.apache.pulsar.broker.service.BrokerServiceException;
-import org.apache.pulsar.broker.systopic.SystemTopicClient;
import org.apache.pulsar.broker.web.PulsarWebResource;
import org.apache.pulsar.broker.web.RestException;
import org.apache.pulsar.client.admin.PulsarAdminException;
@@ -586,12 +585,12 @@ public abstract class AdminResource extends PulsarWebResource {
}
// new create check
- if (maxTopicsPerNamespace > 0 && !SystemTopicClient.isSystemTopic(topicName)) {
+ if (maxTopicsPerNamespace > 0 && !pulsar().getBrokerService().isSystemTopic(topicName)) {
List<String> partitionedTopics = getTopicPartitionList(TopicDomain.persistent);
// exclude created system topic
long topicsCount =
- partitionedTopics.stream().filter(t -> !SystemTopicClient.isSystemTopic(TopicName.get(t)))
- .count();
+ partitionedTopics.stream().filter(t ->
+ !pulsar().getBrokerService().isSystemTopic(TopicName.get(t))).count();
if (topicsCount + numPartitions > maxTopicsPerNamespace) {
log.error("[{}] Failed to create partitioned topic {}, "
+ "exceed maximum number of topics in namespace", clientAppId(), topicName);
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/NamespacesBase.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/NamespacesBase.java
index 310c0d9..11d7b39 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/NamespacesBase.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/NamespacesBase.java
@@ -58,7 +58,6 @@ import org.apache.pulsar.broker.service.Subscription;
import org.apache.pulsar.broker.service.Topic;
import org.apache.pulsar.broker.service.persistent.PersistentReplicator;
import org.apache.pulsar.broker.service.persistent.PersistentTopic;
-import org.apache.pulsar.broker.systopic.SystemTopicClient;
import org.apache.pulsar.broker.web.RestException;
import org.apache.pulsar.client.admin.PulsarAdminException;
import org.apache.pulsar.client.api.SubscriptionType;
@@ -237,7 +236,7 @@ public abstract class NamespacesBase extends AdminResource {
}
boolean hasNonSystemTopic = false;
for (String topic : topics) {
- if (!SystemTopicClient.isSystemTopic(TopicName.get(topic))) {
+ if (!pulsar().getBrokerService().isSystemTopic(TopicName.get(topic))) {
hasNonSystemTopic = true;
break;
}
@@ -1892,14 +1891,14 @@ public abstract class NamespacesBase extends AdminResource {
}
for (Topic topic : topicList) {
if (topic instanceof PersistentTopic
- && !SystemTopicClient.isSystemTopic(TopicName.get(topic.getName()))) {
+ && !pulsar().getBrokerService().isSystemTopic(TopicName.get(topic.getName()))) {
futures.add(((PersistentTopic) topic).clearBacklog(subscription));
}
}
} else {
for (Topic topic : topicList) {
if (topic instanceof PersistentTopic
- && !SystemTopicClient.isSystemTopic(TopicName.get(topic.getName()))) {
+ && !pulsar().getBrokerService().isSystemTopic(TopicName.get(topic.getName()))) {
futures.add(((PersistentTopic) topic).clearBacklog());
}
}
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractTopic.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractTopic.java
index 620d35e..cc29202 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractTopic.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractTopic.java
@@ -47,7 +47,6 @@ import org.apache.pulsar.broker.service.schema.BookkeeperSchemaStorage;
import org.apache.pulsar.broker.service.schema.SchemaRegistryService;
import org.apache.pulsar.broker.service.schema.exceptions.IncompatibleSchemaException;
import org.apache.pulsar.broker.stats.prometheus.metrics.Summary;
-import org.apache.pulsar.broker.systopic.SystemTopicClient;
import org.apache.pulsar.common.naming.TopicName;
import org.apache.pulsar.common.policies.data.InactiveTopicDeleteMode;
import org.apache.pulsar.common.policies.data.InactiveTopicPolicies;
@@ -540,7 +539,7 @@ public abstract class AbstractTopic implements Topic {
}
protected void setSchemaCompatibilityStrategy(Policies policies) {
- if (SystemTopicClient.isSystemTopic(TopicName.get(this.topic))) {
+ if (isSystemTopic()) {
schemaCompatibilityStrategy =
brokerService.pulsar().getConfig().getSystemTopicSchemaCompatibilityStrategy();
return;
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java
index 1c45202..939a389 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java
@@ -117,7 +117,7 @@ import org.apache.pulsar.broker.service.persistent.SystemTopic;
import org.apache.pulsar.broker.stats.ClusterReplicationMetrics;
import org.apache.pulsar.broker.stats.prometheus.metrics.ObserverGauge;
import org.apache.pulsar.broker.stats.prometheus.metrics.Summary;
-import org.apache.pulsar.broker.systopic.SystemTopicClient;
+import org.apache.pulsar.broker.transaction.pendingack.impl.MLPendingAckStore;
import org.apache.pulsar.broker.validator.BindAddressValidator;
import org.apache.pulsar.client.admin.PulsarAdmin;
import org.apache.pulsar.client.admin.PulsarAdminBuilder;
@@ -129,6 +129,7 @@ import org.apache.pulsar.client.impl.conf.ClientConfigurationData;
import org.apache.pulsar.common.allocator.PulsarByteBufAllocator;
import org.apache.pulsar.common.configuration.BindAddress;
import org.apache.pulsar.common.configuration.FieldContext;
+import org.apache.pulsar.common.events.EventsTopicNames;
import org.apache.pulsar.common.intercept.AppendIndexMetadataInterceptor;
import org.apache.pulsar.common.intercept.BrokerEntryMetadataInterceptor;
import org.apache.pulsar.common.intercept.BrokerEntryMetadataUtils;
@@ -2671,8 +2672,30 @@ public class BrokerService implements Closeable {
log.debug("No autoSubscriptionCreateOverride policy found for {}", topicName);
return null;
}
- private boolean isSystemTopic(String topic) {
- return SystemTopicClient.isSystemTopic(TopicName.get(topic));
+
+ public boolean isSystemTopic(String topic) {
+ return isSystemTopic(TopicName.get(topic));
+ }
+
+ public boolean isSystemTopic(TopicName topicName) {
+ if (topicName.getNamespaceObject().equals(NamespaceName.SYSTEM_NAMESPACE)
+ || topicName.getNamespaceObject().equals(pulsar.getHeartbeatNamespaceV2())) {
+ return true;
+ }
+
+ TopicName nonePartitionedTopicName = TopicName.get(topicName.getPartitionedTopicName());
+
+ // event topic
+ if (EventsTopicNames.checkTopicIsEventsNames(nonePartitionedTopicName)) {
+ return true;
+ }
+
+ String localName = nonePartitionedTopicName.getLocalName();
+ // transaction pending ack topic
+ if (StringUtils.endsWith(localName, MLPendingAckStore.PENDING_ACK_STORE_SUFFIX)) {
+ return true;
+ }
+ return false;
}
/**
@@ -2707,13 +2730,13 @@ public class BrokerService implements Closeable {
int maxTopicsPerNamespace = optPolicies.map(p -> p.max_topics_per_namespace)
.orElse(pulsar.getConfig().getMaxTopicsPerNamespace());
- if (maxTopicsPerNamespace > 0 && !SystemTopicClient.isSystemTopic(topicName)) {
+ if (maxTopicsPerNamespace > 0 && !isSystemTopic(topicName)) {
return pulsar().getPulsarResources().getTopicResources()
.getExistingPartitions(topicName)
.thenCompose(topics -> {
// exclude created system topic
long topicsCount = topics.stream()
- .filter(t -> !SystemTopicClient.isSystemTopic(TopicName.get(t)))
+ .filter(t -> !isSystemTopic(TopicName.get(t)))
.count();
if (topicsCount + numPartitions > maxTopicsPerNamespace) {
log.error("Failed to create persistent topic {}, "
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/systopic/SystemTopicClient.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/systopic/SystemTopicClient.java
index 122bd9f..33bfc59 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/systopic/SystemTopicClient.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/systopic/SystemTopicClient.java
@@ -21,13 +21,9 @@ package org.apache.pulsar.broker.systopic;
import java.io.IOException;
import java.util.List;
import java.util.concurrent.CompletableFuture;
-import org.apache.commons.lang3.StringUtils;
-import org.apache.pulsar.broker.admin.impl.BrokersBase;
-import org.apache.pulsar.broker.transaction.pendingack.impl.MLPendingAckStore;
import org.apache.pulsar.client.api.Message;
import org.apache.pulsar.client.api.MessageId;
import org.apache.pulsar.client.api.PulsarClientException;
-import org.apache.pulsar.common.events.EventsTopicNames;
import org.apache.pulsar.common.naming.TopicName;
/**
@@ -188,25 +184,4 @@ public interface SystemTopicClient<T> {
*/
SystemTopicClient<T> getSystemTopic();
}
-
- static boolean isSystemTopic(TopicName topicName) {
- TopicName nonePartitionedTopicName = TopicName.get(topicName.getPartitionedTopicName());
-
- // event topic
- if (EventsTopicNames.checkTopicIsEventsNames(nonePartitionedTopicName)) {
- return true;
- }
-
- String localName = nonePartitionedTopicName.getLocalName();
- // transaction pending ack topic
- if (StringUtils.endsWith(localName, MLPendingAckStore.PENDING_ACK_STORE_SUFFIX)) {
- return true;
- }
- // health check topic
- if (StringUtils.endsWith(localName, BrokersBase.HEALTH_CHECK_TOPIC_SUFFIX)){
- return true;
- }
- return false;
- }
-
}
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerServiceTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerServiceTest.java
index af8ddb5..c7ffef1 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerServiceTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerServiceTest.java
@@ -18,10 +18,13 @@
*/
package org.apache.pulsar.broker.service;
+import static org.apache.pulsar.common.naming.TopicName.TRANSACTION_COORDINATOR_ASSIGN;
+import static org.apache.pulsar.common.naming.TopicName.TRANSACTION_COORDINATOR_LOG;
import static org.mockito.Mockito.any;
import static org.mockito.Mockito.doReturn;
import static org.mockito.Mockito.spy;
import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertFalse;
import static org.testng.Assert.assertNotNull;
import static org.testng.Assert.assertNull;
import static org.testng.Assert.assertTrue;
@@ -66,6 +69,7 @@ import org.apache.http.HttpResponse;
import org.apache.http.client.HttpClient;
import org.apache.http.client.methods.HttpGet;
import org.apache.http.impl.client.HttpClientBuilder;
+import org.apache.pulsar.broker.namespace.NamespaceService;
import org.apache.pulsar.broker.service.BrokerServiceException.PersistenceException;
import org.apache.pulsar.broker.service.persistent.PersistentTopic;
import org.apache.pulsar.broker.stats.prometheus.PrometheusRawMetricsProvider;
@@ -723,6 +727,7 @@ public class BrokerServiceTest extends BrokerTestBase {
} finally {
pulsarClient.close();
}
+ resetState();
}
@SuppressWarnings("deprecation")
@@ -1275,4 +1280,33 @@ public class BrokerServiceTest extends BrokerTestBase {
getStatsThread.join();
}
}
+
+ @Test
+ public void testIsSystemTopic() {
+ BrokerService brokerService = pulsar.getBrokerService();
+ assertFalse(brokerService.isSystemTopic(TopicName.get("test")));
+ assertFalse(brokerService.isSystemTopic(TopicName.get("public/default/test")));
+ assertFalse(brokerService.isSystemTopic(TopicName.get("healthcheck")));
+ assertFalse(brokerService.isSystemTopic(TopicName.get("public/default/healthcheck")));
+ assertFalse(brokerService.isSystemTopic(TopicName.get("persistent://public/default/test")));
+ assertFalse(brokerService.isSystemTopic(TopicName.get("non-persistent://public/default/test")));
+
+ assertTrue(brokerService.isSystemTopic(TopicName.get("__change_events")));
+ assertTrue(brokerService.isSystemTopic(TopicName.get("__change_events-partition-0")));
+ assertTrue(brokerService.isSystemTopic(TopicName.get("__change_events-partition-1")));
+ assertTrue(brokerService.isSystemTopic(TopicName.get("__transaction_buffer_snapshot")));
+ assertTrue(brokerService.isSystemTopic(TopicName.get("__transaction_buffer_snapshot-partition-0")));
+ assertTrue(brokerService.isSystemTopic(TopicName.get("__transaction_buffer_snapshot-partition-1")));
+ assertTrue(brokerService.isSystemTopic(TopicName
+ .get("topicxxx-partition-0-multiTopicsReader-f433329d68__transaction_pending_ack")));
+ assertTrue(brokerService.isSystemTopic(
+ TopicName.get("topicxxx-multiTopicsReader-f433329d68__transaction_pending_ack")));
+
+ assertTrue(brokerService.isSystemTopic(TRANSACTION_COORDINATOR_ASSIGN));
+ assertTrue(brokerService.isSystemTopic(TRANSACTION_COORDINATOR_LOG));
+ NamespaceName heartbeatNamespaceV1 = NamespaceService.getHeartbeatNamespace(pulsar.getAdvertisedAddress(), pulsar.getConfig());
+ NamespaceName heartbeatNamespaceV2 = NamespaceService.getHeartbeatNamespaceV2(pulsar.getAdvertisedAddress(), pulsar.getConfig());
+ assertTrue(brokerService.isSystemTopic("persistent://" + heartbeatNamespaceV1.toString() + "/healthcheck"));
+ assertTrue(brokerService.isSystemTopic(heartbeatNamespaceV2.toString() + "/healthcheck"));
+ }
}
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/InactiveTopicDeleteTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/InactiveTopicDeleteTest.java
index 3b7d9aa..1a2dd42 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/InactiveTopicDeleteTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/InactiveTopicDeleteTest.java
@@ -35,11 +35,13 @@ import java.util.Map;
import java.util.Set;
import java.util.UUID;
import java.util.concurrent.TimeUnit;
-import org.apache.pulsar.broker.admin.impl.BrokersBase;
+import org.apache.pulsar.broker.namespace.NamespaceService;
import org.apache.pulsar.broker.service.persistent.PersistentTopic;
import org.apache.pulsar.client.api.Consumer;
import org.apache.pulsar.client.api.Producer;
+import org.apache.pulsar.common.naming.NamespaceName;
import org.apache.pulsar.common.naming.TopicName;
+import org.apache.pulsar.common.naming.TopicVersion;
import org.apache.pulsar.common.policies.data.InactiveTopicDeleteMode;
import org.apache.pulsar.common.policies.data.InactiveTopicPolicies;
import org.apache.pulsar.zookeeper.ZooKeeperManagedLedgerCache;
@@ -580,49 +582,32 @@ public class InactiveTopicDeleteTest extends BrokerTestBase {
}
@Test(timeOut = 30000)
- public void testInternalTopicInactiveNotClean() throws Exception {
+ public void testHealthTopicInactiveNotClean() throws Exception {
conf.setSystemTopicEnabled(true);
conf.setBrokerDeleteInactiveTopicsMode(InactiveTopicDeleteMode.delete_when_no_subscriptions);
conf.setBrokerDeleteInactiveTopicsFrequencySeconds(1);
super.baseSetup();
// init topic
- final String healthCheckTopic = "persistent://prop/ns-abc/"+ BrokersBase.HEALTH_CHECK_TOPIC_SUFFIX;
- final String topic = "persistent://prop/ns-abc/testDeleteWhenNoSubscriptions";
-
- Producer<byte[]> producer = pulsarClient.newProducer()
- .topic(topic)
- .create();
- Consumer<byte[]> consumer = pulsarClient.newConsumer()
- .topic(topic)
- .subscriptionName("sub")
- .subscribe();
-
- Producer<byte[]> heathCheckProducer = pulsarClient.newProducer()
- .topic(healthCheckTopic)
- .create();
- Consumer<byte[]> heathCheckConsumer = pulsarClient.newConsumer()
- .topic(healthCheckTopic)
- .subscriptionName("healthCheck")
- .subscribe();
-
- consumer.close();
- producer.close();
- heathCheckConsumer.close();
- heathCheckProducer.close();
-
- Awaitility.await().untilAsserted(() -> Assert.assertTrue(admin.topics().getList("prop/ns-abc")
- .contains(topic)));
- Awaitility.await().untilAsserted(() -> {
- Assert.assertTrue(admin.topics().getList("prop/ns-abc").contains(healthCheckTopic));
- });
-
- admin.topics().deleteSubscription(topic, "sub");
- admin.topics().deleteSubscription(healthCheckTopic, "healthCheck");
-
- Awaitility.await().untilAsserted(() -> Assert.assertFalse(admin.topics().getList("prop/ns-abc")
- .contains(topic)));
- Awaitility.await().pollDelay(2, TimeUnit.SECONDS)
- .untilAsserted(() -> Assert.assertTrue(admin.topics().getList("prop/ns-abc")
- .contains(healthCheckTopic)));
+ NamespaceName heartbeatNamespaceV1 = NamespaceService.getHeartbeatNamespace(pulsar.getAdvertisedAddress(), pulsar.getConfig());
+ final String healthCheckTopicV1 = "persistent://" + heartbeatNamespaceV1 + "/healthcheck";
+
+ NamespaceName heartbeatNamespaceV2 = NamespaceService.getHeartbeatNamespaceV2(pulsar.getAdvertisedAddress(), pulsar.getConfig());
+ final String healthCheckTopicV2 = "persistent://" + heartbeatNamespaceV2 + "/healthcheck";
+
+ admin.brokers().healthcheck(TopicVersion.V1);
+ admin.brokers().healthcheck(TopicVersion.V2);
+
+ List<String> V1Partitions = pulsar
+ .getPulsarResources()
+ .getTopicResources()
+ .getExistingPartitions(TopicName.get(healthCheckTopicV1))
+ .get(10, TimeUnit.SECONDS);
+ List<String> V2Partitions = pulsar
+ .getPulsarResources()
+ .getTopicResources()
+ .getExistingPartitions(TopicName.get(healthCheckTopicV2))
+ .get(10, TimeUnit.SECONDS);
+ Assert.assertTrue(V1Partitions.contains(healthCheckTopicV1));
+ Assert.assertTrue(V2Partitions.contains(healthCheckTopicV2));
}
}
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/systopic/NamespaceEventsSystemTopicServiceTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/systopic/NamespaceEventsSystemTopicServiceTest.java
index 2daca67..d81c89d 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/systopic/NamespaceEventsSystemTopicServiceTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/systopic/NamespaceEventsSystemTopicServiceTest.java
@@ -21,6 +21,7 @@ package org.apache.pulsar.broker.systopic;
import com.google.common.collect.Sets;
import lombok.Cleanup;
import org.apache.pulsar.broker.auth.MockedPulsarServiceBaseTest;
+import org.apache.pulsar.broker.service.BrokerService;
import org.apache.pulsar.broker.service.persistent.PersistentTopic;
import org.apache.pulsar.client.admin.PulsarAdminException;
import org.apache.pulsar.client.api.Message;
@@ -143,9 +144,9 @@ public class NamespaceEventsSystemTopicServiceTest extends MockedPulsarServiceBa
admin.topics().createPartitionedTopic(normalTopic, 3);
TopicName systemTopicName = TopicName.get(systemTopic);
TopicName normalTopicName = TopicName.get(normalTopic);
-
- Assert.assertEquals(SystemTopicClient.isSystemTopic(systemTopicName), true);
- Assert.assertEquals(SystemTopicClient.isSystemTopic(normalTopicName), false);
+ BrokerService brokerService = pulsar.getBrokerService();
+ Assert.assertEquals(brokerService.isSystemTopic(systemTopicName), true);
+ Assert.assertEquals(brokerService.isSystemTopic(normalTopicName), false);
}
private void prepareData() throws PulsarAdminException {
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/systopic/SystemTopicClientTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/systopic/SystemTopicClientTest.java
deleted file mode 100644
index d6790b7..0000000
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/systopic/SystemTopicClientTest.java
+++ /dev/null
@@ -1,47 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-package org.apache.pulsar.broker.systopic;
-
-import static org.testng.Assert.assertFalse;
-import static org.testng.Assert.assertTrue;
-import org.apache.pulsar.common.naming.TopicName;
-import org.testng.annotations.Test;
-
-public class SystemTopicClientTest {
-
- @Test
- public void testIsSystemTopic() {
- assertFalse(SystemTopicClient.isSystemTopic(TopicName.get("test")));
- assertFalse(SystemTopicClient.isSystemTopic(TopicName.get("public/default/test")));
- assertFalse(SystemTopicClient.isSystemTopic(TopicName.get("persistent://public/default/test")));
- assertFalse(SystemTopicClient.isSystemTopic(TopicName.get("non-persistent://public/default/test")));
-
- assertTrue(SystemTopicClient.isSystemTopic(TopicName.get("__change_events")));
- assertTrue(SystemTopicClient.isSystemTopic(TopicName.get("__change_events-partition-0")));
- assertTrue(SystemTopicClient.isSystemTopic(TopicName.get("__change_events-partition-1")));
- assertTrue(SystemTopicClient.isSystemTopic(TopicName.get("__transaction_buffer_snapshot")));
- assertTrue(SystemTopicClient.isSystemTopic(TopicName.get("__transaction_buffer_snapshot-partition-0")));
- assertTrue(SystemTopicClient.isSystemTopic(TopicName.get("__transaction_buffer_snapshot-partition-1")));
- assertTrue(SystemTopicClient.isSystemTopic(TopicName
- .get("topicxxx-partition-0-multiTopicsReader-f433329d68__transaction_pending_ack")));
- assertTrue(SystemTopicClient.isSystemTopic(
- TopicName.get("topicxxx-multiTopicsReader-f433329d68__transaction_pending_ack")));
-
- }
-}