You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@fluss.apache.org by ho...@apache.org on 2026/07/16 12:22:30 UTC
(fluss) branch main updated: [server] Coordinator standby tracks dynamic config changes to avoid stale config after becoming leader .(#3645)
This is an automated email from the ASF dual-hosted git repository.
loserwang1024 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/fluss.git
The following commit(s) were added to refs/heads/main by this push:
new f6aff1a7b [server] Coordinator standby tracks dynamic config changes to avoid stale config after becoming leader .(#3645)
f6aff1a7b is described below
commit f6aff1a7b720d8b21b96a65a702a6b8ce94562d4
Author: fhan <aa...@gmail.com>
AuthorDate: Thu Jul 16 20:22:25 2026 +0800
[server] Coordinator standby tracks dynamic config changes to avoid stale config after becoming leader .(#3645)
---
.../apache/fluss/server/DynamicConfigManager.java | 54 +++++-
.../server/coordinator/CoordinatorServer.java | 14 +-
.../apache/fluss/server/tablet/TabletServer.java | 2 +-
.../fluss/server/DynamicConfigChangeTest.java | 206 +++++++++++++++------
.../CoordinatorHighAvailabilityITCase.java | 69 +++++++
5 files changed, 285 insertions(+), 60 deletions(-)
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/DynamicConfigManager.java b/fluss-server/src/main/java/org/apache/fluss/server/DynamicConfigManager.java
index 74d6147c7..e6ff957c8 100644
--- a/fluss-server/src/main/java/org/apache/fluss/server/DynamicConfigManager.java
+++ b/fluss-server/src/main/java/org/apache/fluss/server/DynamicConfigManager.java
@@ -40,7 +40,21 @@ import java.util.List;
import java.util.Map;
import java.util.Objects;
-/** Manager for dynamic configurations. */
+/**
+ * Manager for dynamic configurations.
+ *
+ * <p>Used by both the CoordinatorServer and the TabletServer. A TabletServer always applies dynamic
+ * config-change notifications from ZooKeeper. A CoordinatorServer applies them only while it is a
+ * standby (follower): a standby keeps tracking changes so that, once it is promoted to leader, its
+ * components (e.g. SASL credentials) already reflect the latest config rather than a stale
+ * snapshot.
+ *
+ * <p>An active coordinator leader stops consuming notifications (see {@link #pauseListening()}),
+ * because it is itself the only writer of dynamic configs and already holds the latest values;
+ * letting it react to its own notifications could roll a config back to an older value (A to B to
+ * A). On losing leadership it resumes consuming (see {@link #resumeListening()}) and re-syncs from
+ * ZooKeeper to pick up any changes made by the new leader while it was not listening.
+ */
public class DynamicConfigManager {
private static final Logger LOG = LoggerFactory.getLogger(DynamicConfigManager.class);
private static final long CHANGE_NOTIFICATION_EXPIRATION_MS = 15 * 60 * 1000L;
@@ -48,13 +62,17 @@ public class DynamicConfigManager {
private final DynamicServerConfig dynamicServerConfig;
private final ZooKeeperClient zooKeeperClient;
private final ZkNodeChangeNotificationWatcher configChangeListener;
- private final boolean isCoordinator;
- public DynamicConfigManager(
- ZooKeeperClient zooKeeperClient, Configuration configuration, boolean isCoordinator) {
+ /**
+ * Whether change notifications are applied. Always true on a TabletServer and on a standby
+ * CoordinatorServer; set to false while a CoordinatorServer is the active leader. Volatile
+ * because it is flipped by the leader-election thread and read by the notification thread.
+ */
+ private volatile boolean listeningEnabled = true;
+
+ public DynamicConfigManager(ZooKeeperClient zooKeeperClient, Configuration configuration) {
this.dynamicServerConfig = new DynamicServerConfig(configuration);
this.zooKeeperClient = zooKeeperClient;
- this.isCoordinator = isCoordinator;
this.configChangeListener =
new ZkNodeChangeNotificationWatcher(
zooKeeperClient,
@@ -97,6 +115,28 @@ public class DynamicConfigManager {
configChangeListener.stop();
}
+ /**
+ * Stops applying config-change notifications. Called when a CoordinatorServer becomes the
+ * active leader: the leader is the sole writer of dynamic configs and already holds the latest
+ * values, so reacting to its own notifications is unnecessary and could roll a value back (A to
+ * B to A).
+ */
+ public void pauseListening() {
+ listeningEnabled = false;
+ }
+
+ /**
+ * Resumes applying config-change notifications and re-syncs the current config from ZooKeeper.
+ * Called when a CoordinatorServer loses leadership: while it was leader it ignored
+ * notifications, so it must re-read ZooKeeper once to pick up any changes made by the new
+ * leader.
+ */
+ public void resumeListening() throws Exception {
+ listeningEnabled = true;
+ Map<String, String> entityConfigs = zooKeeperClient.fetchEntityConfig();
+ dynamicServerConfig.updateDynamicConfig(entityConfigs, true);
+ }
+
public List<ConfigEntry> describeConfigs() {
Map<String, String> dynamicDefaultConfigs = dynamicServerConfig.getDynamicConfigs();
Map<String, String> staticServerConfigs = dynamicServerConfig.getInitialServerConfigs();
@@ -340,7 +380,9 @@ public class DynamicConfigManager {
@Override
public void processNotification(byte[] notification) throws Exception {
- if (isCoordinator) {
+ // An active coordinator leader is the sole writer of dynamic configs, so it ignores
+ // notifications to avoid rolling back a value it just set (A to B to A).
+ if (!listeningEnabled) {
return;
}
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorServer.java b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorServer.java
index ba9287b27..3cbaac40d 100644
--- a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorServer.java
+++ b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorServer.java
@@ -241,7 +241,7 @@ public class CoordinatorServer extends ServerBase {
this.lakeCatalogDynamicLoader = new LakeCatalogDynamicLoader(conf, pluginManager, true);
this.remoteDirDynamicLoader = new RemoteDirDynamicLoader(conf);
- this.dynamicConfigManager = new DynamicConfigManager(zkClient, conf, true);
+ this.dynamicConfigManager = new DynamicConfigManager(zkClient, conf);
this.metadataCache = new CoordinatorMetadataCache();
this.authorizer = AuthorizerLoader.createAuthorizer(conf, zkClient, pluginManager);
@@ -349,6 +349,10 @@ public class CoordinatorServer extends ServerBase {
clock);
coordinatorEventProcessor.startup();
+ // As the active leader, this server is the sole writer of dynamic configs and holds the
+ // latest values, so stop consuming change notifications to avoid rolling a value back.
+ dynamicConfigManager.pauseListening();
+
createDefaultDatabase();
}
}
@@ -420,6 +424,14 @@ public class CoordinatorServer extends ServerBase {
LOG.warn("Failed to close client metric group", t);
}
+ try {
+ // Back to standby: resume consuming config-change notifications and re-sync from
+ // ZooKeeper to pick up any changes the new leader made while we were not listening.
+ dynamicConfigManager.resumeListening();
+ } catch (Throwable t) {
+ LOG.warn("Failed to resume dynamic config listening", t);
+ }
+
LOG.info("Coordinator leader services cleaned up successfully.");
}
}
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/tablet/TabletServer.java b/fluss-server/src/main/java/org/apache/fluss/server/tablet/TabletServer.java
index 7cf7d2678..12a18a70f 100644
--- a/fluss-server/src/main/java/org/apache/fluss/server/tablet/TabletServer.java
+++ b/fluss-server/src/main/java/org/apache/fluss/server/tablet/TabletServer.java
@@ -234,7 +234,7 @@ public class TabletServer extends ServerBase {
new LakeCatalogDynamicLoader(conf, pluginManager, false);
MetadataManager metadataManager =
new MetadataManager(zkClient, conf, lakeCatalogDynamicLoader);
- this.dynamicConfigManager = new DynamicConfigManager(zkClient, conf, false);
+ this.dynamicConfigManager = new DynamicConfigManager(zkClient, conf);
this.metadataCache = new TabletServerMetadataCache(metadataManager);
diff --git a/fluss-server/src/test/java/org/apache/fluss/server/DynamicConfigChangeTest.java b/fluss-server/src/test/java/org/apache/fluss/server/DynamicConfigChangeTest.java
index a311a2986..f8fadffb7 100644
--- a/fluss-server/src/test/java/org/apache/fluss/server/DynamicConfigChangeTest.java
+++ b/fluss-server/src/test/java/org/apache/fluss/server/DynamicConfigChangeTest.java
@@ -34,7 +34,8 @@ import org.apache.fluss.server.zk.NOPErrorHandler;
import org.apache.fluss.server.zk.ZooKeeperClient;
import org.apache.fluss.server.zk.ZooKeeperExtension;
import org.apache.fluss.server.zk.ZooKeeperUtils;
-import org.apache.fluss.server.zk.data.ZkData.ConfigEntityZNode;
+import org.apache.fluss.server.zk.data.ZkData.ConfigZNode;
+import org.apache.fluss.shaded.zookeeper3.org.apache.zookeeper.KeeperException;
import org.apache.fluss.testutils.common.AllCallbackWrapper;
import org.junit.jupiter.api.AfterAll;
@@ -46,9 +47,11 @@ import org.junit.jupiter.api.io.TempDir;
import java.io.File;
import java.time.Duration;
+import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
+import java.util.List;
import java.util.Map;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
@@ -70,6 +73,13 @@ public class DynamicConfigChangeTest {
protected static ZooKeeperClient zookeeperClient;
+ /**
+ * Managers created during a test. Each one starts a ZooKeeper notification watcher on {@link
+ * DynamicConfigManager#startup()}, so they must be closed after the test to stop reacting to
+ * config changes written by later tests that share the same static {@link #zookeeperClient}.
+ */
+ private final List<DynamicConfigManager> managers = new ArrayList<>();
+
@BeforeAll
static void beforeAll() {
final Configuration configuration = new Configuration();
@@ -90,15 +100,51 @@ public class DynamicConfigChangeTest {
@AfterEach
void after() throws Exception {
- zookeeperClient.deletePath(ConfigEntityZNode.path());
+ for (DynamicConfigManager manager : managers) {
+ manager.close();
+ }
+ managers.clear();
+ // Recursively delete /config so both the entity config (/config/server/global) and the
+ // change notifications (/config/changes/...) are removed for the next test.
+ try {
+ zookeeperClient
+ .getCuratorClient()
+ .delete()
+ .deletingChildrenIfNeeded()
+ .forPath(ConfigZNode.path());
+ } catch (KeeperException.NoNodeException ignored) {
+ // Nothing was written in this test.
+ }
+ }
+
+ /**
+ * Creates a manager that does not consume change notifications, modelling a coordinator leader
+ * (the sole writer via {@code alterConfigs}). This is the default because most tests exercise
+ * the write path, and a self-consumed notification could roll back a value the test just set.
+ * Use {@link #createListeningManager} for tests that verify notification tracking. The manager
+ * is tracked so {@link #after()} closes it.
+ */
+ private DynamicConfigManager createManager(Configuration configuration) {
+ DynamicConfigManager manager = createListeningManager(configuration);
+ manager.pauseListening();
+ return manager;
+ }
+
+ /**
+ * Creates a manager that consumes change notifications, modelling a standby coordinator or a
+ * tablet server. The manager is tracked so {@link #after()} closes it (stopping its watcher).
+ */
+ private DynamicConfigManager createListeningManager(Configuration configuration) {
+ DynamicConfigManager manager = new DynamicConfigManager(zookeeperClient, configuration);
+ managers.add(manager);
+ return manager;
}
@Test
void testAlterLakehouseConfigs() throws Exception {
try (LakeCatalogDynamicLoader lakeCatalogDynamicLoader =
new LakeCatalogDynamicLoader(new Configuration(), null, true)) {
- DynamicConfigManager dynamicConfigManager =
- new DynamicConfigManager(zookeeperClient, new Configuration(), true);
+ DynamicConfigManager dynamicConfigManager = createManager(new Configuration());
dynamicConfigManager.register(lakeCatalogDynamicLoader);
dynamicConfigManager.startup();
assertThatThrownBy(
@@ -149,8 +195,7 @@ public class DynamicConfigChangeTest {
configuration.setString(DATALAKE_FORMAT.key(), "paimon");
try (LakeCatalogDynamicLoader lakeCatalogDynamicLoader =
new LakeCatalogDynamicLoader(configuration, null, true)) {
- DynamicConfigManager dynamicConfigManager =
- new DynamicConfigManager(zookeeperClient, configuration, true);
+ DynamicConfigManager dynamicConfigManager = createManager(configuration);
dynamicConfigManager.register(lakeCatalogDynamicLoader);
dynamicConfigManager.startup();
assertThat(lakeCatalogDynamicLoader.getLakeCatalogContainer().getDataLakeFormat())
@@ -174,8 +219,7 @@ public class DynamicConfigChangeTest {
Configuration configuration = new Configuration();
try (LakeCatalogDynamicLoader lakeCatalogDynamicLoader =
new LakeCatalogDynamicLoader(configuration, null, true)) {
- DynamicConfigManager dynamicConfigManager =
- new DynamicConfigManager(zookeeperClient, configuration, true);
+ DynamicConfigManager dynamicConfigManager = createManager(configuration);
dynamicConfigManager.register(lakeCatalogDynamicLoader);
dynamicConfigManager.startup();
assertThatThrownBy(
@@ -199,8 +243,7 @@ public class DynamicConfigChangeTest {
Configuration configuration = new Configuration();
try (LakeCatalogDynamicLoader lakeCatalogDynamicLoader =
new LakeCatalogDynamicLoader(configuration, null, true)) {
- DynamicConfigManager dynamicConfigManager =
- new DynamicConfigManager(zookeeperClient, configuration, true);
+ DynamicConfigManager dynamicConfigManager = createManager(configuration);
dynamicConfigManager.register(lakeCatalogDynamicLoader);
dynamicConfigManager.startup();
@@ -235,8 +278,7 @@ public class DynamicConfigChangeTest {
Configuration configuration = new Configuration();
try (LakeCatalogDynamicLoader lakeCatalogDynamicLoader =
new LakeCatalogDynamicLoader(configuration, null, true)) {
- DynamicConfigManager dynamicConfigManager =
- new DynamicConfigManager(zookeeperClient, configuration, true);
+ DynamicConfigManager dynamicConfigManager = createManager(configuration);
dynamicConfigManager.register(lakeCatalogDynamicLoader);
dynamicConfigManager.startup();
assertThatThrownBy(
@@ -264,8 +306,7 @@ public class DynamicConfigChangeTest {
Configuration configuration = new Configuration();
try (LakeCatalogDynamicLoader lakeCatalogDynamicLoader =
new LakeCatalogDynamicLoader(configuration, null, false)) {
- DynamicConfigManager dynamicConfigManager =
- new DynamicConfigManager(zookeeperClient, configuration, false);
+ DynamicConfigManager dynamicConfigManager = createListeningManager(configuration);
dynamicConfigManager.register(lakeCatalogDynamicLoader);
dynamicConfigManager.startup();
assertThat(lakeCatalogDynamicLoader.getLakeCatalogContainer().getDataLakeFormat())
@@ -285,6 +326,86 @@ public class DynamicConfigChangeTest {
}
}
+ /**
+ * Regression test for #3625: a {@link DynamicConfigManager} must keep applying config-change
+ * notifications pushed to ZooKeeper by another writer. This is what lets a standby
+ * CoordinatorServer stay current so that, once promoted, it does not run with stale config.
+ */
+ @Test
+ void testTracksConfigChangeNotificationsFromOtherWriter() throws Exception {
+ Configuration configuration = new Configuration();
+ configuration.set(ConfigOptions.LOG_REPLICA_MIN_IN_SYNC_REPLICAS_NUMBER, 1);
+
+ DynamicConfigManager dynamicConfigManager = createListeningManager(configuration);
+ AtomicInteger reconfiguredValue = new AtomicInteger();
+ dynamicConfigManager.register(
+ new ServerReconfigurable() {
+ @Override
+ public void validate(Configuration newConfig) throws ConfigException {}
+
+ @Override
+ public void reconfigure(Configuration newConfig) {
+ reconfiguredValue.set(
+ newConfig.get(
+ ConfigOptions.LOG_REPLICA_MIN_IN_SYNC_REPLICAS_NUMBER));
+ }
+ });
+ dynamicConfigManager.startup();
+
+ // Simulate another server (e.g. the active leader) persisting a config change to ZK,
+ // which inserts a change notification the manager under test must react to.
+ Map<String, String> config = new HashMap<>();
+ config.put(ConfigOptions.LOG_REPLICA_MIN_IN_SYNC_REPLICAS_NUMBER.key(), "3");
+ zookeeperClient.upsertServerEntityConfig(config);
+
+ retry(Duration.ofMinutes(1), () -> assertThat(reconfiguredValue.get()).isEqualTo(3));
+ }
+
+ /**
+ * Regression test for #3645 review: an active leader ({@link
+ * DynamicConfigManager#pauseListening()}) must ignore change notifications so it cannot roll a
+ * value it just set back to an older one (A to B to A). On losing leadership ({@link
+ * DynamicConfigManager#resumeListening()}) it re-syncs from ZooKeeper.
+ */
+ @Test
+ void testPausedManagerIgnoresNotificationsAndResumeReSyncs() throws Exception {
+ Configuration configuration = new Configuration();
+ configuration.set(ConfigOptions.LOG_REPLICA_MIN_IN_SYNC_REPLICAS_NUMBER, 1);
+
+ DynamicConfigManager dynamicConfigManager = createListeningManager(configuration);
+ // Seed with the initial value; reconfigure() is only invoked on an effective change.
+ AtomicInteger reconfiguredValue = new AtomicInteger(1);
+ dynamicConfigManager.register(
+ new ServerReconfigurable() {
+ @Override
+ public void validate(Configuration newConfig) throws ConfigException {}
+
+ @Override
+ public void reconfigure(Configuration newConfig) {
+ reconfiguredValue.set(
+ newConfig.get(
+ ConfigOptions.LOG_REPLICA_MIN_IN_SYNC_REPLICAS_NUMBER));
+ }
+ });
+ dynamicConfigManager.startup();
+
+ // Become leader: stop consuming notifications.
+ dynamicConfigManager.pauseListening();
+
+ // Another writer changes the config in ZooKeeper. A paused manager must not apply it.
+ Map<String, String> config = new HashMap<>();
+ config.put(ConfigOptions.LOG_REPLICA_MIN_IN_SYNC_REPLICAS_NUMBER.key(), "3");
+ zookeeperClient.upsertServerEntityConfig(config);
+
+ // Give the notification a chance to be (wrongly) applied, then assert it was ignored.
+ Thread.sleep(2000);
+ assertThat(reconfiguredValue.get()).isEqualTo(1);
+
+ // Lose leadership: resume and re-sync from ZooKeeper picks up the missed change.
+ dynamicConfigManager.resumeListening();
+ assertThat(reconfiguredValue.get()).isEqualTo(3);
+ }
+
@Test
void testReStartupContainsNoMatchedDynamicConfig() throws Exception {
Configuration configuration = new Configuration();
@@ -297,8 +418,7 @@ public class DynamicConfigChangeTest {
try (LakeCatalogDynamicLoader lakeCatalogDynamicLoader =
new LakeCatalogDynamicLoader(configuration, null, true)) {
- DynamicConfigManager dynamicConfigManager =
- new DynamicConfigManager(zookeeperClient, configuration, true);
+ DynamicConfigManager dynamicConfigManager = createManager(configuration);
dynamicConfigManager.register(lakeCatalogDynamicLoader);
// Startup dynamic manager even is not matched now.
dynamicConfigManager.startup();
@@ -313,8 +433,7 @@ public class DynamicConfigChangeTest {
Configuration configuration = new Configuration();
configuration.setString(ConfigOptions.KV_SHARED_RATE_LIMITER_BYTES_PER_SEC.key(), "100MB");
- DynamicConfigManager dynamicConfigManager =
- new DynamicConfigManager(zookeeperClient, configuration, true);
+ DynamicConfigManager dynamicConfigManager = createManager(configuration);
dynamicConfigManager.startup();
// Try to set rate limiter to an invalid value - should be rejected by generic type
@@ -340,8 +459,7 @@ public class DynamicConfigChangeTest {
Configuration configuration = new Configuration();
configuration.setString(ConfigOptions.KV_SHARED_RATE_LIMITER_BYTES_PER_SEC.key(), "100MB");
- DynamicConfigManager dynamicConfigManager =
- new DynamicConfigManager(zookeeperClient, configuration, true);
+ DynamicConfigManager dynamicConfigManager = createManager(configuration);
dynamicConfigManager.startup();
// Adjust rate limiter value - should succeed
@@ -365,8 +483,7 @@ public class DynamicConfigChangeTest {
try (LakeCatalogDynamicLoader lakeCatalogDynamicLoader =
new LakeCatalogDynamicLoader(configuration, null, true)) {
- DynamicConfigManager dynamicConfigManager =
- new DynamicConfigManager(zookeeperClient, configuration, true);
+ DynamicConfigManager dynamicConfigManager = createManager(configuration);
// Register reconfigurables - generic type validation works automatically
dynamicConfigManager.register(lakeCatalogDynamicLoader);
@@ -397,8 +514,7 @@ public class DynamicConfigChangeTest {
Configuration configuration = new Configuration();
configuration.set(ConfigOptions.KV_SNAPSHOT_INTERVAL, Duration.ofMinutes(10));
- DynamicConfigManager dynamicConfigManager =
- new DynamicConfigManager(zookeeperClient, configuration, true);
+ DynamicConfigManager dynamicConfigManager = createManager(configuration);
dynamicConfigManager.startup();
// Try to set snapshot interval to an invalid value - should be rejected by type validation
@@ -420,8 +536,7 @@ public class DynamicConfigChangeTest {
Configuration configuration = new Configuration();
configuration.set(ConfigOptions.KV_SNAPSHOT_INTERVAL, Duration.ofMinutes(10));
- DynamicConfigManager dynamicConfigManager =
- new DynamicConfigManager(zookeeperClient, configuration, true);
+ DynamicConfigManager dynamicConfigManager = createManager(configuration);
AtomicReference<Duration> reconfiguredInterval = new AtomicReference<>();
dynamicConfigManager.register(
@@ -461,8 +576,7 @@ public class DynamicConfigChangeTest {
try (RemoteDirDynamicLoader remoteDirDynamicLoader =
new RemoteDirDynamicLoader(configuration)) {
- DynamicConfigManager dynamicConfigManager =
- new DynamicConfigManager(zookeeperClient, configuration, true);
+ DynamicConfigManager dynamicConfigManager = createManager(configuration);
dynamicConfigManager.register(remoteDirDynamicLoader);
dynamicConfigManager.startup();
@@ -509,8 +623,7 @@ public class DynamicConfigChangeTest {
Configuration configuration = new Configuration();
configuration.setInt(ConfigOptions.LOG_REPLICA_MIN_IN_SYNC_REPLICAS_NUMBER, 1);
- DynamicConfigManager dynamicConfigManager =
- new DynamicConfigManager(zookeeperClient, configuration, true);
+ DynamicConfigManager dynamicConfigManager = createManager(configuration);
dynamicConfigManager.startup();
// Try to set min-in-sync-replicas to an invalid value - should be rejected by type
@@ -536,8 +649,7 @@ public class DynamicConfigChangeTest {
Configuration configuration = new Configuration();
configuration.setInt(ConfigOptions.LOG_REPLICA_MIN_IN_SYNC_REPLICAS_NUMBER, 1);
- DynamicConfigManager dynamicConfigManager =
- new DynamicConfigManager(zookeeperClient, configuration, true);
+ DynamicConfigManager dynamicConfigManager = createManager(configuration);
AtomicInteger reconfiguredValue = new AtomicInteger();
dynamicConfigManager.register(
@@ -575,8 +687,7 @@ public class DynamicConfigChangeTest {
void testExplicitDataLakeEnabledRequiresDataLakeFormat() throws Exception {
try (LakeCatalogDynamicLoader lakeCatalogDynamicLoader =
new LakeCatalogDynamicLoader(new Configuration(), null, true)) {
- DynamicConfigManager dynamicConfigManager =
- new DynamicConfigManager(zookeeperClient, new Configuration(), true);
+ DynamicConfigManager dynamicConfigManager = createManager(new Configuration());
dynamicConfigManager.register(lakeCatalogDynamicLoader);
dynamicConfigManager.startup();
@@ -647,8 +758,7 @@ public class DynamicConfigChangeTest {
configuration.setString(ConfigOptions.DATA_DIR, dataDir.getAbsolutePath());
configuration.set(ConfigOptions.SERVER_DATA_DISK_WRITE_LIMIT_RATIO, 0.85);
- DynamicConfigManager dynamicConfigManager =
- new DynamicConfigManager(zookeeperClient, configuration, true);
+ DynamicConfigManager dynamicConfigManager = createManager(configuration);
// Create LocalDiskManager and register it
try (LocalDiskManager localDiskManager = LocalDiskManager.create(configuration)) {
@@ -711,10 +821,8 @@ public class DynamicConfigChangeTest {
}
}
- private static DynamicConfigManager createDiskWriteLimitDynamicConfigManager()
- throws Exception {
- DynamicConfigManager dynamicConfigManager =
- new DynamicConfigManager(zookeeperClient, new Configuration(), true);
+ private DynamicConfigManager createDiskWriteLimitDynamicConfigManager() throws Exception {
+ DynamicConfigManager dynamicConfigManager = createManager(new Configuration());
dynamicConfigManager.register(new DiskWriteLimitConfigValidator());
dynamicConfigManager.startup();
return dynamicConfigManager;
@@ -745,8 +853,7 @@ public class DynamicConfigChangeTest {
@Test
void testAppendAndSubtractOnMapConfig() throws Exception {
Configuration configuration = new Configuration();
- DynamicConfigManager dynamicConfigManager =
- new DynamicConfigManager(zookeeperClient, configuration, true);
+ DynamicConfigManager dynamicConfigManager = createManager(configuration);
AtomicReference<Map<String, String>> reconfiguredUsers = new AtomicReference<>();
dynamicConfigManager.register(
@@ -821,8 +928,7 @@ public class DynamicConfigChangeTest {
@Test
void testAppendOnNonListOrMapConfigIsRejected() throws Exception {
Configuration configuration = new Configuration();
- DynamicConfigManager dynamicConfigManager =
- new DynamicConfigManager(zookeeperClient, configuration, true);
+ DynamicConfigManager dynamicConfigManager = createManager(configuration);
dynamicConfigManager.startup();
// APPEND on a non-list/non-map config (kv.snapshot.interval is Duration type)
@@ -862,8 +968,7 @@ public class DynamicConfigChangeTest {
"org.apache.fluss.security.auth.sasl.plain.PlainLoginModule required "
+ "user_admin=\"admin-secret\";");
- DynamicConfigManager dynamicConfigManager =
- new DynamicConfigManager(zookeeperClient, configuration, true);
+ DynamicConfigManager dynamicConfigManager = createManager(configuration);
dynamicConfigManager.startup();
assertThat(dynamicConfigManager.describeConfigs())
@@ -884,8 +989,7 @@ public class DynamicConfigChangeTest {
// Pre-configure a static user in server.yaml equivalent
configuration.setString(ConfigOptions.SERVER_SASL_CREDENTIALS.key(), "admin:admin-secret");
- DynamicConfigManager dynamicConfigManager =
- new DynamicConfigManager(zookeeperClient, configuration, true);
+ DynamicConfigManager dynamicConfigManager = createManager(configuration);
AtomicReference<Map<String, String>> reconfiguredUsers = new AtomicReference<>();
dynamicConfigManager.register(
@@ -922,8 +1026,7 @@ public class DynamicConfigChangeTest {
// Pre-configure a static user in server.yaml equivalent
configuration.setString(ConfigOptions.SERVER_SASL_CREDENTIALS.key(), "admin:admin-secret");
- DynamicConfigManager dynamicConfigManager =
- new DynamicConfigManager(zookeeperClient, configuration, true);
+ DynamicConfigManager dynamicConfigManager = createManager(configuration);
AtomicReference<Map<String, String>> reconfiguredUsers = new AtomicReference<>();
dynamicConfigManager.register(
@@ -955,8 +1058,7 @@ public class DynamicConfigChangeTest {
@Test
void testSubtractTrimsWhitespaceAndRemovesByKey() throws Exception {
Configuration configuration = new Configuration();
- DynamicConfigManager dynamicConfigManager =
- new DynamicConfigManager(zookeeperClient, configuration, true);
+ DynamicConfigManager dynamicConfigManager = createManager(configuration);
dynamicConfigManager.startup();
// Set up a config with whitespace around entries
diff --git a/fluss-server/src/test/java/org/apache/fluss/server/coordinator/CoordinatorHighAvailabilityITCase.java b/fluss-server/src/test/java/org/apache/fluss/server/coordinator/CoordinatorHighAvailabilityITCase.java
index 23b8a70bf..20f34e3ad 100644
--- a/fluss-server/src/test/java/org/apache/fluss/server/coordinator/CoordinatorHighAvailabilityITCase.java
+++ b/fluss-server/src/test/java/org/apache/fluss/server/coordinator/CoordinatorHighAvailabilityITCase.java
@@ -23,6 +23,9 @@ import org.apache.fluss.cluster.ServerNode;
import org.apache.fluss.cluster.ServerType;
import org.apache.fluss.config.ConfigOptions;
import org.apache.fluss.config.Configuration;
+import org.apache.fluss.config.cluster.AlterConfig;
+import org.apache.fluss.config.cluster.AlterConfigOpType;
+import org.apache.fluss.config.cluster.ConfigEntry;
import org.apache.fluss.exception.InvalidCoordinatorException;
import org.apache.fluss.exception.NotCoordinatorLeaderException;
import org.apache.fluss.metadata.TableBucket;
@@ -222,6 +225,72 @@ class CoordinatorHighAvailabilityITCase {
createGatewayForServer(leader).metadata(new MetadataRequest()).get();
}
+ /**
+ * Regression test for #3625: a standby coordinator must keep applying dynamic config changes so
+ * that, after failover, the promoted leader uses the latest config rather than a stale snapshot
+ * taken when it started up as standby.
+ */
+ @Test
+ void testStandbyTracksDynamicConfigAndNewLeaderUsesItAfterFailover() throws Exception {
+ coordinatorServer1 = new CoordinatorServer(createConfiguration());
+ coordinatorServer2 = new CoordinatorServer(createConfiguration());
+
+ coordinatorServer1.start();
+ coordinatorServer2.start();
+
+ waitUntilCoordinatorServerElected();
+ CoordinatorAddress firstLeaderAddr = zookeeperClient.getCoordinatorLeaderAddress().get();
+
+ CoordinatorServer leader = findServerById(firstLeaderAddr.getId());
+ CoordinatorServer standby = findServerByNotId(firstLeaderAddr.getId());
+ assertThat(leader).isNotNull();
+ assertThat(standby).isNotNull();
+
+ String configKey = ConfigOptions.LOG_REPLICA_MIN_IN_SYNC_REPLICAS_NUMBER.key();
+
+ // Alter a dynamic config through the current leader. This persists it to ZK and inserts a
+ // change notification that every server (including the standby) should react to.
+ leader.getDynamicConfigManager()
+ .alterConfigs(
+ Collections.singletonList(
+ new AlterConfig(configKey, "3", AlterConfigOpType.SET)));
+
+ // The standby must pick up the change while it is still a standby.
+ waitUntil(
+ () -> "3".equals(dynamicConfigValue(standby, configKey)),
+ Duration.ofSeconds(30),
+ "Standby coordinator did not apply dynamic config change while standby");
+
+ // Fail over: killing the leader's ZK session promotes the standby to leader.
+ killZkSession(leader);
+ waitUntilNewLeaderElected(leader.getServerId());
+ assertThat(zookeeperClient.getCoordinatorLeaderAddress().get().getId())
+ .as("After killing leader, standby should become leader")
+ .isEqualTo(standby.getServerId());
+
+ // The newly promoted leader must still see the up-to-date value, not a stale one.
+ assertThat(dynamicConfigValue(standby, configKey))
+ .as("Newly promoted leader should use the dynamic config it tracked while standby")
+ .isEqualTo("3");
+
+ // As the new leader it is now the sole writer: a further alter must take effect and stick.
+ standby.getDynamicConfigManager()
+ .alterConfigs(
+ Collections.singletonList(
+ new AlterConfig(configKey, "5", AlterConfigOpType.SET)));
+ assertThat(dynamicConfigValue(standby, configKey)).isEqualTo("5");
+ }
+
+ /** Reads the effective value of a config key from a server's DynamicConfigManager. */
+ private static String dynamicConfigValue(CoordinatorServer server, String configKey) {
+ for (ConfigEntry entry : server.getDynamicConfigManager().describeConfigs()) {
+ if (entry.key().equals(configKey)) {
+ return entry.value();
+ }
+ }
+ return null;
+ }
+
@Test
void testMultipleLeadershipLossAndRecovery() throws Exception {
coordinatorServer1 = new CoordinatorServer(createConfiguration());