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/23 10:08:48 UTC
(pulsar) 17/23: [fix][broker] Don't swap the local metadata store's synchronizer when configurationMetadataSyncEventTopic is updated on a shared store (#22695)
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-5.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit d8f3da9d71edab9f368bdee6d042673074aa0849
Author: fengyubiao <yu...@streamnative.io>
AuthorDate: Fri Oct 23 09:41:48 2026 +0800
[fix][broker] Don't swap the local metadata store's synchronizer when configurationMetadataSyncEventTopic is updated on a shared store (#22695)
Co-authored-by: Matteo Merli <mm...@apache.org>
(cherry picked from commit ec8456bb04397538de7f678f5c4700d9bb3a1360)
---
.../org/apache/pulsar/broker/PulsarService.java | 17 +++++
.../pulsar/broker/admin/impl/BrokersBase.java | 6 ++
.../SyncConfigStoreSharedWithLocalTest.java | 89 ++++++++++++++++++++++
3 files changed, 112 insertions(+)
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/PulsarService.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/PulsarService.java
index 0bfeeb4e9a5e..8da321abf029 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/PulsarService.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/PulsarService.java
@@ -2451,6 +2451,14 @@ public class PulsarService implements AutoCloseable, ShutdownService {
}
}
+ /**
+ * Whether the configuration metadata store and the local metadata store are the same instance, which happens
+ * when both use the same URL.
+ */
+ public boolean isConfigurationMetadataStoreSharedWithLocal() {
+ return configurationMetadataStore == localMetadataStore;
+ }
+
public void initConfigMetadataSynchronizerIfNeeded() {
mutex.lock();
try {
@@ -2464,6 +2472,15 @@ public class PulsarService implements AutoCloseable, ShutdownService {
+ " does not support.");
return;
}
+ // Skip if the configuration metadata store is the local one: swapping its synchronizer would make every
+ // local metadata write depend on the configuration sync topic.
+ if (isConfigurationMetadataStoreSharedWithLocal()) {
+ log.warn()
+ .attr("topic", newTopic)
+ .log("Skip to update Metadata Synchronizer because the Configuration Metadata Store is the"
+ + " same as the Local Metadata Store.");
+ return;
+ }
// Skip if no changes.
// case-1: both null.
// case-2: both topics are the same.
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/BrokersBase.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/BrokersBase.java
index 12ab07609757..d38484492365 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/BrokersBase.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/BrokersBase.java
@@ -342,6 +342,12 @@ public class BrokersBase extends AdminResource {
return FutureUtil
.failedFuture(new RestException(Status.PRECONDITION_FAILED, " Invalid dynamic-config value"));
}
+ if ("configurationMetadataSyncEventTopic".equals(configName)
+ && pulsar().isConfigurationMetadataStoreSharedWithLocal()) {
+ return FutureUtil.failedFuture(new RestException(Status.PRECONDITION_FAILED,
+ "Can't update configurationMetadataSyncEventTopic because the configuration metadata store is"
+ + " the same as the local metadata store"));
+ }
if (pulsar().getBrokerService().isDynamicConfiguration(configName)) {
return dynamicConfigurationResources().setDynamicConfigurationWithCreateAsync(old -> {
Map<String, String> configurationMap = old.orElseGet(Maps::newHashMap);
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/SyncConfigStoreSharedWithLocalTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/SyncConfigStoreSharedWithLocalTest.java
new file mode 100644
index 000000000000..8edf063584d0
--- /dev/null
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/SyncConfigStoreSharedWithLocalTest.java
@@ -0,0 +1,89 @@
+/*
+ * 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;
+
+import static org.apache.pulsar.common.naming.NamespaceName.SYSTEM_NAMESPACE;
+import static org.testng.Assert.assertFalse;
+import static org.testng.Assert.assertNull;
+import static org.testng.Assert.assertThrows;
+import static org.testng.Assert.assertTrue;
+import org.apache.pulsar.broker.ServiceConfiguration;
+import org.apache.pulsar.client.admin.PulsarAdminException;
+import org.apache.pulsar.common.naming.TopicDomain;
+import org.apache.pulsar.zookeeper.LocalBookkeeperEnsemble;
+import org.apache.pulsar.zookeeper.ZookeeperServerTest;
+import org.testng.annotations.AfterClass;
+import org.testng.annotations.BeforeClass;
+import org.testng.annotations.Test;
+
+/**
+ * Covers brokers whose configuration metadata store URL is the same as the local one, so both use one store instance.
+ */
+@Test(groups = "broker")
+public class SyncConfigStoreSharedWithLocalTest extends GeoReplicationWithConfigurationSyncTestBase {
+
+ private static final String CONF_NAME_SYNC_EVENT_TOPIC = "configurationMetadataSyncEventTopic";
+ private static final String SYNC_EVENT_TOPIC = TopicDomain.persistent.value() + "://" + SYSTEM_NAMESPACE
+ + "/__sync_config_meta";
+
+ @Override
+ @BeforeClass(alwaysRun = true, timeOut = 300000)
+ public void setup() throws Exception {
+ super.setup();
+ }
+
+ @Override
+ @AfterClass(alwaysRun = true, timeOut = 300000)
+ public void cleanup() throws Exception {
+ super.cleanup();
+ }
+
+ @Override
+ protected void setConfigDefaults(ServiceConfiguration config, String clusterName,
+ LocalBookkeeperEnsemble bookkeeperEnsemble, ZookeeperServerTest brokerConfigZk) {
+ super.setConfigDefaults(config, clusterName, bookkeeperEnsemble, brokerConfigZk);
+ config.setConfigurationMetadataStoreUrl(config.getMetadataStoreUrl());
+ }
+
+ @Test
+ public void testRejectDynamicUpdateOfConfigurationMetadataSyncEventTopic() throws Exception {
+ assertTrue(pulsar1.isConfigurationMetadataStoreSharedWithLocal());
+
+ assertThrows(PulsarAdminException.PreconditionFailedException.class,
+ () -> admin1.brokers().updateDynamicConfiguration(CONF_NAME_SYNC_EVENT_TOPIC, SYNC_EVENT_TOPIC));
+ assertFalse(admin1.brokers().getAllDynamicConfigurations().containsKey(CONF_NAME_SYNC_EVENT_TOPIC));
+ }
+
+ @Test
+ public void testSkipSynchronizerSwapKeepsLocalWritesWorking() throws Exception {
+ // The value can still reach the broker without the admin endpoint, e.g. persisted before an upgrade.
+ pulsar1.getConfig().setConfigurationMetadataSyncEventTopic(SYNC_EVENT_TOPIC);
+ try {
+ pulsar1.initConfigMetadataSynchronizerIfNeeded();
+ assertNull(pulsar1.getConfigMetadataSynchronizer());
+
+ // Without the guard, these writes fail because the shared store waits for the sync topic producer.
+ admin1.namespaces().createNamespace(defaultTenant + "/ns-after-sync-skip");
+ admin1.topics().createNonPartitionedTopic(
+ "persistent://" + defaultTenant + "/ns-after-sync-skip/tp-after-sync-skip");
+ } finally {
+ pulsar1.getConfig().setConfigurationMetadataSyncEventTopic(null);
+ }
+ }
+}