You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by lh...@apache.org on 2026/10/01 19:50:10 UTC
(pulsar) 11/32: [improve][broker] Add configuration to enable shadow topics (#26739)
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch lh-branch-4.0-backports-2026-10-01
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit edfbc86bc7a171d5cffd041a4bef5c846a295c1d
Author: Lari Hotari <lh...@users.noreply.github.com>
AuthorDate: Tue Sep 29 13:30:56 2026 +0300
[improve][broker] Add configuration to enable shadow topics (#26739)
(cherry picked from commit 9f827de0f062e4ca8ef7e6320f8bd2b936d727f0)
---
conf/broker.conf | 3 +
conf/standalone.conf | 3 +
.../apache/pulsar/broker/ServiceConfiguration.java | 6 +
.../apache/pulsar/broker/admin/AdminResource.java | 28 ++++
.../broker/admin/impl/PersistentTopicsBase.java | 10 +-
.../pulsar/broker/service/BrokerService.java | 3 +
.../broker/service/persistent/PersistentTopic.java | 3 +-
.../apache/pulsar/broker/admin/TopicAuthZTest.java | 101 ++++++++++++++-
.../pulsar/broker/admin/TopicPoliciesTest.java | 1 +
.../InmemoryTopicPoliciesServiceServiceTest.java | 1 +
.../service/persistent/ShadowReplicatorTest.java | 1 +
.../service/persistent/ShadowTopicRealBkTest.java | 1 +
.../broker/service/persistent/ShadowTopicTest.java | 1 +
.../persistent/ShadowTopicsDisabledTest.java | 143 +++++++++++++++++++++
14 files changed, 301 insertions(+), 4 deletions(-)
diff --git a/conf/broker.conf b/conf/broker.conf
index c38e9fb1142f..00ce4a762b92 100644
--- a/conf/broker.conf
+++ b/conf/broker.conf
@@ -600,6 +600,9 @@ topicOrderedExecutorThreadNum=
# Enable broker to load persistent topics
enablePersistentTopics=true
+# Enable shadow topic creation, loading and replication. Requires a broker restart.
+enableShadowTopics=false
+
# Enable broker to load non-persistent topics
enableNonPersistentTopics=true
diff --git a/conf/standalone.conf b/conf/standalone.conf
index 4a7b2dd5ec3e..b6d46695e643 100644
--- a/conf/standalone.conf
+++ b/conf/standalone.conf
@@ -356,6 +356,9 @@ topicOrderedExecutorThreadNum=8
# Enable broker to load persistent topics
enablePersistentTopics=true
+# Enable shadow topic creation, loading and replication. Requires a broker restart.
+enableShadowTopics=false
+
# Enable broker to load non-persistent topics
enableNonPersistentTopics=true
diff --git a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java
index 84154cf8842d..0628812e9c55 100644
--- a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java
+++ b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java
@@ -1487,6 +1487,12 @@ public class ServiceConfiguration implements PulsarConfiguration {
)
private boolean enablePersistentTopics = true;
+ @FieldContext(
+ category = CATEGORY_SERVER,
+ doc = "Enable shadow topic creation, loading and replication. Requires a broker restart."
+ )
+ private boolean enableShadowTopics = false;
+
@FieldContext(
category = CATEGORY_SERVER,
doc = "Enable broker to load non-persistent topics"
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 03fa76e48bf2..e6da151f1a52 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
@@ -18,6 +18,7 @@
*/
package org.apache.pulsar.broker.admin;
+import static org.apache.bookkeeper.mledger.ManagedLedgerConfig.PROPERTY_SOURCE_TOPIC_KEY;
import com.fasterxml.jackson.databind.ObjectReader;
import com.fasterxml.jackson.databind.ObjectWriter;
import com.google.errorprone.annotations.CanIgnoreReturnValue;
@@ -580,6 +581,7 @@ public abstract class AdminResource extends PulsarWebResource {
return;
}
validateNamespaceOperationAsync(topicName.getNamespaceObject(), NamespaceOperation.CREATE_TOPIC)
+ .thenCompose(__ -> validateShadowTopicPropertiesAsync(properties))
.thenCompose((__) -> getNamespacePoliciesAsync(namespaceName).exceptionally(ex -> {
Throwable unwrapped = FutureUtil.unwrapCompletionException(ex);
if (unwrapped instanceof RestException re) {
@@ -787,6 +789,32 @@ public abstract class AdminResource extends PulsarWebResource {
});
}
+ protected CompletableFuture<Void> validateShadowTopicPropertiesAsync(Map<String, String> properties) {
+ if (properties == null || !properties.containsKey(PROPERTY_SOURCE_TOPIC_KEY)) {
+ return CompletableFuture.completedFuture(null);
+ }
+ if (!pulsar().getConfiguration().isEnableShadowTopics()) {
+ return FutureUtil.failedFuture(new RestException(Status.METHOD_NOT_ALLOWED, "Shadow topics are disabled"));
+ }
+ String shadowSource = properties.get(PROPERTY_SOURCE_TOPIC_KEY);
+ if (shadowSource == null) {
+ return CompletableFuture.completedFuture(null);
+ }
+ final TopicName sourceTopic;
+ try {
+ sourceTopic = TopicName.get(shadowSource);
+ } catch (IllegalArgumentException e) {
+ return FutureUtil.failedFuture(new RestException(Status.PRECONDITION_FAILED,
+ "Invalid shadow source topic name"));
+ }
+ return validateShadowTopicTenantAsync(sourceTopic);
+ }
+
+ protected CompletableFuture<Void> validateShadowTopicTenantAsync(TopicName relatedTopic) {
+ return topicName.getTenant().equals(relatedTopic.getTenant())
+ ? CompletableFuture.completedFuture(null) : validateSuperUserAccessAsync();
+ }
+
private CompletableFuture<Void> provisionPartitionedTopicPath(int numPartitions,
boolean createLocalTopicOnly,
Map<String, String> properties) {
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java
index 053276501fc8..00df3d1ac506 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java
@@ -326,7 +326,8 @@ public class PersistentTopicsBase extends AdminResource {
protected CompletableFuture<Void> internalCreateNonPartitionedTopicAsync(boolean authoritative,
Map<String, String> properties) {
- CompletableFuture<Void> ret = validateNonPartitionTopicNameAsync(topicName.getLocalName());
+ CompletableFuture<Void> ret = validateNonPartitionTopicNameAsync(topicName.getLocalName())
+ .thenCompose(__ -> validateShadowTopicPropertiesAsync(properties));
if (topicName.isGlobal()) {
ret = ret.thenCompose(__ -> validateGlobalNamespaceOwnershipAsync(namespaceName));
}
@@ -655,6 +656,7 @@ public class PersistentTopicsBase extends AdminResource {
return CompletableFuture.completedFuture(null);
}
return validateTopicOperationAsync(topicName, TopicOperation.UPDATE_METADATA)
+ .thenCompose(__ -> validateShadowTopicPropertiesAsync(properties))
.thenCompose(__ -> validateTopicOwnershipAsync(topicName, authoritative))
.thenCompose(__ -> {
if (topicName.isPartitioned()) {
@@ -5357,7 +5359,8 @@ public class PersistentTopicsBase extends AdminResource {
return FutureUtil.failedFuture(new RestException(Status.PRECONDITION_FAILED,
"Only persistent topic can be set as shadow topic"));
}
- futures.add(pulsar().getNamespaceService().checkTopicExistsAsync(shadowTopicName)
+ futures.add(validateShadowTopicTenantAsync(shadowTopicName)
+ .thenCompose(__ -> pulsar().getNamespaceService().checkTopicExistsAsync(shadowTopicName))
.thenAccept(info -> {
boolean exists = info.isExists();
info.recycle();
@@ -5375,6 +5378,9 @@ public class PersistentTopicsBase extends AdminResource {
}
protected CompletableFuture<Void> internalSetShadowTopic(List<String> shadowTopics) {
+ if (!pulsar().getConfiguration().isEnableShadowTopics()) {
+ return FutureUtil.failedFuture(new RestException(Status.METHOD_NOT_ALLOWED, "Shadow topics are disabled"));
+ }
if (!topicName.isPersistent()) {
return FutureUtil.failedFuture(new RestException(Status.PRECONDITION_FAILED,
"Only persistent source topic is supported with shadow topics."));
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 068a5106592a..9329457caa57 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
@@ -1870,6 +1870,9 @@ public class BrokerService implements Closeable {
}
String shadowSource = managedLedgerConfig.getShadowSource();
if (shadowSource != null) {
+ if (!pulsar().getConfiguration().isEnableShadowTopics()) {
+ throw new CompletionException(new NotAllowedException("Shadow topics are disabled"));
+ }
managedLedgerConfig.setShadowSourceName(TopicName.get(shadowSource).getPersistenceNamingEncoding());
}
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java
index 729cb1fd7171..5bc9fd1505b0 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java
@@ -2259,7 +2259,8 @@ public class PersistentTopic extends AbstractTopic implements Topic, AddEntryCal
}
private CompletableFuture<Void> checkShadowReplication() {
- if (CollectionUtils.isEmpty(shadowTopics)) {
+ if (!brokerService.pulsar().getConfiguration().isEnableShadowTopics()
+ || CollectionUtils.isEmpty(shadowTopics)) {
return CompletableFuture.completedFuture(null);
}
List<String> configuredShadowTopics = shadowTopics;
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/TopicAuthZTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/TopicAuthZTest.java
index c9fdf1980f6b..f5956b3203cb 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/TopicAuthZTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/TopicAuthZTest.java
@@ -19,6 +19,9 @@
package org.apache.pulsar.broker.admin;
+import static org.apache.bookkeeper.mledger.ManagedLedgerConfig.PROPERTY_SOURCE_TOPIC_KEY;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
import static org.mockito.Mockito.doReturn;
import com.google.common.collect.Lists;
import com.google.common.collect.Sets;
@@ -26,15 +29,19 @@ import io.jsonwebtoken.Jwts;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.HashSet;
+import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.Set;
import java.util.UUID;
+import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import lombok.Cleanup;
import lombok.SneakyThrows;
import org.apache.commons.lang3.reflect.FieldUtils;
+import org.apache.pulsar.broker.service.persistent.PersistentReplicator;
+import org.apache.pulsar.broker.service.persistent.PersistentTopic;
import org.apache.pulsar.broker.service.plugin.EntryFilterDefinition;
import org.apache.pulsar.broker.service.plugin.EntryFilterProvider;
import org.apache.pulsar.broker.service.plugin.EntryFilterTest;
@@ -65,6 +72,7 @@ import org.apache.pulsar.common.policies.data.RetentionPolicies;
import org.apache.pulsar.common.policies.data.SubscribeRate;
import org.apache.pulsar.common.policies.data.TenantInfo;
import org.apache.pulsar.common.policies.data.TopicOperation;
+import org.awaitility.Awaitility;
import org.mockito.Mockito;
import org.testng.Assert;
import org.testng.annotations.AfterClass;
@@ -78,6 +86,7 @@ public class TopicAuthZTest extends AuthZTest {
@SneakyThrows
@BeforeClass(alwaysRun = true)
public void setup() {
+ getServiceConfiguration().setEnableShadowTopics(true);
configureTokenAuthentication();
configureDefaultAuthorization();
start();
@@ -1347,6 +1356,96 @@ public class TopicAuthZTest extends AuthZTest {
"entryFilterProvider", oldEntryFilterProvider, true);
}
+ @DataProvider
+ public Object[][] shadowTopicPartitionCounts() {
+ return new Object[][] {{0}, {2}};
+ }
+
+ @Test(dataProvider = "shadowTopicPartitionCounts")
+ public void testTenantAdminShadowTopicsRequireSameTenant(int partitions) throws Exception {
+ String otherTenant = "shadow-tenant-" + UUID.randomUUID();
+ String source = "persistent://" + otherTenant + "/ns/source";
+ String destination = "persistent://public/default/shadow-" + UUID.randomUUID();
+ superUserAdmin.tenants().createTenant(otherTenant, TenantInfo.builder()
+ .allowedClusters(superUserAdmin.tenants().getTenantInfo("public").getAllowedClusters())
+ .build());
+ superUserAdmin.namespaces().createNamespace(otherTenant + "/ns");
+ try {
+ superUserAdmin.topics().createNonPartitionedTopic(source);
+ Map<String, String> properties = Map.of(PROPERTY_SOURCE_TOPIC_KEY, source);
+ assertThatThrownBy(() -> createTopicWithProperties(tenantManagerAdmin, destination, partitions, properties))
+ .isInstanceOf(PulsarAdminException.NotAuthorizedException.class);
+ createTopicWithProperties(tenantManagerAdmin, destination, partitions, Map.of("key", "value"));
+ assertThatThrownBy(() -> tenantManagerAdmin.topics().updateProperties(destination, properties))
+ .isInstanceOf(PulsarAdminException.NotAuthorizedException.class);
+ assertThat(tenantManagerAdmin.topics().getProperties(destination)).containsExactlyEntriesOf(
+ Map.of("key", "value"));
+ assertThatThrownBy(() -> tenantManagerAdmin.topics().setShadowTopics(destination, List.of(source)))
+ .isInstanceOf(PulsarAdminException.NotAuthorizedException.class);
+ assertThat(tenantManagerAdmin.topics().getShadowTopics(destination)).isNull();
+ } finally {
+ deleteTopic(destination, partitions > 0);
+ superUserAdmin.topics().delete(source, true);
+ superUserAdmin.namespaces().deleteNamespace(otherTenant + "/ns");
+ superUserAdmin.tenants().deleteTenant(otherTenant);
+ }
+ }
+
+ @Test(dataProvider = "shadowTopicPartitionCounts")
+ public void testSuperUserShadowTopicsAcrossTenants(int partitions) throws Exception {
+ String otherTenant = "shadow-tenant-" + UUID.randomUUID();
+ String source = "persistent://" + otherTenant + "/ns/source";
+ String destination = "persistent://public/default/shadow-" + UUID.randomUUID();
+ String updatedDestination = destination + "-updated";
+ superUserAdmin.tenants().createTenant(otherTenant, TenantInfo.builder()
+ .allowedClusters(superUserAdmin.tenants().getTenantInfo("public").getAllowedClusters())
+ .build());
+ superUserAdmin.namespaces().createNamespace(otherTenant + "/ns");
+ try {
+ createTopicWithProperties(superUserAdmin, source, partitions, Map.of());
+ Map<String, String> properties = Map.of(PROPERTY_SOURCE_TOPIC_KEY, source);
+ createTopicWithProperties(superUserAdmin, destination, partitions, properties);
+ createTopicWithProperties(superUserAdmin, updatedDestination, partitions, Map.of());
+ superUserAdmin.topics().updateProperties(updatedDestination, properties);
+ assertThat(superUserAdmin.topics().getProperties(updatedDestination))
+ .containsEntry(PROPERTY_SOURCE_TOPIC_KEY, source);
+ superUserAdmin.topics().setShadowTopics(source, List.of(destination));
+ for (int i = 0; i < Math.max(1, partitions); i++) {
+ String sourcePartition = partitions == 0 ? source : TopicName.get(source).getPartition(i).toString();
+ String destinationPartition = partitions == 0 ? destination
+ : TopicName.get(destination).getPartition(i).toString();
+ superUserAdmin.lookups().lookupTopic(destinationPartition);
+ superUserAdmin.lookups().lookupTopic(sourcePartition);
+ PersistentTopic sourceTopic = (PersistentTopic) getPulsarService().getBrokerService()
+ .getTopic(sourcePartition, true).get(10, TimeUnit.SECONDS).orElseThrow();
+ PersistentTopic shadow = (PersistentTopic) getPulsarService().getBrokerService()
+ .getTopic(destinationPartition, true).get(10, TimeUnit.SECONDS).orElseThrow();
+ assertThat(shadow.getShadowSourceTopic()).contains(TopicName.get(sourcePartition));
+ Awaitility.await().untilAsserted(() -> {
+ assertThat(sourceTopic.getShadowReplicators()).containsKey(destination);
+ assertThat(((PersistentReplicator) sourceTopic.getShadowReplicators().get(destination))
+ .getState().toString()).isEqualTo("Started");
+ });
+ }
+ } finally {
+ superUserAdmin.topics().removeShadowTopics(source);
+ deleteTopic(destination, partitions > 0);
+ deleteTopic(updatedDestination, partitions > 0);
+ deleteTopic(source, partitions > 0);
+ superUserAdmin.namespaces().deleteNamespace(otherTenant + "/ns");
+ superUserAdmin.tenants().deleteTenant(otherTenant);
+ }
+ }
+
+ private void createTopicWithProperties(PulsarAdmin caller, String topic, int partitions,
+ Map<String, String> properties) throws PulsarAdminException {
+ if (partitions == 0) {
+ caller.topics().createNonPartitionedTopic(topic, properties);
+ } else {
+ caller.topics().createPartitionedTopic(topic, partitions, properties);
+ }
+ }
+
@Test
@SneakyThrows
public void testShadowTopic() {
@@ -1363,7 +1462,7 @@ public class TopicAuthZTest extends AuthZTest {
.build();
String shadowTopic = topic + "-shadow-topic";
- superUserAdmin.topics().createShadowTopic(shadowTopic, topic);
+ tenantManagerAdmin.topics().createShadowTopic(shadowTopic, topic);
superUserAdmin.topics().setShadowTopics(topic, Lists.newArrayList(shadowTopic));
superUserAdmin.topics().getShadowTopics(topic);
superUserAdmin.topics().removeShadowTopics(topic);
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/TopicPoliciesTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/TopicPoliciesTest.java
index 302f060953bc..0c02e5bcc0a5 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/TopicPoliciesTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/TopicPoliciesTest.java
@@ -147,6 +147,7 @@ public class TopicPoliciesTest extends MockedPulsarServiceBaseTest {
@BeforeClass(alwaysRun = true)
@Override
protected void setup() throws Exception {
+ conf.setEnableShadowTopics(true);
this.conf.setDefaultNumberOfNamespaceBundles(1);
this.conf.setForceDeleteNamespaceAllowed(true);
super.internalSetup();
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/InmemoryTopicPoliciesServiceServiceTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/InmemoryTopicPoliciesServiceServiceTest.java
index a27fcd97f583..4c8f0c0e9712 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/InmemoryTopicPoliciesServiceServiceTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/InmemoryTopicPoliciesServiceServiceTest.java
@@ -38,6 +38,7 @@ public class InmemoryTopicPoliciesServiceServiceTest extends MockedPulsarService
@Override
protected void setup() throws Exception {
conf.setTopicPoliciesServiceClassName(InmemoryTopicPoliciesService.class.getName());
+ conf.setEnableShadowTopics(true);
conf.setSystemTopicEnabled(false); // verify topic policies don't rely on system topics
super.internalSetup();
super.setupDefaultTenantAndNamespace();
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/ShadowReplicatorTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/ShadowReplicatorTest.java
index 007fc3766872..2c682c16eb69 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/ShadowReplicatorTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/ShadowReplicatorTest.java
@@ -55,6 +55,7 @@ public class ShadowReplicatorTest extends BrokerTestBase {
@BeforeClass(alwaysRun = true)
@Override
protected void setup() throws Exception {
+ conf.setEnableShadowTopics(true);
super.baseSetup();
admin.tenants().createTenant("prop1",
new TenantInfoImpl(Sets.newHashSet("appid1"), Sets.newHashSet("test")));
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/ShadowTopicRealBkTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/ShadowTopicRealBkTest.java
index b0e572a826c4..f4d7404c422b 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/ShadowTopicRealBkTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/ShadowTopicRealBkTest.java
@@ -51,6 +51,7 @@ public class ShadowTopicRealBkTest {
public void setup() throws Exception {
bk.start();
final var config = new ServiceConfiguration();
+ config.setEnableShadowTopics(true);
config.setClusterName(cluster);
config.setAdvertisedAddress("localhost");
config.setBrokerServicePort(Optional.of(0));
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/ShadowTopicTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/ShadowTopicTest.java
index 840454b4e210..0c78480883a6 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/ShadowTopicTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/ShadowTopicTest.java
@@ -51,6 +51,7 @@ public class ShadowTopicTest extends BrokerTestBase {
@BeforeClass(alwaysRun = true)
@Override
protected void setup() throws Exception {
+ conf.setEnableShadowTopics(true);
baseSetup();
}
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/ShadowTopicsDisabledTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/ShadowTopicsDisabledTest.java
new file mode 100644
index 000000000000..cb2c1ca57cde
--- /dev/null
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/ShadowTopicsDisabledTest.java
@@ -0,0 +1,143 @@
+/*
+ * 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.service.persistent;
+
+import static org.apache.bookkeeper.mledger.ManagedLedgerConfig.PROPERTY_SOURCE_TOPIC_KEY;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+import org.apache.pulsar.broker.ServiceConfiguration;
+import org.apache.pulsar.broker.service.BrokerServiceException;
+import org.apache.pulsar.broker.service.BrokerTestBase;
+import org.apache.pulsar.client.admin.PulsarAdminException;
+import org.apache.pulsar.client.api.Consumer;
+import org.apache.pulsar.client.api.Producer;
+import org.apache.pulsar.client.api.Schema;
+import org.apache.pulsar.common.naming.TopicName;
+import org.apache.pulsar.common.partition.PartitionedTopicMetadata;
+import org.testng.annotations.AfterClass;
+import org.testng.annotations.BeforeClass;
+import org.testng.annotations.DataProvider;
+import org.testng.annotations.Test;
+
+public class ShadowTopicsDisabledTest extends BrokerTestBase {
+
+ @BeforeClass(alwaysRun = true)
+ @Override
+ protected void setup() throws Exception {
+ baseSetup();
+ }
+
+ @AfterClass(alwaysRun = true)
+ @Override
+ protected void cleanup() throws Exception {
+ internalCleanup();
+ }
+
+ @Override
+ protected String newTopicName() {
+ return "persistent://" + super.newTopicName();
+ }
+
+ @DataProvider
+ public Object[][] partitionCounts() {
+ return new Object[][] {{0}, {2}};
+ }
+
+ @Test(dataProvider = "partitionCounts")
+ public void testCreationAndConfigurationDisabled(int partitions) throws Exception {
+ assertThat(new ServiceConfiguration().isEnableShadowTopics()).isFalse();
+ assertThat(pulsar.getConfig().isEnableShadowTopics()).isFalse();
+ String source = newTopicName();
+ String shadow = source + "-shadow";
+ if (partitions == 0) {
+ admin.topics().createNonPartitionedTopic(source);
+ } else {
+ admin.topics().createPartitionedTopic(source, partitions);
+ }
+ assertThatThrownBy(() -> admin.topics().createShadowTopic(shadow, source))
+ .isInstanceOf(PulsarAdminException.NotAllowedException.class)
+ .hasMessageContaining("Shadow topics are disabled");
+ assertThatThrownBy(() -> admin.topics().setShadowTopics(source, List.of(shadow)))
+ .isInstanceOf(PulsarAdminException.NotAllowedException.class);
+ assertThatThrownBy(() -> admin.topics().updateProperties(source,
+ Map.of(PROPERTY_SOURCE_TOPIC_KEY, shadow)))
+ .isInstanceOf(PulsarAdminException.NotAllowedException.class);
+ assertThat(admin.topics().getShadowTopics(source)).isNull();
+ admin.topics().removeShadowTopics(source);
+ try (Consumer<String> consumer = pulsarClient.newConsumer(Schema.STRING).topic(source)
+ .subscriptionName("ordinary").subscribe();
+ Producer<String> producer = pulsarClient.newProducer(Schema.STRING).topic(source).create()) {
+ producer.send("message");
+ assertThat(consumer.receive(10, TimeUnit.SECONDS).getValue()).isEqualTo("message");
+ }
+ }
+
+ @Test(dataProvider = "partitionCounts")
+ public void testExistingShadowTopicCannotLoad(int partitions) throws Exception {
+ String source = newTopicName();
+ String shadow = source + "-shadow";
+ String topicToLoad = shadow;
+ if (partitions == 0) {
+ admin.topics().createNonPartitionedTopic(shadow);
+ PersistentTopic topic = (PersistentTopic) pulsar.getBrokerService().getTopicIfExists(shadow).get()
+ .orElseThrow();
+ topic.getManagedLedger().setProperty(PROPERTY_SOURCE_TOPIC_KEY, source);
+ admin.topics().unload(shadow);
+ } else {
+ pulsar.getPulsarResources().getNamespaceResources().getPartitionedTopicResources()
+ .createPartitionedTopic(TopicName.get(shadow),
+ new PartitionedTopicMetadata(partitions, Map.of(PROPERTY_SOURCE_TOPIC_KEY, source)));
+ topicToLoad = TopicName.get(shadow).getPartition(0).toString();
+ }
+ String finalTopicToLoad = topicToLoad;
+ try {
+ assertThatThrownBy(() -> pulsar.getBrokerService().getTopic(finalTopicToLoad, true)
+ .get(10, TimeUnit.SECONDS))
+ .hasRootCauseInstanceOf(BrokerServiceException.NotAllowedException.class)
+ .hasStackTraceContaining("Shadow topics are disabled");
+ assertThat(pulsar.getBrokerService().getTopicReference(finalTopicToLoad)).isEmpty();
+ } finally {
+ if (partitions == 0) {
+ pulsar.getDefaultManagedLedgerFactory().delete(
+ TopicName.get(shadow).getPersistenceNamingEncoding());
+ } else {
+ admin.topics().deletePartitionedTopic(shadow, true);
+ }
+ }
+ }
+
+ @Test
+ public void testExistingShadowPolicyDoesNotStartReplication() throws Exception {
+ String source = newTopicName();
+ String shadow = source + "-shadow";
+ admin.topics().createNonPartitionedTopic(source);
+ pulsar.getTopicPoliciesService().updateTopicPoliciesAsync(TopicName.get(source), false, false,
+ policies -> policies.setShadowTopics(List.of(shadow))).get(10, TimeUnit.SECONDS);
+ admin.topics().unload(source);
+ PersistentTopic topic = (PersistentTopic) pulsar.getBrokerService().getTopicIfExists(source).get()
+ .orElseThrow();
+ assertThat(admin.topics().getShadowTopics(source)).containsExactly(shadow);
+ topic.checkReplication().get(10, TimeUnit.SECONDS);
+ assertThat(topic.getShadowReplicators()).isEmpty();
+ admin.topics().removeShadowTopics(source);
+ }
+}