Skip to content

[fix][broker] Enable shadow topics by default #26921

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

Closed
zymap wants to merge 1 commit into from
Closed

[fix][broker] Enable shadow topics by default #26921

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
2 changes: 1 addition & 1 deletion conf/broker.conf
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -667,7 +667,7 @@ topicOrderedExecutorThreadNum=
enablePersistentTopics=true

# Enable shadow topic creation, loading and replication. Requires a broker restart.
enableShadowTopics=false
enableShadowTopics=true

# Enable broker to load non-persistent topics
enableNonPersistentTopics=true
Expand Down
2 changes: 1 addition & 1 deletion conf/standalone.conf
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -416,7 +416,7 @@ topicOrderedExecutorThreadNum=8
enablePersistentTopics=true

# Enable shadow topic creation, loading and replication. Requires a broker restart.
enableShadowTopics=false
enableShadowTopics=true

# Enable broker to load non-persistent topics
enableNonPersistentTopics=true
Expand Down
2 changes: 1 addition & 1 deletion pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1822,7 +1822,7 @@ The max allowed delay for delayed delivery (in milliseconds). If the broker rece
category = CATEGORY_SERVER,
doc = "Enable shadow topic creation, loading and replication. Requires a broker restart."
)
private boolean enableShadowTopics = false;
private boolean enableShadowTopics = true;

@FieldContext(
category = CATEGORY_SERVER,
Expand Down
11 changes: 11 additions & 0 deletions ...n/src/test/java/org/apache/pulsar/common/configuration/PulsarConfigurationLoaderTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -67,6 +67,17 @@ public void setProperties(Properties properties) {
}
}

@Test
public void testShadowTopicsEnabledByDefault() throws Exception {
assertTrue(new ServiceConfiguration().isEnableShadowTopics());
Properties properties = new Properties();
ServiceConfiguration config = PulsarConfigurationLoader.create(properties, ServiceConfiguration.class);
assertTrue(config.isEnableShadowTopics());
properties.setProperty("enableShadowTopics", "false");
config = PulsarConfigurationLoader.create(properties, ServiceConfiguration.class);
assertFalse(config.isEnableShadowTopics());
}

@Test
public void testConfigurationConverting() {
MockConfiguration mockConfiguration = new MockConfiguration();
Expand Down
1 change: 0 additions & 1 deletion pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/ShadowTopicTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -76,7 +76,6 @@ public class ShadowTopicTest extends BrokerTestBase {
@BeforeClass(alwaysRun = true)
@Override
protected void setup() throws Exception {
conf.setEnableShadowTopics(true);
super.baseSetup();
}

Expand Down
39 changes: 30 additions & 9 deletions ...r/src/test/java/org/apache/pulsar/broker/service/persistent/ShadowTopicsDisabledTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -24,19 +24,38 @@
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.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 SharedPulsarBaseTest {
public class ShadowTopicsDisabledTest extends BrokerTestBase {

@BeforeClass(alwaysRun = true)
@Override
protected void setup() throws Exception {
conf.setEnableShadowTopics(false);
super.baseSetup();
}

@AfterClass(alwaysRun = true)
@Override
protected void cleanup() throws Exception {
super.internalCleanup();
}

@Override
protected String newTopicName() {
return TopicName.get(super.newTopicName()).toString();
}

@DataProvider
public Object[][] partitionCounts() {
Expand All @@ -45,8 +64,7 @@ public Object[][] partitionCounts() {

@Test(dataProvider = "partitionCounts")
public void testCreationAndConfigurationDisabled(int partitions) throws Exception {
assertThat(new ServiceConfiguration().isEnableShadowTopics()).isFalse();
assertThat(getConfig().isEnableShadowTopics()).isFalse();
assertThat(conf.isEnableShadowTopics()).isFalse();
String source = newTopicName();
String shadow = source + "-shadow";
if (partitions == 0) {
Expand Down Expand Up @@ -79,7 +97,8 @@ public void testExistingShadowTopicCannotLoad(int partitions) throws Exception {
String topicToLoad = shadow;
if (partitions == 0) {
admin.topics().createNonPartitionedTopic(shadow);
PersistentTopic topic = (PersistentTopic) getTopicIfExists(shadow).get().orElseThrow();
PersistentTopic topic = (PersistentTopic) pulsar.getBrokerService()
.getTopicIfExists(shadow).get().orElseThrow();
topic.getManagedLedger().setProperty(PROPERTY_SOURCE_TOPIC_KEY, source);
admin.topics().unload(shadow);
} else {
Expand All @@ -90,10 +109,11 @@ public void testExistingShadowTopicCannotLoad(int partitions) throws Exception {
}
String finalTopicToLoad = topicToLoad;
try {
assertThatThrownBy(() -> getTopic(finalTopicToLoad, true).get(10, TimeUnit.SECONDS))
assertThatThrownBy(() -> pulsar.getBrokerService().getTopic(finalTopicToLoad, true)
.get(10, TimeUnit.SECONDS))
.hasRootCauseInstanceOf(BrokerServiceException.NotAllowedException.class)
.hasStackTraceContaining("Shadow topics are disabled");
assertThat(getTopicReference(finalTopicToLoad)).isEmpty();
assertThat(pulsar.getBrokerService().getTopicReference(finalTopicToLoad)).isEmpty();
} finally {
if (partitions == 0) {
getPulsar().getDefaultManagedLedgerFactory().delete(
Expand All @@ -112,7 +132,8 @@ public void testExistingShadowPolicyDoesNotStartReplication() throws Exception {
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();
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();
Expand Down
Loading