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/09/29 10:31:02 UTC
(pulsar) branch master updated: [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 master
in repository https://gitbox.apache.org/repos/asf/pulsar.git
The following commit(s) were added to refs/heads/master by this push:
new 9f827de0f062 [improve][broker] Add configuration to enable shadow topics (#26739)
9f827de0f062 is described below
commit 9f827de0f062e4ca8ef7e6320f8bd2b936d727f0
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)
---
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 | 8 +-
.../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 | 44 ++++++--
.../persistent/ShadowTopicsDisabledTest.java | 121 +++++++++++++++++++++
14 files changed, 310 insertions(+), 14 deletions(-)
diff --git a/conf/broker.conf b/conf/broker.conf
index 4a6017bade16..64a9819537dd 100644
--- a/conf/broker.conf
+++ b/conf/broker.conf
@@ -666,6 +666,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 b2ca3964facb..7e48b6853e2f 100644
--- a/conf/standalone.conf
+++ b/conf/standalone.conf
@@ -415,6 +415,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 99f52591bc1f..a5aff1d767a8 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
@@ -1811,6 +1811,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 182b25dbeb13..f624e7221c57 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;
@@ -569,6 +570,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) {
@@ -809,6 +811,32 @@ public abstract class AdminResource extends PulsarWebResource {
return pulsar().getNamespaceService().checkTopicExistsAsync(topicName);
}
+ 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 bffe01a9651a..0834c80065b9 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
@@ -339,6 +339,7 @@ public class PersistentTopicsBase extends AdminResource {
protected CompletableFuture<Void> internalCreateNonPartitionedTopicAsync(boolean authoritative,
Map<String, String> properties) {
return validateNonPartitionTopicNameAsync(topicName.getLocalName())
+ .thenCompose(__ -> validateShadowTopicPropertiesAsync(properties))
.thenCompose(__ -> validateGlobalNamespaceOwnershipAsync(namespaceName))
.thenCompose(__ -> validateTopicOwnershipAsync(topicName, authoritative))
.thenCompose(__ -> validateNamespaceOperationAsync(topicName.getNamespaceObject(),
@@ -678,6 +679,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()) {
@@ -5664,7 +5666,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();
@@ -5682,6 +5685,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 96d9d746560a..cead1b20b141 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
@@ -2223,6 +2223,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 ae7c598349ce..77cc345ff5c3 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
@@ -2371,7 +2371,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 c43b83e894ed..866588433815 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
@@ -149,6 +149,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 666926c3d55c..a0f74de0d112 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
@@ -63,6 +63,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 9b52582f6f39..74599c3549db 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
@@ -52,6 +52,7 @@ public class ShadowTopicRealBkTest {
bk.start();
final int zkPort = bk.getZookeeperPort();
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 58ebeeee4f73..8c9a208ff7ad 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
@@ -29,7 +29,8 @@ import lombok.CustomLog;
import lombok.Data;
import lombok.NoArgsConstructor;
import org.apache.bookkeeper.mledger.impl.ShadowManagedLedgerImpl;
-import org.apache.pulsar.broker.service.SharedPulsarBaseTest;
+import org.apache.pulsar.broker.BrokerTestUtil;
+import org.apache.pulsar.broker.service.BrokerTestBase;
import org.apache.pulsar.client.api.Consumer;
import org.apache.pulsar.client.api.Message;
import org.apache.pulsar.client.api.MessageId;
@@ -40,10 +41,30 @@ import org.apache.pulsar.client.api.SubscriptionInitialPosition;
import org.apache.pulsar.common.naming.TopicName;
import org.awaitility.Awaitility;
import org.testng.Assert;
+import org.testng.annotations.AfterClass;
+import org.testng.annotations.BeforeClass;
import org.testng.annotations.Test;
@CustomLog
-public class ShadowTopicTest extends SharedPulsarBaseTest {
+public class ShadowTopicTest extends BrokerTestBase {
+
+ @BeforeClass(alwaysRun = true)
+ @Override
+ protected void setup() throws Exception {
+ conf.setEnableShadowTopics(true);
+ super.baseSetup();
+ }
+
+ @AfterClass(alwaysRun = true)
+ @Override
+ protected void cleanup() throws Exception {
+ super.internalCleanup();
+ }
+
+ @Override
+ protected String newTopicName() {
+ return BrokerTestUtil.newUniqueName("persistent://prop/ns-abc/shadow-test");
+ }
@Test
public void testNonPartitionedShadowTopicSetup() throws Exception {
@@ -53,16 +74,16 @@ public class ShadowTopicTest extends SharedPulsarBaseTest {
admin.topics().createNonPartitionedTopic(sourceTopic);
admin.topics().createShadowTopic(shadowTopic, sourceTopic);
PersistentTopic brokerShadowTopic =
- (PersistentTopic) getTopicIfExists(shadowTopic).get().get();
+ (PersistentTopic) pulsar.getBrokerService().getTopicIfExists(shadowTopic).get().get();
Assert.assertTrue(brokerShadowTopic.getManagedLedger() instanceof ShadowManagedLedgerImpl);
Assert.assertEquals(brokerShadowTopic.getShadowSourceTopic().get().toString(), sourceTopic);
Assert.assertEquals(admin.topics().getShadowSource(shadowTopic), sourceTopic);
//2. test shadow topic could be properly loaded after unload.
- admin.namespaces().unload(getNamespace());
- Assert.assertTrue(getTopicReference(shadowTopic).isEmpty());
+ admin.namespaces().unload("prop/ns-abc");
+ Assert.assertTrue(pulsar.getBrokerService().getTopicReference(shadowTopic).isEmpty());
Assert.assertEquals(admin.topics().getShadowSource(shadowTopic), sourceTopic);
- brokerShadowTopic = (PersistentTopic) getTopicIfExists(shadowTopic).get().get();
+ brokerShadowTopic = (PersistentTopic) pulsar.getBrokerService().getTopicIfExists(shadowTopic).get().get();
Assert.assertTrue(brokerShadowTopic.getManagedLedger() instanceof ShadowManagedLedgerImpl);
Assert.assertEquals(brokerShadowTopic.getShadowSourceTopic().get().toString(), sourceTopic);
}
@@ -79,18 +100,19 @@ public class ShadowTopicTest extends SharedPulsarBaseTest {
admin.topics().createShadowTopic(shadowTopic, sourceTopic);
pulsarClient.newProducer().topic(shadowTopic).create().close(); //trigger loading partitions.
- PersistentTopic brokerShadowTopic = (PersistentTopic) getTopicIfExists(shadowTopicPartition).get().get();
+ PersistentTopic brokerShadowTopic =
+ (PersistentTopic) pulsar.getBrokerService().getTopicIfExists(shadowTopicPartition).get().get();
Assert.assertTrue(brokerShadowTopic.getManagedLedger() instanceof ShadowManagedLedgerImpl);
Assert.assertEquals(brokerShadowTopic.getShadowSourceTopic().get().toString(), sourceTopicPartition);
Assert.assertEquals(admin.topics().getShadowSource(shadowTopic), sourceTopic);
//2. test shadow topic could be properly loaded after unload.
- admin.namespaces().unload(getNamespace());
- Assert.assertTrue(getTopicReference(shadowTopic).isEmpty());
+ admin.namespaces().unload("prop/ns-abc");
+ Assert.assertTrue(pulsar.getBrokerService().getTopicReference(shadowTopic).isEmpty());
Assert.assertEquals(admin.topics().getShadowSource(shadowTopic), sourceTopic);
brokerShadowTopic =
- (PersistentTopic) getTopicIfExists(shadowTopicPartition).get().get();
+ (PersistentTopic) pulsar.getBrokerService().getTopicIfExists(shadowTopicPartition).get().get();
Assert.assertTrue(brokerShadowTopic.getManagedLedger() instanceof ShadowManagedLedgerImpl);
Assert.assertEquals(brokerShadowTopic.getShadowSourceTopic().get().toString(), sourceTopicPartition);
}
@@ -138,7 +160,7 @@ public class ShadowTopicTest extends SharedPulsarBaseTest {
private void awaitUntilShadowReplicatorReady(String sourceTopic, String shadowTopic) {
Awaitility.await().untilAsserted(() -> {
PersistentTopic sourcePersistentTopic =
- (PersistentTopic) getTopicIfExists(sourceTopic).get().get();
+ (PersistentTopic) pulsar.getBrokerService().getTopicIfExists(sourceTopic).get().get();
ShadowReplicator
replicator = (ShadowReplicator) sourcePersistentTopic.getShadowReplicators().get(shadowTopic);
Assert.assertNotNull(replicator);
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..84f4b9ff8c59
--- /dev/null
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/ShadowTopicsDisabledTest.java
@@ -0,0 +1,121 @@
+/*
+ * 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.SharedPulsarBaseTest;
+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.DataProvider;
+import org.testng.annotations.Test;
+
+public class ShadowTopicsDisabledTest extends SharedPulsarBaseTest {
+
+ @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(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) getTopicIfExists(shadow).get().orElseThrow();
+ topic.getManagedLedger().setProperty(PROPERTY_SOURCE_TOPIC_KEY, source);
+ admin.topics().unload(shadow);
+ } else {
+ getPulsar().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(() -> getTopic(finalTopicToLoad, true).get(10, TimeUnit.SECONDS))
+ .hasRootCauseInstanceOf(BrokerServiceException.NotAllowedException.class)
+ .hasStackTraceContaining("Shadow topics are disabled");
+ assertThat(getTopicReference(finalTopicToLoad)).isEmpty();
+ } finally {
+ if (partitions == 0) {
+ getPulsar().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);
+ getPulsar().getTopicPoliciesService().updateTopicPoliciesAsync(TopicName.get(source), false, false,
+ policies -> policies.setShadowTopics(List.of(shadow))).get(10, TimeUnit.SECONDS);
+ admin.topics().unload(source);
+ PersistentTopic topic = (PersistentTopic) 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);
+ }
+}