You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@fluss.apache.org by ja...@apache.org on 2026/04/09 11:37:13 UTC
(fluss) 01/02: [server] Coordinator supports coordinator epoch protection (#2781)
This is an automated email from the ASF dual-hosted git repository.
jark pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/fluss.git
commit 04e1a9f57a15aec36b977185268ca791d305cca8
Author: whenzhou.zc <wh...@alibaba-inc.com>
AuthorDate: Thu Jul 24 16:08:18 2025 +0800
[server] Coordinator supports coordinator epoch protection (#2781)
---
.../exception/CoordinatorEpochFencedException.java | 26 +++
.../server/coordinator/CoordinatorContext.java | 16 +-
.../coordinator/CoordinatorEventProcessor.java | 25 ++-
.../coordinator/CoordinatorLeaderElection.java | 22 +-
.../coordinator/CoordinatorRequestBatch.java | 1 +
.../server/coordinator/CoordinatorServer.java | 28 ++-
.../statemachine/ReplicaStateMachine.java | 3 +-
.../statemachine/TableBucketStateMachine.java | 18 +-
.../metrics/group/CoordinatorMetricGroup.java | 2 +-
.../fluss/server/replica/ReplicaManager.java | 5 +
.../fluss/server/utils/ServerRpcMessageUtils.java | 4 +
.../java/org/apache/fluss/server/zk/ZkEpoch.java | 42 ++++
.../apache/fluss/server/zk/ZooKeeperClient.java | 249 +++++++++++++++++++--
.../org/apache/fluss/server/zk/ZooKeeperOp.java | 64 ++++++
.../org/apache/fluss/server/zk/data/ZkData.java | 18 ++
.../org/apache/fluss/server/zk/data/ZkVersion.java | 35 +++
.../coordinator/CoordinatorChannelManagerTest.java | 1 +
.../coordinator/CoordinatorEventProcessorTest.java | 6 +-
.../CoordinatorHighAvailabilityITCase.java | 138 ++++++++++++
.../coordinator/CoordinatorServerElectionTest.java | 6 +
.../coordinator/CoordinatorServerITCase.java | 1 -
.../server/coordinator/TableManagerITCase.java | 10 +-
.../fluss/server/coordinator/TableManagerTest.java | 2 +-
.../server/coordinator/TestCoordinatorContext.java | 39 ++++
.../rebalance/RebalanceManagerTest.java | 6 +-
.../statemachine/ReplicaStateMachineTest.java | 18 +-
.../statemachine/TableBucketStateMachineTest.java | 21 +-
.../metadata/ZkBasedMetadataProviderTest.java | 25 ++-
.../fluss/server/tablet/TabletServiceITCase.java | 1 +
.../fluss/server/zk/ZooKeeperClientTest.java | 43 ++--
30 files changed, 798 insertions(+), 77 deletions(-)
diff --git a/fluss-common/src/main/java/org/apache/fluss/exception/CoordinatorEpochFencedException.java b/fluss-common/src/main/java/org/apache/fluss/exception/CoordinatorEpochFencedException.java
new file mode 100644
index 000000000..0e0e4e409
--- /dev/null
+++ b/fluss-common/src/main/java/org/apache/fluss/exception/CoordinatorEpochFencedException.java
@@ -0,0 +1,26 @@
+/*
+ * 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.fluss.exception;
+
+/** Exception thrown when the Coordinator leader epoch is fenced. */
+public class CoordinatorEpochFencedException extends RuntimeException {
+ public CoordinatorEpochFencedException(String message) {
+ super(message);
+ }
+}
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorContext.java b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorContext.java
index 57ed11528..138ddb175 100644
--- a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorContext.java
+++ b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorContext.java
@@ -55,6 +55,7 @@ public class CoordinatorContext {
private static final Logger LOG = LoggerFactory.getLogger(CoordinatorContext.class);
public static final int INITIAL_COORDINATOR_EPOCH = 0;
+ public static final int INITIAL_COORDINATOR_EPOCH_ZK_VERSION = 0;
// for simplicity, we just use retry time, may consider make it a configurable value
// and use combine retry times and retry delay
@@ -109,6 +110,7 @@ public class CoordinatorContext {
private ServerInfo coordinatorServerInfo = null;
private int coordinatorEpoch = INITIAL_COORDINATOR_EPOCH;
+ private int coordinatorEpochZkVersion = INITIAL_COORDINATOR_EPOCH_ZK_VERSION;
public CoordinatorContext() {}
@@ -116,6 +118,15 @@ public class CoordinatorContext {
return coordinatorEpoch;
}
+ public int getCoordinatorEpochZkVersion() {
+ return coordinatorEpochZkVersion;
+ }
+
+ public void setCoordinatorEpochAndZkVersion(int newEpoch, int newZkVersion) {
+ this.coordinatorEpoch = newEpoch;
+ this.coordinatorEpochZkVersion = newZkVersion;
+ }
+
public Set<String> getLiveCoordinatorServers() {
return liveCoordinatorServers;
}
@@ -708,12 +719,13 @@ public class CoordinatorContext {
public void resetContext() {
tablesToBeDeleted.clear();
- coordinatorEpoch = 0;
+ coordinatorEpoch = INITIAL_COORDINATOR_EPOCH;
+ coordinatorEpochZkVersion = INITIAL_COORDINATOR_EPOCH_ZK_VERSION;
clearTablesState();
liveTabletServers.clear();
+ liveCoordinatorServers.clear();
shuttingDownTabletServers.clear();
serverTags.clear();
- liveCoordinatorServers.clear();
}
public int getTotalPartitionCount() {
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorEventProcessor.java b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorEventProcessor.java
index 4f53ba666..423d27622 100644
--- a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorEventProcessor.java
+++ b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorEventProcessor.java
@@ -108,6 +108,7 @@ import org.apache.fluss.server.metadata.CoordinatorMetadataCache;
import org.apache.fluss.server.metadata.ServerInfo;
import org.apache.fluss.server.metrics.group.CoordinatorMetricGroup;
import org.apache.fluss.server.utils.ServerRpcMessageUtils;
+import org.apache.fluss.server.zk.ZkEpoch;
import org.apache.fluss.server.zk.ZooKeeperClient;
import org.apache.fluss.server.zk.data.BucketAssignment;
import org.apache.fluss.server.zk.data.LeaderAndIsr;
@@ -188,6 +189,7 @@ public class CoordinatorEventProcessor implements EventProcessor {
public CoordinatorEventProcessor(
ZooKeeperClient zooKeeperClient,
+ ZkEpoch zkEpoch,
CoordinatorMetadataCache serverMetadataCache,
CoordinatorChannelManager coordinatorChannelManager,
CoordinatorContext coordinatorContext,
@@ -253,6 +255,9 @@ public class CoordinatorEventProcessor implements EventProcessor {
this.ioExecutor = ioExecutor;
this.lakeTableHelper =
new LakeTableHelper(zooKeeperClient, conf.getString(ConfigOptions.REMOTE_DATA_DIR));
+
+ this.coordinatorContext.setCoordinatorEpochAndZkVersion(
+ zkEpoch.getCoordinatorEpoch(), zkEpoch.getCoordinatorEpochZkVersion());
}
public CoordinatorEventManager getCoordinatorEventManager() {
@@ -1618,7 +1623,10 @@ public class CoordinatorEventProcessor implements EventProcessor {
tableAssignment.forEach(
(bucket, replicas) ->
newTableAssignment.put(bucket, new BucketAssignment(replicas)));
- zooKeeperClient.updateTableAssignment(tableId, new TableAssignment(newTableAssignment));
+ zooKeeperClient.updateTableAssignment(
+ tableId,
+ new TableAssignment(newTableAssignment),
+ coordinatorContext.getCoordinatorEpochZkVersion());
} else {
Map<Integer, List<Integer>> partitionAssignment =
coordinatorContext.getPartitionAssignment(
@@ -1629,7 +1637,9 @@ public class CoordinatorEventProcessor implements EventProcessor {
(bucket, replicas) ->
newPartitionAssignment.put(bucket, new BucketAssignment(replicas)));
zooKeeperClient.updatePartitionAssignment(
- partitionId, new PartitionAssignment(tableId, newPartitionAssignment));
+ partitionId,
+ new PartitionAssignment(tableId, newPartitionAssignment),
+ coordinatorContext.getCoordinatorEpochZkVersion());
}
}
@@ -1675,7 +1685,8 @@ public class CoordinatorEventProcessor implements EventProcessor {
}
try {
- zooKeeperClient.batchUpdateLeaderAndIsr(newLeaderAndIsrList);
+ zooKeeperClient.batchUpdateLeaderAndIsr(
+ newLeaderAndIsrList, coordinatorContext.getCoordinatorEpochZkVersion());
newLeaderAndIsrList.forEach(
(tableBucket, newLeaderAndIsr) ->
result.add(new AdjustIsrResultForBucket(tableBucket, newLeaderAndIsr)));
@@ -1686,7 +1697,10 @@ public class CoordinatorEventProcessor implements EventProcessor {
TableBucket tableBucket = entry.getKey();
LeaderAndIsr newLeaderAndIsr = entry.getValue();
try {
- zooKeeperClient.updateLeaderAndIsr(tableBucket, newLeaderAndIsr);
+ zooKeeperClient.updateLeaderAndIsr(
+ tableBucket,
+ newLeaderAndIsr,
+ coordinatorContext.getCoordinatorEpochZkVersion());
} catch (Exception e) {
LOG.error("Error when register leader and isr.", e);
result.add(
@@ -2212,7 +2226,8 @@ public class CoordinatorEventProcessor implements EventProcessor {
LeaderAndIsr newLeaderAndIsr = leaderAndIsr.newLeaderAndIsr(leaderAndIsr.isr());
coordinatorContext.putBucketLeaderAndIsr(tableBucket, newLeaderAndIsr);
- zooKeeperClient.updateLeaderAndIsr(tableBucket, newLeaderAndIsr);
+ zooKeeperClient.updateLeaderAndIsr(
+ tableBucket, newLeaderAndIsr, coordinatorContext.getCoordinatorEpochZkVersion());
coordinatorRequestBatch.newBatch();
coordinatorRequestBatch.addNotifyLeaderRequestForTabletServers(
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorLeaderElection.java b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorLeaderElection.java
index d226123b5..d67f77fdf 100644
--- a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorLeaderElection.java
+++ b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorLeaderElection.java
@@ -18,6 +18,7 @@
package org.apache.fluss.server.coordinator;
+import org.apache.fluss.exception.CoordinatorEpochFencedException;
import org.apache.fluss.server.zk.ZooKeeperClient;
import org.apache.fluss.server.zk.data.ZkData;
import org.apache.fluss.shaded.curator5.org.apache.curator.framework.recipes.leader.LeaderLatch;
@@ -51,6 +52,8 @@ public class CoordinatorLeaderElection implements AutoCloseable {
private static final Logger LOG = LoggerFactory.getLogger(CoordinatorLeaderElection.class);
private final String serverId;
+ private final ZooKeeperClient zkClient;
+ private final CoordinatorContext coordinatorContext;
private final LeaderLatch leaderLatch;
private final AtomicBoolean isLeader = new AtomicBoolean(false);
// Cached thread pool to run leader init/cleanup callbacks outside Curator's EventThread.
@@ -63,8 +66,11 @@ public class CoordinatorLeaderElection implements AutoCloseable {
private final AtomicReference<CompletableFuture<Void>> pendingCleanup =
new AtomicReference<>(CompletableFuture.completedFuture(null));
- public CoordinatorLeaderElection(ZooKeeperClient zkClient, String serverId) {
+ public CoordinatorLeaderElection(
+ ZooKeeperClient zkClient, String serverId, CoordinatorContext coordinatorContext) {
this.serverId = serverId;
+ this.zkClient = zkClient;
+ this.coordinatorContext = coordinatorContext;
this.leaderLatch =
new LeaderLatch(
zkClient.getCuratorClient(),
@@ -106,6 +112,11 @@ public class CoordinatorLeaderElection implements AutoCloseable {
CompletableFuture<Void> cleanup = pendingCleanup.get();
// Run init on a separate thread to avoid deadlock with
// Curator's EventThread when performing ZK operations.
+
+ // Set leader flag before init completes, so when zk found this leader, the
+ // leader can accept requests
+ isLeader.set(true);
+
leaderCallbackExecutor.execute(
() -> {
// Wait for any pending cleanup to finish first.
@@ -123,16 +134,19 @@ public class CoordinatorLeaderElection implements AutoCloseable {
}
try {
initLeaderServices.run();
+ } catch (CoordinatorEpochFencedException e) {
+ LOG.warn(
+ "Coordinator server {} has been fenced and not become leader successfully.",
+ serverId);
+ notLeader();
} catch (Exception e) {
LOG.error(
"Failed to initialize leader services for server {}",
serverId,
e);
+ notLeader();
}
});
- // Set leader flag before init completes, so when zk found this leader, the
- // leader can accept requests
- isLeader.set(true);
}
@Override
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorRequestBatch.java b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorRequestBatch.java
index 60eaa988c..344105367 100644
--- a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorRequestBatch.java
+++ b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorRequestBatch.java
@@ -683,6 +683,7 @@ public class CoordinatorRequestBatch {
// tablet servers.
return makeUpdateMetadataRequest(
coordinatorContext.getCoordinatorServerInfo(),
+ coordinatorContext.getCoordinatorEpoch(),
new HashSet<>(coordinatorContext.getLiveTabletServers().values()),
tableMetadataList,
partitionMetadataList);
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 25a0d866f..a54fc2d58 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
@@ -39,6 +39,7 @@ import org.apache.fluss.server.metadata.ServerMetadataCache;
import org.apache.fluss.server.metrics.ServerMetricUtils;
import org.apache.fluss.server.metrics.group.CoordinatorMetricGroup;
import org.apache.fluss.server.metrics.group.LakeTieringMetricGroup;
+import org.apache.fluss.server.zk.ZkEpoch;
import org.apache.fluss.server.zk.ZooKeeperClient;
import org.apache.fluss.server.zk.ZooKeeperUtils;
import org.apache.fluss.server.zk.data.CoordinatorAddress;
@@ -220,9 +221,11 @@ public class CoordinatorServer extends ServerBase {
serverId);
this.zkClient = ZooKeeperUtils.startZookeeperClient(conf, this);
+ this.coordinatorContext = new CoordinatorContext();
// CoordinatorLeaderElection must be created after zkClient is initialized.
- this.coordinatorLeaderElection = new CoordinatorLeaderElection(zkClient, serverId);
+ this.coordinatorLeaderElection =
+ new CoordinatorLeaderElection(zkClient, serverId, coordinatorContext);
this.lakeCatalogDynamicLoader = new LakeCatalogDynamicLoader(conf, pluginManager, true);
this.dynamicConfigManager = new DynamicConfigManager(zkClient, conf, true);
@@ -232,7 +235,6 @@ public class CoordinatorServer extends ServerBase {
dynamicConfigManager.startup();
- this.coordinatorContext = new CoordinatorContext();
this.metadataCache = new CoordinatorMetadataCache();
this.authorizer = AuthorizerLoader.createAuthorizer(conf, zkClient, pluginManager);
@@ -294,7 +296,10 @@ public class CoordinatorServer extends ServerBase {
}
protected void initCoordinatorLeader() throws Exception {
+ // to avoid split-brain
+ ZkEpoch zkEpoch = zkClient.fenceBecomeCoordinatorLeader(serverId);
+ registerCoordinatorLeader();
synchronized (lock) {
this.clientMetricGroup = new ClientMetricGroup(metricRegistry, SERVER_NAME);
this.rpcClient = RpcClient.create(conf, clientMetricGroup);
@@ -305,7 +310,6 @@ public class CoordinatorServer extends ServerBase {
new AutoPartitionManager(metadataCache, metadataManager, conf);
autoPartitionManager.start();
- registerCoordinatorLeader();
// start coordinator event processor after we register coordinator leader to zk
// so that the event processor can get the coordinator leader node from zk during start
// up.
@@ -314,6 +318,7 @@ public class CoordinatorServer extends ServerBase {
this.coordinatorEventProcessor =
new CoordinatorEventProcessor(
zkClient,
+ zkEpoch,
metadataCache,
coordinatorChannelManager,
coordinatorContext,
@@ -337,10 +342,20 @@ public class CoordinatorServer extends ServerBase {
* from leader to standby. It cleans up leader-only resources while keeping the server running
* as a standby, ready to participate in future elections.
*/
- protected void cleanupCoordinatorLeader() throws Exception {
+ protected void cleanupCoordinatorLeader() {
synchronized (lock) {
LOG.info("Cleaning up coordinator leader services.");
+ try {
+ // make sure the current coordinator leader node is unregistered.
+ // Different from ZK disconnection,
+ // when we actively release the Leader's election,
+ // we need to manually delete the node
+ unregisterCoordinatorLeader();
+ } catch (Throwable t) {
+ LOG.warn("Failed to unregister coordinator leader from Zookeeper", t);
+ }
+
// Clean up leader-specific resources in reverse order of initialization
try {
if (coordinatorEventProcessor != null) {
@@ -427,6 +442,11 @@ public class CoordinatorServer extends ServerBase {
"coordinator leader", () -> zkClient.registerCoordinatorLeader(coordinatorAddress));
}
+ private void unregisterCoordinatorLeader() throws Exception {
+ CoordinatorAddress coordinatorAddress = buildCoordinatorAddress();
+ zkClient.unregisterCoordinatorLeader(coordinatorAddress);
+ }
+
private CoordinatorAddress buildCoordinatorAddress() {
List<Endpoint> bindEndpoints = rpcServer.getBindEndpoints();
return new CoordinatorAddress(
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/statemachine/ReplicaStateMachine.java b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/statemachine/ReplicaStateMachine.java
index 091c2cbc9..2416b163b 100644
--- a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/statemachine/ReplicaStateMachine.java
+++ b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/statemachine/ReplicaStateMachine.java
@@ -487,7 +487,8 @@ public class ReplicaStateMachine {
toUpdateLeaderAndIsrList.put(tableBucket, adjustLeaderAndIsr);
}
try {
- zooKeeperClient.batchUpdateLeaderAndIsr(toUpdateLeaderAndIsrList);
+ zooKeeperClient.batchUpdateLeaderAndIsr(
+ toUpdateLeaderAndIsrList, coordinatorContext.getCoordinatorEpochZkVersion());
toUpdateLeaderAndIsrList.forEach(coordinatorContext::putBucketLeaderAndIsr);
return adjustedLeaderAndIsr;
} catch (Exception e) {
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/statemachine/TableBucketStateMachine.java b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/statemachine/TableBucketStateMachine.java
index 85dcc434f..b01e65c92 100644
--- a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/statemachine/TableBucketStateMachine.java
+++ b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/statemachine/TableBucketStateMachine.java
@@ -311,7 +311,10 @@ public class TableBucketStateMachine {
ElectionResult electionResult = optionalElectionResult.get();
LeaderAndIsr leaderAndIsr = electionResult.leaderAndIsr;
try {
- zooKeeperClient.registerLeaderAndIsr(tableBucket, leaderAndIsr);
+ zooKeeperClient.registerLeaderAndIsr(
+ tableBucket,
+ leaderAndIsr,
+ coordinatorContext.getCoordinatorEpochZkVersion());
} catch (Exception e) {
LOG.error(
"Fail to create state node for table bucket {} in zookeeper.",
@@ -379,7 +382,8 @@ public class TableBucketStateMachine {
if (!tableBucketLeadAndIsrInfos.isEmpty()) {
try {
zooKeeperClient.batchRegisterLeaderAndIsrForTablePartition(
- tableBucketLeadAndIsrInfos);
+ tableBucketLeadAndIsrInfos,
+ coordinatorContext.getCoordinatorEpochZkVersion());
registerSuccessList.addAll(tableBucketLeadAndIsrInfos);
} catch (Exception e) {
LOG.error(
@@ -454,7 +458,10 @@ public class TableBucketStateMachine {
List<RegisterTableBucketLeadAndIsrInfo> registerSuccessList = new ArrayList<>();
for (RegisterTableBucketLeadAndIsrInfo info : registerList) {
try {
- zooKeeperClient.registerLeaderAndIsr(info.getTableBucket(), info.getLeaderAndIsr());
+ zooKeeperClient.registerLeaderAndIsr(
+ info.getTableBucket(),
+ info.getLeaderAndIsr(),
+ coordinatorContext.getCoordinatorEpochZkVersion());
registerSuccessList.add(info);
} catch (Exception e) {
LOG.error(
@@ -496,7 +503,10 @@ public class TableBucketStateMachine {
}
ElectionResult electionResult = optionalElectionResult.get();
try {
- zooKeeperClient.updateLeaderAndIsr(tableBucket, electionResult.leaderAndIsr);
+ zooKeeperClient.updateLeaderAndIsr(
+ tableBucket,
+ electionResult.leaderAndIsr,
+ coordinatorContext.getCoordinatorEpochZkVersion());
} catch (Exception e) {
LOG.error(
"Fail to update bucket LeaderAndIsr for table bucket {}.",
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/metrics/group/CoordinatorMetricGroup.java b/fluss-server/src/main/java/org/apache/fluss/server/metrics/group/CoordinatorMetricGroup.java
index 72ff328a7..5ae099923 100644
--- a/fluss-server/src/main/java/org/apache/fluss/server/metrics/group/CoordinatorMetricGroup.java
+++ b/fluss-server/src/main/java/org/apache/fluss/server/metrics/group/CoordinatorMetricGroup.java
@@ -71,7 +71,7 @@ public class CoordinatorMetricGroup extends AbstractMetricGroup {
protected final void putVariables(Map<String, String> variables) {
variables.put("cluster_id", clusterId);
variables.put("host", hostname);
- variables.put("server_id", serverId);
+ variables.put("server_id", String.valueOf(serverId));
}
public CoordinatorEventMetricGroup getOrAddEventTypeMetricGroup(
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/replica/ReplicaManager.java b/fluss-server/src/main/java/org/apache/fluss/server/replica/ReplicaManager.java
index ee3237032..6960b5191 100644
--- a/fluss-server/src/main/java/org/apache/fluss/server/replica/ReplicaManager.java
+++ b/fluss-server/src/main/java/org/apache/fluss/server/replica/ReplicaManager.java
@@ -1991,6 +1991,11 @@ public class ReplicaManager implements ServerReconfigurable {
return serverMetricGroup;
}
+ @VisibleForTesting
+ public void resetCoordinatorEpoch() {
+ this.coordinatorEpoch = CoordinatorContext.INITIAL_COORDINATOR_EPOCH;
+ }
+
/**
* Interface to represent the state of hosted {@link Replica}. We create a concrete (active)
* {@link Replica} instance when the TabletServer receives a createLogLeader request or
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/utils/ServerRpcMessageUtils.java b/fluss-server/src/main/java/org/apache/fluss/server/utils/ServerRpcMessageUtils.java
index 02d02c29e..d37c36148 100644
--- a/fluss-server/src/main/java/org/apache/fluss/server/utils/ServerRpcMessageUtils.java
+++ b/fluss-server/src/main/java/org/apache/fluss/server/utils/ServerRpcMessageUtils.java
@@ -461,6 +461,7 @@ public class ServerRpcMessageUtils {
public static UpdateMetadataRequest makeUpdateMetadataRequest(
@Nullable ServerInfo coordinatorServer,
+ @Nullable Integer coordinatorEpoch,
Set<ServerInfo> aliveTableServers,
List<TableMetadata> tableMetadataList,
List<PartitionMetadata> partitionMetadataList) {
@@ -503,6 +504,9 @@ public class ServerRpcMessageUtils {
updateMetadataRequest.addAllTableMetadatas(pbTableMetadataList);
updateMetadataRequest.addAllPartitionMetadatas(pbPartitionMetadataList);
+ if (coordinatorEpoch != null) {
+ updateMetadataRequest.setCoordinatorEpoch(coordinatorEpoch);
+ }
return updateMetadataRequest;
}
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/zk/ZkEpoch.java b/fluss-server/src/main/java/org/apache/fluss/server/zk/ZkEpoch.java
new file mode 100644
index 000000000..f3d401a25
--- /dev/null
+++ b/fluss-server/src/main/java/org/apache/fluss/server/zk/ZkEpoch.java
@@ -0,0 +1,42 @@
+/*
+ * 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.fluss.server.zk;
+
+/** Class for coordinator epoch and coordinator epoch zk version. */
+public class ZkEpoch {
+ private final int coordinatorEpoch;
+ private final int coordinatorEpochZkVersion;
+
+ public ZkEpoch(int coordinatorEpoch, int coordinatorEpochZkVersion) {
+ this.coordinatorEpoch = coordinatorEpoch;
+ this.coordinatorEpochZkVersion = coordinatorEpochZkVersion;
+ }
+
+ public ZkEpoch nextZkEpoch() {
+ return new ZkEpoch(coordinatorEpoch + 1, coordinatorEpochZkVersion + 1);
+ }
+
+ public int getCoordinatorEpoch() {
+ return coordinatorEpoch;
+ }
+
+ public int getCoordinatorEpochZkVersion() {
+ return coordinatorEpochZkVersion;
+ }
+}
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/zk/ZooKeeperClient.java b/fluss-server/src/main/java/org/apache/fluss/server/zk/ZooKeeperClient.java
index 78667ca48..3c8b58fa2 100644
--- a/fluss-server/src/main/java/org/apache/fluss/server/zk/ZooKeeperClient.java
+++ b/fluss-server/src/main/java/org/apache/fluss/server/zk/ZooKeeperClient.java
@@ -22,6 +22,7 @@ import org.apache.fluss.annotation.VisibleForTesting;
import org.apache.fluss.config.ConfigOptions;
import org.apache.fluss.config.Configuration;
import org.apache.fluss.config.FlussConfigUtils;
+import org.apache.fluss.exception.CoordinatorEpochFencedException;
import org.apache.fluss.metadata.DatabaseSummary;
import org.apache.fluss.metadata.PhysicalTablePath;
import org.apache.fluss.metadata.ResolvedPartitionSpec;
@@ -34,6 +35,7 @@ import org.apache.fluss.security.acl.AccessControlEntry;
import org.apache.fluss.security.acl.Resource;
import org.apache.fluss.security.acl.ResourceType;
import org.apache.fluss.server.authorizer.DefaultAuthorizer.VersionedAcls;
+import org.apache.fluss.server.coordinator.CoordinatorContext;
import org.apache.fluss.server.entity.RegisterTableBucketLeadAndIsrInfo;
import org.apache.fluss.server.metadata.BucketMetadata;
import org.apache.fluss.server.zk.ZkAsyncRequest.ZkCheckExistsRequest;
@@ -86,6 +88,7 @@ import org.apache.fluss.server.zk.data.ZkData.TableSequenceIdZNode;
import org.apache.fluss.server.zk.data.ZkData.TableZNode;
import org.apache.fluss.server.zk.data.ZkData.TablesZNode;
import org.apache.fluss.server.zk.data.ZkData.WriterIdZNode;
+import org.apache.fluss.server.zk.data.ZkVersion;
import org.apache.fluss.server.zk.data.lake.LakeTable;
import org.apache.fluss.server.zk.data.lake.LakeTableSnapshot;
import org.apache.fluss.server.zk.data.lease.KvSnapshotLeaseMetadata;
@@ -127,11 +130,33 @@ import java.util.stream.Collectors;
import static java.util.stream.Collectors.toMap;
import static org.apache.fluss.metadata.ResolvedPartitionSpec.fromPartitionName;
+import static org.apache.fluss.server.zk.ZooKeeperOp.multiRequest;
import static org.apache.fluss.utils.Preconditions.checkNotNull;
/**
* This class includes methods for write/read various metadata (leader address, tablet server
* registration, table assignment, table, schema) in Zookeeper.
+ *
+ * <p>In some method, 'expectedZkVersion' is used to execute an epoch Zookeeper version check.
+ * Conditions requiring epoch checks (all must be met):
+ * ┌─────────────────────────────────────────────────────────┐ │ 1. Invoked by the Coordinator (not
+ * the TabletServer) │ │ 2. Operates on persistent nodes (not ephemeral) │ │ 3. Constitutes a
+ * "control plane" operation: │ │ partition assignment or LeaderAndIsr election │ │ 4. Concurrent
+ * access to the same path by old and new │ │ leaders during leader failover │ │ 5. No other
+ * mechanisms (optimistic locking, │ │ idempotency, or reloading) provide fallback │
+ * └─────────────────────────────────────────────────────────┘
+ *
+ * <p>In practice, only two types of operations truly require this: - CRUD for Table/Partition
+ * Assignment (assignment decisions). - CRUD for LeaderAndIsr (leader election results).
+ *
+ * <p>These operations are inevitably executed concurrently by the old and new coordinators during
+ * failover (as the new leader immediately reassigns partitions), and overwrites cannot be
+ * automatically recovered.
+ *
+ * <p>All other operations do not require epoch checks because: - DDL operations are protected
+ * against concurrency by client reconnection mechanisms. - TabletServer operations are unaffected
+ * by coordinator failovers. - ACLs and Configs have their own version control or idempotency
+ * guarantees. - Ephemeral nodes are managed via session lifecycle.
*/
@Internal
public class ZooKeeperClient implements AutoCloseable {
@@ -144,6 +169,7 @@ public class ZooKeeperClient implements AutoCloseable {
private final CuratorFrameworkWithUnhandledErrorListener curatorFrameworkWrapper;
private final CuratorFramework zkClient;
+ private final ZooKeeperOp zkOp;
private final ZkSequenceIDCounter tableIdCounter;
private final ZkSequenceIDCounter partitionIdCounter;
private final ZkSequenceIDCounter writerIdCounter;
@@ -158,6 +184,7 @@ public class ZooKeeperClient implements AutoCloseable {
Configuration configuration) {
this.curatorFrameworkWrapper = curatorFrameworkWrapper;
this.zkClient = curatorFrameworkWrapper.asCuratorFramework();
+ this.zkOp = new ZooKeeperOp(zkClient);
this.tableIdCounter = new ZkSequenceIDCounter(zkClient, TableSequenceIdZNode.path());
this.partitionIdCounter =
new ZkSequenceIDCounter(zkClient, PartitionSequenceIdZNode.path());
@@ -197,6 +224,49 @@ public class ZooKeeperClient implements AutoCloseable {
LOG.info("Registered Coordinator server {} at path {}.", coordinatorAddress, path);
}
+ /**
+ * Become coordinator leader. This method is a step after electCoordinatorLeader() and before
+ * registerCoordinatorLeader(). This is to ensure the coordinator get and update the coordinator
+ * epoch and coordinator epoch zk version.
+ */
+ public ZkEpoch fenceBecomeCoordinatorLeader(String coordinatorId) throws Exception {
+ ensureEpochZnodeExists();
+
+ try {
+ ZkEpoch getEpoch = getCurrentEpoch();
+ int currentCoordinatorEpoch = getEpoch.getCoordinatorEpoch();
+ int currentCoordinatorEpochZkVersion = getEpoch.getCoordinatorEpochZkVersion();
+ int newCoordinatorEpoch = currentCoordinatorEpoch + 1;
+ LOG.info(
+ "Coordinator leader {} tries to update epoch. Current epoch={}, Zookeeper version={}, new epoch={}",
+ coordinatorId,
+ currentCoordinatorEpoch,
+ currentCoordinatorEpochZkVersion,
+ newCoordinatorEpoch);
+
+ // atomically update epoch
+ Stat stat =
+ zkClient.setData()
+ .withVersion(currentCoordinatorEpochZkVersion)
+ .forPath(
+ ZkData.CoordinatorEpochZNode.path(),
+ ZkData.CoordinatorEpochZNode.encode(newCoordinatorEpoch));
+
+ LOG.info(
+ "Coordinator leader has updated epoch. Current epoch={}, Zookeeper version={}",
+ newCoordinatorEpoch,
+ stat.getVersion());
+
+ return new ZkEpoch(newCoordinatorEpoch, stat.getVersion());
+ } catch (KeeperException.BadVersionException e) {
+ // Other coordinator leader has updated epoch.
+ // If this happens, it means our fence is in effect.
+ LOG.info("Coordinator leader {} failed to update epoch.", coordinatorId);
+ throw new CoordinatorEpochFencedException(
+ "Coordinator leader election has been fenced.");
+ }
+ }
+
/** Register a coordinator leader to ZK. */
public void registerCoordinatorLeader(CoordinatorAddress coordinatorAddress) throws Exception {
String path = ZkData.CoordinatorLeaderZNode.path();
@@ -207,6 +277,23 @@ public class ZooKeeperClient implements AutoCloseable {
LOG.info("Registered Coordinator leader {} at path {}.", coordinatorAddress, path);
}
+ /** Manually unregister a coordinator leader from ZK. */
+ public void unregisterCoordinatorLeader(CoordinatorAddress coordinatorAddress)
+ throws Exception {
+ String path = ZkData.CoordinatorLeaderZNode.path();
+ Stat stat = new Stat();
+ byte[] bytes = zkClient.getData().storingStatIn(stat).forPath(path);
+ if (bytes == null || bytes.length == 0) {
+ // not exists, not need to unregister
+ return;
+ }
+ CoordinatorAddress storedAddress = ZkData.CoordinatorLeaderZNode.decode(bytes);
+ if (storedAddress.getId().equals(coordinatorAddress.getId())) {
+ zkClient.delete().withVersion(stat.getVersion()).forPath(path);
+ LOG.info("Unregistered Coordinator leader {} at path {}.", coordinatorAddress, path);
+ }
+ }
+
/** Get the leader address registered in ZK. */
public Optional<CoordinatorAddress> getCoordinatorLeaderAddress() throws Exception {
Optional<byte[]> bytes = getOrEmpty(ZkData.CoordinatorLeaderZNode.path());
@@ -221,6 +308,35 @@ public class ZooKeeperClient implements AutoCloseable {
return getChildren(ZkData.CoordinatorIdsZNode.path());
}
+ /** Ensure epoch znode exists. */
+ public void ensureEpochZnodeExists() throws Exception {
+ String path = ZkData.CoordinatorEpochZNode.path();
+ if (zkClient.checkExists().forPath(path) == null) {
+ try {
+ zkClient.create()
+ .creatingParentsIfNeeded()
+ .withMode(CreateMode.PERSISTENT)
+ .forPath(
+ path,
+ ZkData.CoordinatorEpochZNode.encode(
+ CoordinatorContext.INITIAL_COORDINATOR_EPOCH - 1));
+ } catch (KeeperException.NodeExistsException e) {
+ // can be ignored when two coordinator almost simultaneously create the epoch znode
+ }
+ }
+ }
+
+ /** Get epoch now in ZK. */
+ public ZkEpoch getCurrentEpoch() throws Exception {
+ Stat currentStat = new Stat();
+ byte[] bytes =
+ zkClient.getData()
+ .storingStatIn(currentStat)
+ .forPath(ZkData.CoordinatorEpochZNode.path());
+ int currentEpoch = ZkData.CoordinatorEpochZNode.decode(bytes);
+ int currentVersion = currentStat.getVersion();
+ return new ZkEpoch(currentEpoch, currentVersion);
+ }
// --------------------------------------------------------------------------------------------
// Tablet server
// --------------------------------------------------------------------------------------------
@@ -329,17 +445,26 @@ public class ZooKeeperClient implements AutoCloseable {
"partition assignment");
}
- public void updateTableAssignment(long tableId, TableAssignment tableAssignment)
- throws Exception {
+ public void updateTableAssignment(
+ long tableId, TableAssignment tableAssignment, int expectedZkVersion) throws Exception {
String path = TableIdZNode.path(tableId);
- zkClient.setData().forPath(path, TableIdZNode.encode(tableAssignment));
+ byte[] data = TableIdZNode.encode(tableAssignment);
+ CuratorOp updateOp = zkOp.updateOp(path, data);
+ List<CuratorOp> ops = wrapRequestWithEpochCheck(updateOp, expectedZkVersion);
+
+ zkClient.transaction().forOperations(ops);
LOG.debug("Updated table assignment {} for table id {}.", tableAssignment, tableId);
}
- public void updatePartitionAssignment(long partitionId, PartitionAssignment partitionAssignment)
+ public void updatePartitionAssignment(
+ long partitionId, PartitionAssignment partitionAssignment, int expectedZkVersion)
throws Exception {
String path = PartitionIdZNode.path(partitionId);
- zkClient.setData().forPath(path, PartitionIdZNode.encode(partitionAssignment));
+ byte[] data = PartitionIdZNode.encode(partitionAssignment);
+ CuratorOp updateOp = zkOp.updateOp(path, data);
+ List<CuratorOp> ops = wrapRequestWithEpochCheck(updateOp, expectedZkVersion);
+
+ zkClient.transaction().forOperations(ops);
LOG.debug(
"Updated partition assignment {} for partition id {}.",
partitionAssignment,
@@ -363,18 +488,20 @@ public class ZooKeeperClient implements AutoCloseable {
// --------------------------------------------------------------------------------------------
/** Register bucket LeaderAndIsr to ZK. */
- public void registerLeaderAndIsr(TableBucket tableBucket, LeaderAndIsr leaderAndIsr)
+ public void registerLeaderAndIsr(
+ TableBucket tableBucket, LeaderAndIsr leaderAndIsr, int expectedZkVersion)
throws Exception {
+
String path = LeaderAndIsrZNode.path(tableBucket);
- zkClient.create()
- .creatingParentsIfNeeded()
- .withMode(CreateMode.PERSISTENT)
- .forPath(path, LeaderAndIsrZNode.encode(leaderAndIsr));
+ byte[] data = LeaderAndIsrZNode.encode(leaderAndIsr);
+
+ createRecursiveWithEpochCheck(path, data, expectedZkVersion, false);
LOG.info("Registered {} for bucket {} in Zookeeper.", leaderAndIsr, tableBucket);
}
public void batchRegisterLeaderAndIsrForTablePartition(
- List<RegisterTableBucketLeadAndIsrInfo> registerList) throws Exception {
+ List<RegisterTableBucketLeadAndIsrInfo> registerList, int expectedZkVersion)
+ throws Exception {
if (registerList.isEmpty()) {
return;
}
@@ -410,12 +537,14 @@ public class ZooKeeperClient implements AutoCloseable {
ops.add(parentNodeCreate);
ops.add(currentNodeCreate);
if (ops.size() == MAX_BATCH_SIZE) {
- zkClient.transaction().forOperations(ops);
+ List<CuratorOp> wrapOps = wrapRequestsWithEpochCheck(ops, expectedZkVersion);
+ zkClient.transaction().forOperations(wrapOps);
ops.clear();
}
}
if (!ops.isEmpty()) {
- zkClient.transaction().forOperations(ops);
+ List<CuratorOp> wrapOps = wrapRequestsWithEpochCheck(ops, expectedZkVersion);
+ zkClient.transaction().forOperations(wrapOps);
}
LOG.info(
"Batch registered leadAndIsr for tableId: {}, partitionId: {}, partitionName: {} in Zookeeper.",
@@ -447,14 +576,21 @@ public class ZooKeeperClient implements AutoCloseable {
"leader and isr");
}
- public void updateLeaderAndIsr(TableBucket tableBucket, LeaderAndIsr leaderAndIsr)
+ public void updateLeaderAndIsr(
+ TableBucket tableBucket, LeaderAndIsr leaderAndIsr, int expectedZkVersion)
throws Exception {
String path = LeaderAndIsrZNode.path(tableBucket);
- zkClient.setData().forPath(path, LeaderAndIsrZNode.encode(leaderAndIsr));
+ byte[] data = LeaderAndIsrZNode.encode(leaderAndIsr);
+
+ CuratorOp updateOp = zkOp.updateOp(path, data);
+ List<CuratorOp> ops = wrapRequestWithEpochCheck(updateOp, expectedZkVersion);
+
+ zkClient.transaction().forOperations(ops);
LOG.info("Updated {} for bucket {} in Zookeeper.", leaderAndIsr, tableBucket);
}
- public void batchUpdateLeaderAndIsr(Map<TableBucket, LeaderAndIsr> leaderAndIsrList)
+ public void batchUpdateLeaderAndIsr(
+ Map<TableBucket, LeaderAndIsr> leaderAndIsrList, int expectedZkVersion)
throws Exception {
if (leaderAndIsrList.isEmpty()) {
return;
@@ -471,16 +607,18 @@ public class ZooKeeperClient implements AutoCloseable {
CuratorOp updateOp = zkClient.transactionOp().setData().forPath(path, data);
ops.add(updateOp);
if (ops.size() == MAX_BATCH_SIZE) {
- zkClient.transaction().forOperations(ops);
+ List<CuratorOp> wrapOps = wrapRequestsWithEpochCheck(ops, expectedZkVersion);
+ zkClient.transaction().forOperations(wrapOps);
ops.clear();
}
}
if (!ops.isEmpty()) {
- zkClient.transaction().forOperations(ops);
+ List<CuratorOp> wrapOps = wrapRequestsWithEpochCheck(ops, expectedZkVersion);
+ zkClient.transaction().forOperations(wrapOps);
}
}
- public void deleteLeaderAndIsr(TableBucket tableBucket) throws Exception {
+ protected void deleteLeaderAndIsr(TableBucket tableBucket) throws Exception {
String path = LeaderAndIsrZNode.path(tableBucket);
zkClient.delete().forPath(path);
LOG.info("Deleted LeaderAndIsr for bucket {} in Zookeeper.", tableBucket);
@@ -1781,6 +1919,79 @@ public class ZooKeeperClient implements AutoCloseable {
return result;
}
+ /**
+ * create a node (recursively if parent path not exists) with Zk epoch version check.
+ *
+ * @param path the path to create
+ * @param data the data to write
+ * @param throwIfPathExists whether to throw exception if path exist
+ * @throws Exception if any error occurs
+ */
+ public void createRecursiveWithEpochCheck(
+ String path, byte[] data, int expectedZkVersion, boolean throwIfPathExists)
+ throws Exception {
+ CuratorOp createOp = zkOp.createOp(path, data, CreateMode.PERSISTENT);
+ List<CuratorOp> ops = wrapRequestWithEpochCheck(createOp, expectedZkVersion);
+
+ try {
+ // try to directly create
+ zkClient.transaction().forOperations(ops);
+ } catch (KeeperException.NodeExistsException e) {
+ // should not exist
+ if (throwIfPathExists) {
+ throw e;
+ }
+ } catch (KeeperException.NoNodeException e) {
+ // if parent does not exist, create parent first
+ int indexOfLastSlash = path.lastIndexOf("/");
+ if (indexOfLastSlash == -1) {
+ throw new IllegalArgumentException("Invalid path: " + path);
+ } else if (indexOfLastSlash == 0) {
+ // root path can be directly create without fence
+ try {
+ zkClient.create()
+ .creatingParentsIfNeeded()
+ .withMode(CreateMode.PERSISTENT)
+ .forPath(path);
+ } catch (KeeperException.NodeExistsException ignored) {
+ // ignore
+ }
+ } else {
+ // indexOfLastSlash > 0
+ String parentPath = path.substring(0, indexOfLastSlash);
+ createRecursiveWithEpochCheck(
+ parentPath, null, expectedZkVersion, throwIfPathExists);
+ // After creating parent (or if parent is root), retry creating the original path
+ zkClient.transaction().forOperations(ops);
+ }
+ } catch (KeeperException.BadVersionException e) {
+ LOG.error("Bad version for path {}, expected version {} ", path, expectedZkVersion);
+ throw e;
+ }
+ }
+
+ public List<CuratorOp> wrapRequestWithEpochCheck(CuratorOp request, int expectedZkVersion)
+ throws Exception {
+ return wrapRequestsWithEpochCheck(Collections.singletonList(request), expectedZkVersion);
+ }
+
+ public List<CuratorOp> wrapRequestsWithEpochCheck(
+ List<CuratorOp> requestList, int expectedZkVersion) throws Exception {
+ if (ZkVersion.MATCH_ANY_VERSION.getVersion() == expectedZkVersion) {
+ return requestList;
+ } else if (expectedZkVersion >= 0) {
+ CuratorOp checkOp =
+ zkOp.checkOp(ZkData.CoordinatorEpochZNode.path(), expectedZkVersion);
+ return multiRequest(checkOp, requestList);
+ } else {
+ throw new IllegalArgumentException(
+ "Expected coordinator epoch zkVersion "
+ + expectedZkVersion
+ + " should be non-negative or equal to "
+ + ZkVersion.MATCH_ANY_VERSION.getVersion());
+ }
+ }
+
// --------------------------------------------------------------------------------------------
// Producer Offset Snapshot
// --------------------------------------------------------------------------------------------
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/zk/ZooKeeperOp.java b/fluss-server/src/main/java/org/apache/fluss/server/zk/ZooKeeperOp.java
new file mode 100644
index 000000000..e44670d64
--- /dev/null
+++ b/fluss-server/src/main/java/org/apache/fluss/server/zk/ZooKeeperOp.java
@@ -0,0 +1,64 @@
+/*
+ * 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.fluss.server.zk;
+
+import org.apache.fluss.shaded.curator5.org.apache.curator.framework.CuratorFramework;
+import org.apache.fluss.shaded.curator5.org.apache.curator.framework.api.transaction.CuratorOp;
+import org.apache.fluss.shaded.zookeeper3.org.apache.zookeeper.CreateMode;
+
+import java.util.ArrayList;
+import java.util.List;
+
+/** This class contains some utility methods for wrap/unwrap operations for Zookeeper. */
+public class ZooKeeperOp {
+ private final CuratorFramework zkClient;
+
+ public ZooKeeperOp(CuratorFramework zkClient) {
+ this.zkClient = zkClient;
+ }
+
+ public CuratorOp checkOp(String path, int expectZkVersion) throws Exception {
+ return zkClient.transactionOp().check().withVersion(expectZkVersion).forPath(path);
+ }
+
+ public CuratorOp createOp(String path, byte[] data, CreateMode createMode) throws Exception {
+ return zkClient.transactionOp().create().withMode(createMode).forPath(path, data);
+ }
+
+ public CuratorOp updateOp(String path, byte[] data) throws Exception {
+ return zkClient.transactionOp().setData().forPath(path, data);
+ }
+
+ public CuratorOp deleteOp(String path) throws Exception {
+ return zkClient.transactionOp().delete().forPath(path);
+ }
+
+ public static List<CuratorOp> multiRequest(CuratorOp op1, CuratorOp op2) {
+ List<CuratorOp> ops = new ArrayList<>();
+ ops.add(op1);
+ ops.add(op2);
+ return ops;
+ }
+
+ public static List<CuratorOp> multiRequest(CuratorOp op, List<CuratorOp> ops) {
+ List<CuratorOp> result = new ArrayList<>();
+ result.add(op);
+ result.addAll(ops);
+ return result;
+ }
+}
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/zk/data/ZkData.java b/fluss-server/src/main/java/org/apache/fluss/server/zk/data/ZkData.java
index 32b5a7c01..941512e0f 100644
--- a/fluss-server/src/main/java/org/apache/fluss/server/zk/data/ZkData.java
+++ b/fluss-server/src/main/java/org/apache/fluss/server/zk/data/ZkData.java
@@ -366,6 +366,24 @@ public final class ZkData {
}
}
+ /**
+ * The znode for the coordinator epoch. The znode path is:
+ *
+ * <p>/coordinators/epoch
+ */
+ public static final class CoordinatorEpochZNode {
+ public static String path() {
+ return "/coordinators/epoch";
+ }
+
+ public static byte[] encode(int epoch) {
+ return String.valueOf(epoch).getBytes(StandardCharsets.UTF_8);
+ }
+
+ public static int decode(byte[] bytes) {
+ return Integer.parseInt(new String(bytes, StandardCharsets.UTF_8));
+ }
+ }
// ------------------------------------------------------------------------------------------
// ZNodes under "/tabletservers/"
// ------------------------------------------------------------------------------------------
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/zk/data/ZkVersion.java b/fluss-server/src/main/java/org/apache/fluss/server/zk/data/ZkVersion.java
new file mode 100644
index 000000000..26242df0f
--- /dev/null
+++ b/fluss-server/src/main/java/org/apache/fluss/server/zk/data/ZkVersion.java
@@ -0,0 +1,35 @@
+/*
+ * 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.fluss.server.zk.data;
+
+/** Enum to represent the type of special Zookeeper version. */
+public enum ZkVersion {
+ MATCH_ANY_VERSION(-1),
+ UNKNOWN_VERSION(-2);
+
+ private final int version;
+
+ ZkVersion(int version) {
+ this.version = version;
+ }
+
+ public int getVersion() {
+ return version;
+ }
+}
diff --git a/fluss-server/src/test/java/org/apache/fluss/server/coordinator/CoordinatorChannelManagerTest.java b/fluss-server/src/test/java/org/apache/fluss/server/coordinator/CoordinatorChannelManagerTest.java
index 9d8eda15b..31c07d75c 100644
--- a/fluss-server/src/test/java/org/apache/fluss/server/coordinator/CoordinatorChannelManagerTest.java
+++ b/fluss-server/src/test/java/org/apache/fluss/server/coordinator/CoordinatorChannelManagerTest.java
@@ -92,6 +92,7 @@ class CoordinatorChannelManagerTest {
// we use update metadata request to test for simplicity
UpdateMetadataRequest updateMetadataRequest =
makeUpdateMetadataRequest(
+ null,
null,
Collections.emptySet(),
Collections.emptyList(),
diff --git a/fluss-server/src/test/java/org/apache/fluss/server/coordinator/CoordinatorEventProcessorTest.java b/fluss-server/src/test/java/org/apache/fluss/server/coordinator/CoordinatorEventProcessorTest.java
index 997a8738f..cd036b364 100644
--- a/fluss-server/src/test/java/org/apache/fluss/server/coordinator/CoordinatorEventProcessorTest.java
+++ b/fluss-server/src/test/java/org/apache/fluss/server/coordinator/CoordinatorEventProcessorTest.java
@@ -65,6 +65,7 @@ import org.apache.fluss.server.metadata.TableMetadata;
import org.apache.fluss.server.metrics.group.TestingMetricGroups;
import org.apache.fluss.server.tablet.TestTabletServerGateway;
import org.apache.fluss.server.zk.NOPErrorHandler;
+import org.apache.fluss.server.zk.ZkEpoch;
import org.apache.fluss.server.zk.ZooKeeperClient;
import org.apache.fluss.server.zk.ZooKeeperExtension;
import org.apache.fluss.server.zk.data.BucketAssignment;
@@ -148,6 +149,7 @@ class CoordinatorEventProcessorTest {
private static ZooKeeperClient zookeeperClient;
private static MetadataManager metadataManager;
+ private static ZkEpoch zkEpoch;
private CoordinatorEventProcessor eventProcessor;
private final String defaultDatabase = "db";
@@ -176,6 +178,7 @@ class CoordinatorEventProcessorTest {
new CoordinatorAddress(
"2", Endpoint.fromListenersString("CLIENT://localhost:10012")));
+ zkEpoch = zookeeperClient.fenceBecomeCoordinatorLeader("2");
// register 3 tablet servers
for (int i = 0; i < 3; i++) {
zookeeperClient.registerTabletServer(
@@ -1058,9 +1061,10 @@ class CoordinatorEventProcessorTest {
private CoordinatorEventProcessor buildCoordinatorEventProcessor() {
return new CoordinatorEventProcessor(
zookeeperClient,
+ zkEpoch,
serverMetadataCache,
testCoordinatorChannelManager,
- new CoordinatorContext(),
+ new TestCoordinatorContext(),
autoPartitionManager,
lakeTableTieringManager,
TestingMetricGroups.COORDINATOR_METRICS,
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 5d2bcd574..e974ec2d9 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,18 +23,25 @@ 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.exception.InvalidCoordinatorException;
import org.apache.fluss.exception.NotCoordinatorLeaderException;
+import org.apache.fluss.metadata.TableBucket;
import org.apache.fluss.rpc.GatewayClientProxy;
import org.apache.fluss.rpc.RpcClient;
import org.apache.fluss.rpc.gateway.CoordinatorGateway;
+import org.apache.fluss.rpc.gateway.TabletServerGateway;
import org.apache.fluss.rpc.messages.CreateDatabaseRequest;
import org.apache.fluss.rpc.messages.MetadataRequest;
+import org.apache.fluss.rpc.messages.UpdateMetadataRequest;
import org.apache.fluss.rpc.metrics.TestingClientMetricGroup;
+import org.apache.fluss.server.tablet.TabletServer;
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.data.CoordinatorAddress;
+import org.apache.fluss.server.zk.data.LeaderAndIsr;
import org.apache.fluss.shaded.curator5.org.apache.curator.framework.CuratorFramework;
+import org.apache.fluss.shaded.zookeeper3.org.apache.zookeeper.KeeperException;
import org.apache.fluss.shaded.zookeeper3.org.apache.zookeeper.Watcher;
import org.apache.fluss.shaded.zookeeper3.org.apache.zookeeper.ZooKeeper;
import org.apache.fluss.testutils.common.AllCallbackWrapper;
@@ -44,7 +51,9 @@ import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.RegisterExtension;
+import org.junit.jupiter.api.io.TempDir;
+import java.nio.file.Path;
import java.time.Duration;
import java.util.Arrays;
import java.util.List;
@@ -78,8 +87,11 @@ class CoordinatorHighAvailabilityITCase {
private CoordinatorServer coordinatorServer1;
private CoordinatorServer coordinatorServer2;
+ private TabletServer tabletServer;
private RpcClient rpcClient;
+ @TempDir Path tempDir;
+
@BeforeAll
static void baseBeforeAll() {
zookeeperClient =
@@ -102,6 +114,9 @@ class CoordinatorHighAvailabilityITCase {
if (coordinatorServer2 != null) {
coordinatorServer2.close();
}
+ if (tabletServer != null) {
+ tabletServer.close();
+ }
if (rpcClient != null) {
rpcClient.close();
}
@@ -242,6 +257,122 @@ class CoordinatorHighAvailabilityITCase {
createGatewayForServer(finalLeader).metadata(new MetadataRequest()).get();
}
+ @Test
+ void testTabletServerRejectsStaleCoordinatorEpochAfterLeaderSwitch() throws Exception {
+ coordinatorServer1 = new CoordinatorServer(createConfiguration());
+ coordinatorServer2 = new CoordinatorServer(createConfiguration());
+ tabletServer = new TabletServer(createTabletServerConfiguration());
+
+ coordinatorServer1.start();
+ coordinatorServer2.start();
+ tabletServer.start();
+
+ waitUntilCoordinatorServerElected();
+ CoordinatorAddress firstLeaderAddr = zookeeperClient.getCoordinatorLeaderAddress().get();
+
+ CoordinatorServer leader = findServerById(firstLeaderAddr.getId());
+ CoordinatorServer standby = findServerByNotId(firstLeaderAddr.getId());
+
+ // Record old coordinator epoch before killing the leader
+ int oldCoordinatorEpoch = leader.getCoordinatorEventProcessor().getCoordinatorEpoch();
+
+ killZkSession(leader);
+ waitUntilNewLeaderElected(leader.getServerId());
+ assertThat(zookeeperClient.getCoordinatorLeaderAddress().get().getId())
+ .as("After killing leader, standby should become leader")
+ .isEqualTo(standby.getServerId());
+
+ int newCoordinatorEpoch = standby.getCoordinatorEventProcessor().getCoordinatorEpoch();
+
+ TabletServerGateway tsGateway = createGatewayForTabletServer(tabletServer);
+
+ // Send request with new coordinator epoch first to ensure the tablet server
+ // has updated its stored epoch to the latest value
+ tsGateway
+ .updateMetadata(
+ new UpdateMetadataRequest().setCoordinatorEpoch(newCoordinatorEpoch))
+ .get();
+
+ // Send request with old coordinator epoch — tablet server should reject it
+ assertThatThrownBy(
+ () ->
+ tsGateway
+ .updateMetadata(
+ new UpdateMetadataRequest()
+ .setCoordinatorEpoch(oldCoordinatorEpoch))
+ .get())
+ .satisfies(
+ t ->
+ assertThat(getRootCause(t))
+ .isInstanceOf(InvalidCoordinatorException.class));
+ }
+
+ @Test
+ void testZooKeeperRejectsStaleCoordinatorRequestAfterLeaderSwitch() throws Exception {
+ // 1. Start two coordinators, confirm leader
+ // 2. Record current coordinator epoch
+ // 3. Kill leader's ZK session, trigger leader switch
+ // 4. Wait for new leader election (epoch should increment)
+ // 5. Verify new leader can send requests to ZooKeeper
+ // 6. Verify requests with old ZkVersion epoch are rejected with BadVersionException
+ 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());
+
+ killZkSession(leader);
+ waitUntilNewLeaderElected(leader.getServerId());
+ assertThat(zookeeperClient.getCoordinatorLeaderAddress().get().getId())
+ .as("After killing leader, standby should become leader")
+ .isEqualTo(standby.getServerId());
+
+ TableBucket tableBucket = new TableBucket(1, 1);
+ LeaderAndIsr leaderAndIsr = new LeaderAndIsr(0, 0, Arrays.asList(2, 3), 0, 0);
+
+ int newLeaderEpochZkVersion =
+ standby.getCoordinatorEventProcessor()
+ .getCoordinatorContext()
+ .getCoordinatorEpochZkVersion();
+ int oldLeaderEpochZkVersion =
+ leader.getCoordinatorEventProcessor()
+ .getCoordinatorContext()
+ .getCoordinatorEpochZkVersion();
+
+ assertThatThrownBy(
+ () ->
+ leader.getZooKeeperClient()
+ .registerLeaderAndIsr(
+ tableBucket, leaderAndIsr, oldLeaderEpochZkVersion))
+ .satisfies(
+ t ->
+ assertThat(getRootCause(t))
+ .isInstanceOf(KeeperException.BadVersionException.class));
+ standby.getZooKeeperClient()
+ .registerLeaderAndIsr(tableBucket, leaderAndIsr, newLeaderEpochZkVersion);
+ assertThat(zookeeperClient.getLeaderAndIsr(tableBucket)).hasValue(leaderAndIsr);
+ }
+
+ private TabletServerGateway createGatewayForTabletServer(TabletServer server) {
+ List<Endpoint> endpoints = server.getRpcServer().getBindEndpoints();
+ Endpoint endpoint = endpoints.get(0);
+ ServerNode serverNode =
+ new ServerNode(
+ server.getServerId(),
+ endpoint.getHost(),
+ endpoint.getPort(),
+ ServerType.TABLET_SERVER);
+
+ return GatewayClientProxy.createGatewayProxy(
+ () -> serverNode, rpcClient, TabletServerGateway.class);
+ }
+
private CoordinatorGateway createGatewayForServer(CoordinatorServer server) {
List<Endpoint> endpoints = server.getRpcServer().getBindEndpoints();
Endpoint endpoint = endpoints.get(0);
@@ -275,6 +406,13 @@ class CoordinatorHighAvailabilityITCase {
return configuration;
}
+ private Configuration createTabletServerConfiguration() {
+ Configuration configuration = createConfiguration();
+ configuration.set(ConfigOptions.TABLET_SERVER_ID, 0);
+ configuration.setString(ConfigOptions.DATA_DIR, tempDir.toAbsolutePath().toString());
+ return configuration;
+ }
+
private void waitUntilCoordinatorServerElected() {
waitUntil(
() -> zookeeperClient.getCoordinatorLeaderAddress().isPresent(),
diff --git a/fluss-server/src/test/java/org/apache/fluss/server/coordinator/CoordinatorServerElectionTest.java b/fluss-server/src/test/java/org/apache/fluss/server/coordinator/CoordinatorServerElectionTest.java
index cf832a955..e77182393 100644
--- a/fluss-server/src/test/java/org/apache/fluss/server/coordinator/CoordinatorServerElectionTest.java
+++ b/fluss-server/src/test/java/org/apache/fluss/server/coordinator/CoordinatorServerElectionTest.java
@@ -83,6 +83,8 @@ class CoordinatorServerElectionTest {
}
}
assertThat(firstLeader).isNotNull();
+ assertThat(zookeeperClient.getCurrentEpoch().getCoordinatorEpoch())
+ .isEqualTo(CoordinatorContext.INITIAL_COORDINATOR_EPOCH);
firstLeader.close();
firstLeader.start();
@@ -91,6 +93,8 @@ class CoordinatorServerElectionTest {
CoordinatorAddress secondLeaderAddress =
zookeeperClient.getCoordinatorLeaderAddress().get();
assertThat(secondLeaderAddress).isNotEqualTo(firstLeaderAddress);
+ assertThat(zookeeperClient.getCurrentEpoch().getCoordinatorEpoch())
+ .isEqualTo(CoordinatorContext.INITIAL_COORDINATOR_EPOCH + 1);
CoordinatorServer secondLeader = null;
for (CoordinatorServer coordinatorServer : coordinatorServerList) {
@@ -117,6 +121,8 @@ class CoordinatorServerElectionTest {
CoordinatorAddress thirdLeaderAddress = zookeeperClient.getCoordinatorLeaderAddress().get();
assertThat(thirdLeaderAddress.getId()).isEqualTo(firstLeaderAddress.getId());
+ assertThat(zookeeperClient.getCurrentEpoch().getCoordinatorEpoch())
+ .isEqualTo(CoordinatorContext.INITIAL_COORDINATOR_EPOCH + 2);
}
/** Create a configuration with Zookeeper address setting. */
diff --git a/fluss-server/src/test/java/org/apache/fluss/server/coordinator/CoordinatorServerITCase.java b/fluss-server/src/test/java/org/apache/fluss/server/coordinator/CoordinatorServerITCase.java
index ae39bd185..bd060d6ee 100644
--- a/fluss-server/src/test/java/org/apache/fluss/server/coordinator/CoordinatorServerITCase.java
+++ b/fluss-server/src/test/java/org/apache/fluss/server/coordinator/CoordinatorServerITCase.java
@@ -63,7 +63,6 @@ class CoordinatorServerITCase extends ServerITCaseBase {
ConfigOptions.BIND_LISTENERS,
String.format("%s://%s:%d", DEFAULT_LISTENER_NAME, HOSTNAME, getPort()));
conf.set(ConfigOptions.REMOTE_DATA_DIR, "/tmp/fluss/remote-data");
-
return conf;
}
diff --git a/fluss-server/src/test/java/org/apache/fluss/server/coordinator/TableManagerITCase.java b/fluss-server/src/test/java/org/apache/fluss/server/coordinator/TableManagerITCase.java
index 9f6c13064..b2abac830 100644
--- a/fluss-server/src/test/java/org/apache/fluss/server/coordinator/TableManagerITCase.java
+++ b/fluss-server/src/test/java/org/apache/fluss/server/coordinator/TableManagerITCase.java
@@ -514,6 +514,7 @@ class TableManagerITCase {
.updateMetadata(
makeUpdateMetadataRequest(
coordinatorServerInfo,
+ null,
new HashSet<>(tabletServerInfos),
Collections.emptyList(),
Collections.emptyList()))
@@ -652,6 +653,10 @@ class TableManagerITCase {
.updateMetadata(
makeLegacyUpdateMetadataRequest(
Optional.of(coordinatorServerInfo),
+ FLUSS_CLUSTER_EXTENSION
+ .getCoordinatorServer()
+ .getCoordinatorEventProcessor()
+ .getCoordinatorEpoch(),
new HashSet<>(tabletServerInfos)))
.get();
}
@@ -873,7 +878,9 @@ class TableManagerITCase {
@SuppressWarnings("OptionalUsedAsFieldOrParameterType")
private UpdateMetadataRequest makeLegacyUpdateMetadataRequest(
- Optional<ServerInfo> coordinatorServer, Set<ServerInfo> aliveTableServers) {
+ Optional<ServerInfo> coordinatorServer,
+ int coordinatorEpoch,
+ Set<ServerInfo> aliveTableServers) {
UpdateMetadataRequest updateMetadataRequest = new UpdateMetadataRequest();
Set<PbServerNode> aliveTableServerNodes = new HashSet<>();
for (ServerInfo serverInfo : aliveTableServers) {
@@ -895,6 +902,7 @@ class TableManagerITCase {
node -> {
Endpoint endpoint = node.endpoints().get(0);
updateMetadataRequest
+ .setCoordinatorEpoch(coordinatorEpoch)
.setCoordinatorServer()
.setNodeId(node.id())
.setHost(endpoint.getHost())
diff --git a/fluss-server/src/test/java/org/apache/fluss/server/coordinator/TableManagerTest.java b/fluss-server/src/test/java/org/apache/fluss/server/coordinator/TableManagerTest.java
index 024b12b33..2c30fc748 100644
--- a/fluss-server/src/test/java/org/apache/fluss/server/coordinator/TableManagerTest.java
+++ b/fluss-server/src/test/java/org/apache/fluss/server/coordinator/TableManagerTest.java
@@ -113,7 +113,7 @@ class TableManagerTest {
private void initTableManager() {
testingEventManager = new TestingEventManager();
- coordinatorContext = new CoordinatorContext();
+ coordinatorContext = new TestCoordinatorContext();
testCoordinatorChannelManager = new TestCoordinatorChannelManager();
Configuration conf = new Configuration();
conf.setString(ConfigOptions.REMOTE_DATA_DIR, "/tmp/fluss/remote-data");
diff --git a/fluss-server/src/test/java/org/apache/fluss/server/coordinator/TestCoordinatorContext.java b/fluss-server/src/test/java/org/apache/fluss/server/coordinator/TestCoordinatorContext.java
new file mode 100644
index 000000000..b931d564f
--- /dev/null
+++ b/fluss-server/src/test/java/org/apache/fluss/server/coordinator/TestCoordinatorContext.java
@@ -0,0 +1,39 @@
+/*
+ * 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.fluss.server.coordinator;
+
+import org.apache.fluss.server.zk.data.ZkVersion;
+
+/** A coordinator context for test purpose which can set epoch manually. */
+public class TestCoordinatorContext extends CoordinatorContext {
+ public TestCoordinatorContext() {
+ // When create or modify ZooKeeper node, it should check ZooKeeper epoch node version
+ // to ensure the coordinator is still holding the leadership. However, in the test
+ // cases, we don't register epoch node, so we skip the check process by setting
+ // "coordinatorEpochZkVersion" to ZkVersion.MATCH_ANY_VERSION
+ super();
+ this.setCoordinatorEpochAndZkVersion(
+ INITIAL_COORDINATOR_EPOCH, ZkVersion.MATCH_ANY_VERSION.getVersion());
+ }
+
+ public TestCoordinatorContext(int coordinatorEpoch, int coordinatorEpochZkVersion) {
+ super();
+ this.setCoordinatorEpochAndZkVersion(coordinatorEpoch, coordinatorEpochZkVersion);
+ }
+}
diff --git a/fluss-server/src/test/java/org/apache/fluss/server/coordinator/rebalance/RebalanceManagerTest.java b/fluss-server/src/test/java/org/apache/fluss/server/coordinator/rebalance/RebalanceManagerTest.java
index a961dc393..57ba9fff7 100644
--- a/fluss-server/src/test/java/org/apache/fluss/server/coordinator/rebalance/RebalanceManagerTest.java
+++ b/fluss-server/src/test/java/org/apache/fluss/server/coordinator/rebalance/RebalanceManagerTest.java
@@ -30,6 +30,7 @@ import org.apache.fluss.server.coordinator.lease.KvSnapshotLeaseManager;
import org.apache.fluss.server.metadata.CoordinatorMetadataCache;
import org.apache.fluss.server.metrics.group.TestingMetricGroups;
import org.apache.fluss.server.zk.NOPErrorHandler;
+import org.apache.fluss.server.zk.ZkEpoch;
import org.apache.fluss.server.zk.ZooKeeperClient;
import org.apache.fluss.server.zk.ZooKeeperExtension;
import org.apache.fluss.server.zk.data.RebalanceTask;
@@ -60,6 +61,7 @@ public class RebalanceManagerTest {
private static ZooKeeperClient zookeeperClient;
private static MetadataManager metadataManager;
+ private static ZkEpoch zkEpoch;
private CoordinatorMetadataCache serverMetadataCache;
private TestCoordinatorChannelManager testCoordinatorChannelManager;
@@ -69,11 +71,12 @@ public class RebalanceManagerTest {
private KvSnapshotLeaseManager kvSnapshotLeaseManager;
@BeforeAll
- static void baseBeforeAll() {
+ static void baseBeforeAll() throws Exception {
zookeeperClient =
ZOO_KEEPER_EXTENSION_WRAPPER
.getCustomExtension()
.getZooKeeperClient(NOPErrorHandler.INSTANCE);
+ zkEpoch = zookeeperClient.fenceBecomeCoordinatorLeader("1");
}
@BeforeEach
@@ -134,6 +137,7 @@ public class RebalanceManagerTest {
private CoordinatorEventProcessor buildCoordinatorEventProcessor() {
return new CoordinatorEventProcessor(
zookeeperClient,
+ zkEpoch,
serverMetadataCache,
testCoordinatorChannelManager,
new CoordinatorContext(),
diff --git a/fluss-server/src/test/java/org/apache/fluss/server/coordinator/statemachine/ReplicaStateMachineTest.java b/fluss-server/src/test/java/org/apache/fluss/server/coordinator/statemachine/ReplicaStateMachineTest.java
index c7c7591d5..4c89b5722 100644
--- a/fluss-server/src/test/java/org/apache/fluss/server/coordinator/statemachine/ReplicaStateMachineTest.java
+++ b/fluss-server/src/test/java/org/apache/fluss/server/coordinator/statemachine/ReplicaStateMachineTest.java
@@ -30,6 +30,7 @@ import org.apache.fluss.server.coordinator.CoordinatorContext;
import org.apache.fluss.server.coordinator.CoordinatorRequestBatch;
import org.apache.fluss.server.coordinator.CoordinatorTestUtils;
import org.apache.fluss.server.coordinator.TestCoordinatorChannelManager;
+import org.apache.fluss.server.coordinator.TestCoordinatorContext;
import org.apache.fluss.server.coordinator.event.DeleteReplicaResponseReceivedEvent;
import org.apache.fluss.server.entity.DeleteReplicaResultForBucket;
import org.apache.fluss.server.metadata.ServerInfo;
@@ -37,6 +38,7 @@ 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.data.LeaderAndIsr;
+import org.apache.fluss.server.zk.data.ZkVersion;
import org.apache.fluss.testutils.common.AllCallbackWrapper;
import org.junit.jupiter.api.BeforeAll;
@@ -80,7 +82,7 @@ class ReplicaStateMachineTest {
@Test
void testStartup() {
- CoordinatorContext coordinatorContext = new CoordinatorContext();
+ CoordinatorContext coordinatorContext = new TestCoordinatorContext();
// init coordinator server context with a table assignment
TableBucket tableBucket = new TableBucket(1, 0);
@@ -105,7 +107,7 @@ class ReplicaStateMachineTest {
@Test
void testReplicaStateChange() {
- CoordinatorContext coordinatorContext = new CoordinatorContext();
+ CoordinatorContext coordinatorContext = new TestCoordinatorContext();
ReplicaStateMachine replicaStateMachine = createReplicaStateMachine(coordinatorContext);
// test check valid replica state change
@@ -134,7 +136,7 @@ class ReplicaStateMachineTest {
@Test
void testDeleteReplicaStateChange() {
Map<TableBucketReplica, Boolean> isReplicaDeleteSuccess = new HashMap<>();
- CoordinatorContext coordinatorContext = new CoordinatorContext();
+ CoordinatorContext coordinatorContext = new TestCoordinatorContext();
coordinatorContext.setLiveTabletServers(
CoordinatorTestUtils.createServers(Arrays.asList(0, 1)));
// use a context that will return a gateway that always get success ack
@@ -191,7 +193,7 @@ class ReplicaStateMachineTest {
@Test
void testOfflineReplicasShouldBeRemovedFromIsr() throws Exception {
- CoordinatorContext coordinatorContext = new CoordinatorContext();
+ CoordinatorContext coordinatorContext = new TestCoordinatorContext();
coordinatorContext.setLiveTabletServers(createServers(new int[] {0, 1, 2}));
ReplicaStateMachine replicaStateMachine = createReplicaStateMachine(coordinatorContext);
@@ -216,7 +218,8 @@ class ReplicaStateMachineTest {
}
// put leader and isr
LeaderAndIsr leaderAndIsr = new LeaderAndIsr(0, 0, Arrays.asList(0, 1, 2), 0, 0);
- zookeeperClient.registerLeaderAndIsr(tableBucket, leaderAndIsr);
+ zookeeperClient.registerLeaderAndIsr(
+ tableBucket, leaderAndIsr, ZkVersion.MATCH_ANY_VERSION.getVersion());
coordinatorContext.updateBucketReplicaAssignment(tableBucket, Arrays.asList(0, 1, 2));
coordinatorContext.putBucketLeaderAndIsr(tableBucket, leaderAndIsr);
@@ -230,7 +233,7 @@ class ReplicaStateMachineTest {
@Test
void testOfflineReplicaShouldBeRemovedFromIsr() throws Exception {
- CoordinatorContext coordinatorContext = new CoordinatorContext();
+ CoordinatorContext coordinatorContext = new TestCoordinatorContext();
coordinatorContext.setLiveTabletServers(createServers(new int[] {0, 1, 2}));
ReplicaStateMachine replicaStateMachine = createReplicaStateMachine(coordinatorContext);
@@ -253,7 +256,8 @@ class ReplicaStateMachineTest {
}
// put leader and isr
LeaderAndIsr leaderAndIsr = new LeaderAndIsr(0, 0, Arrays.asList(0, 1, 2), 0, 0);
- zookeeperClient.registerLeaderAndIsr(tableBucket, leaderAndIsr);
+ zookeeperClient.registerLeaderAndIsr(
+ tableBucket, leaderAndIsr, ZkVersion.MATCH_ANY_VERSION.getVersion());
coordinatorContext.updateBucketReplicaAssignment(tableBucket, Arrays.asList(0, 1, 2));
coordinatorContext.putBucketLeaderAndIsr(tableBucket, leaderAndIsr);
diff --git a/fluss-server/src/test/java/org/apache/fluss/server/coordinator/statemachine/TableBucketStateMachineTest.java b/fluss-server/src/test/java/org/apache/fluss/server/coordinator/statemachine/TableBucketStateMachineTest.java
index bbac45c6b..d6697f8cb 100644
--- a/fluss-server/src/test/java/org/apache/fluss/server/coordinator/statemachine/TableBucketStateMachineTest.java
+++ b/fluss-server/src/test/java/org/apache/fluss/server/coordinator/statemachine/TableBucketStateMachineTest.java
@@ -34,6 +34,7 @@ import org.apache.fluss.server.coordinator.LakeCatalogDynamicLoader;
import org.apache.fluss.server.coordinator.LakeTableTieringManager;
import org.apache.fluss.server.coordinator.MetadataManager;
import org.apache.fluss.server.coordinator.TestCoordinatorChannelManager;
+import org.apache.fluss.server.coordinator.TestCoordinatorContext;
import org.apache.fluss.server.coordinator.event.CoordinatorEventManager;
import org.apache.fluss.server.coordinator.lease.KvSnapshotLeaseManager;
import org.apache.fluss.server.coordinator.statemachine.ReplicaLeaderElection.ControlledShutdownLeaderElection;
@@ -41,9 +42,11 @@ import org.apache.fluss.server.coordinator.statemachine.TableBucketStateMachine.
import org.apache.fluss.server.metadata.CoordinatorMetadataCache;
import org.apache.fluss.server.metrics.group.TestingMetricGroups;
import org.apache.fluss.server.zk.NOPErrorHandler;
+import org.apache.fluss.server.zk.ZkEpoch;
import org.apache.fluss.server.zk.ZooKeeperClient;
import org.apache.fluss.server.zk.ZooKeeperExtension;
import org.apache.fluss.server.zk.data.LeaderAndIsr;
+import org.apache.fluss.server.zk.data.ZkVersion;
import org.apache.fluss.shaded.guava32.com.google.common.collect.Sets;
import org.apache.fluss.testutils.common.AllCallbackWrapper;
import org.apache.fluss.utils.clock.SystemClock;
@@ -84,6 +87,8 @@ class TableBucketStateMachineTest {
private static ZooKeeperClient zookeeperClient;
private static CoordinatorContext coordinatorContext;
+ private static ZkEpoch zkEpoch;
+
private TestCoordinatorChannelManager testCoordinatorChannelManager;
private CoordinatorRequestBatch coordinatorRequestBatch;
private AutoPartitionManager autoPartitionManager;
@@ -92,11 +97,12 @@ class TableBucketStateMachineTest {
private KvSnapshotLeaseManager kvSnapshotLeaseManager;
@BeforeAll
- static void baseBeforeAll() {
+ static void baseBeforeAll() throws Exception {
zookeeperClient =
ZOO_KEEPER_EXTENSION_WRAPPER
.getCustomExtension()
.getZooKeeperClient(NOPErrorHandler.INSTANCE);
+ zkEpoch = zookeeperClient.fenceBecomeCoordinatorLeader("1");
}
@BeforeEach
@@ -105,7 +111,7 @@ class TableBucketStateMachineTest {
conf.setString(ConfigOptions.COORDINATOR_HOST, "localhost");
String remoteDir = "/tmp/fluss/remote-data";
conf.setString(ConfigOptions.REMOTE_DATA_DIR, remoteDir);
- coordinatorContext = new CoordinatorContext();
+ coordinatorContext = new TestCoordinatorContext();
testCoordinatorChannelManager = new TestCoordinatorChannelManager();
coordinatorRequestBatch =
new CoordinatorRequestBatch(
@@ -157,9 +163,13 @@ class TableBucketStateMachineTest {
// create LeaderAndIsr for t10/t11 info in zk,
zookeeperClient.registerLeaderAndIsr(
- new TableBucket(t1Id, 0), new LeaderAndIsr(0, 0, Arrays.asList(0, 1), 0, 0));
+ new TableBucket(t1Id, 0),
+ new LeaderAndIsr(0, 0, Arrays.asList(0, 1), 0, 0),
+ ZkVersion.MATCH_ANY_VERSION.getVersion());
zookeeperClient.registerLeaderAndIsr(
- new TableBucket(t1Id, 1), new LeaderAndIsr(2, 0, Arrays.asList(2, 3), 0, 0));
+ new TableBucket(t1Id, 1),
+ new LeaderAndIsr(2, 0, Arrays.asList(2, 3), 0, 0),
+ ZkVersion.MATCH_ANY_VERSION.getVersion());
// update the LeaderAndIsr to context
coordinatorContext.putBucketLeaderAndIsr(
t1b0, zookeeperClient.getLeaderAndIsr(new TableBucket(t1Id, 0)).get());
@@ -223,6 +233,8 @@ class TableBucketStateMachineTest {
coordinatorContext.putTablePath(tableId, fakeTablePath);
coordinatorContext.updateBucketReplicaAssignment(tableBucket, Arrays.asList(0, 1, 2));
coordinatorContext.putBucketState(tableBucket, NewBucket);
+ coordinatorContext.setCoordinatorEpochAndZkVersion(
+ 0, ZkVersion.MATCH_ANY_VERSION.getVersion());
// case1: init a new leader for NewBucket to OnlineBucket
tableBucketStateMachine.handleStateChange(Collections.singleton(tableBucket), OnlineBucket);
// non any alive servers, the state change fail
@@ -268,6 +280,7 @@ class TableBucketStateMachineTest {
CoordinatorEventProcessor coordinatorEventProcessor =
new CoordinatorEventProcessor(
zookeeperClient,
+ zkEpoch,
serverMetadataCache,
new CoordinatorChannelManager(
RpcClient.create(
diff --git a/fluss-server/src/test/java/org/apache/fluss/server/metadata/ZkBasedMetadataProviderTest.java b/fluss-server/src/test/java/org/apache/fluss/server/metadata/ZkBasedMetadataProviderTest.java
index aac6f19ed..5597bb01b 100644
--- a/fluss-server/src/test/java/org/apache/fluss/server/metadata/ZkBasedMetadataProviderTest.java
+++ b/fluss-server/src/test/java/org/apache/fluss/server/metadata/ZkBasedMetadataProviderTest.java
@@ -36,6 +36,7 @@ import org.apache.fluss.server.zk.data.LeaderAndIsr;
import org.apache.fluss.server.zk.data.PartitionAssignment;
import org.apache.fluss.server.zk.data.TableAssignment;
import org.apache.fluss.server.zk.data.TableRegistration;
+import org.apache.fluss.server.zk.data.ZkVersion;
import org.apache.fluss.testutils.common.AllCallbackWrapper;
import org.junit.jupiter.api.AfterAll;
@@ -110,8 +111,10 @@ class ZkBasedMetadataProviderTest {
LeaderAndIsr leaderAndIsr0 = new LeaderAndIsr(1, 10, Arrays.asList(1, 2, 3), 100, 1000);
LeaderAndIsr leaderAndIsr1 = new LeaderAndIsr(2, 20, Arrays.asList(2, 3, 4), 200, 2000);
- zookeeperClient.registerLeaderAndIsr(tableBucket0, leaderAndIsr0);
- zookeeperClient.registerLeaderAndIsr(tableBucket1, leaderAndIsr1);
+ zookeeperClient.registerLeaderAndIsr(
+ tableBucket0, leaderAndIsr0, ZkVersion.MATCH_ANY_VERSION.getVersion());
+ zookeeperClient.registerLeaderAndIsr(
+ tableBucket1, leaderAndIsr1, ZkVersion.MATCH_ANY_VERSION.getVersion());
List<TableMetadata> tablesMetadataFromZK =
metadataProvider.getTablesMetadataFromZK(
@@ -176,8 +179,10 @@ class ZkBasedMetadataProviderTest {
LeaderAndIsr leaderAndIsr0 = new LeaderAndIsr(1, 10, Arrays.asList(1, 2), 100, 1000);
LeaderAndIsr leaderAndIsr1 = new LeaderAndIsr(2, 20, Arrays.asList(2, 3), 200, 2000);
- zookeeperClient.registerLeaderAndIsr(partitionBucket0, leaderAndIsr0);
- zookeeperClient.registerLeaderAndIsr(partitionBucket1, leaderAndIsr1);
+ zookeeperClient.registerLeaderAndIsr(
+ partitionBucket0, leaderAndIsr0, ZkVersion.MATCH_ANY_VERSION.getVersion());
+ zookeeperClient.registerLeaderAndIsr(
+ partitionBucket1, leaderAndIsr1, ZkVersion.MATCH_ANY_VERSION.getVersion());
// Test getPartitionMetadataFromZkAsync
PhysicalTablePath partitionPath = PhysicalTablePath.of(tablePath, partitionName);
@@ -275,11 +280,17 @@ class ZkBasedMetadataProviderTest {
TableBucket bucket3 = new TableBucket(tableId2, partitionId3, 0);
zookeeperClient.registerLeaderAndIsr(
- bucket1, new LeaderAndIsr(1, 10, Arrays.asList(1, 2), 100, 1000));
+ bucket1,
+ new LeaderAndIsr(1, 10, Arrays.asList(1, 2), 100, 1000),
+ ZkVersion.MATCH_ANY_VERSION.getVersion());
zookeeperClient.registerLeaderAndIsr(
- bucket2, new LeaderAndIsr(2, 20, Arrays.asList(2, 3), 200, 2000));
+ bucket2,
+ new LeaderAndIsr(2, 20, Arrays.asList(2, 3), 200, 2000),
+ ZkVersion.MATCH_ANY_VERSION.getVersion());
zookeeperClient.registerLeaderAndIsr(
- bucket3, new LeaderAndIsr(1, 30, Arrays.asList(1, 3), 300, 3000));
+ bucket3,
+ new LeaderAndIsr(1, 30, Arrays.asList(1, 3), 300, 3000),
+ ZkVersion.MATCH_ANY_VERSION.getVersion());
// Test getPartitionsMetadataFromZK
List<PhysicalTablePath> partitionPaths =
diff --git a/fluss-server/src/test/java/org/apache/fluss/server/tablet/TabletServiceITCase.java b/fluss-server/src/test/java/org/apache/fluss/server/tablet/TabletServiceITCase.java
index 1137e499a..7bdd78107 100644
--- a/fluss-server/src/test/java/org/apache/fluss/server/tablet/TabletServiceITCase.java
+++ b/fluss-server/src/test/java/org/apache/fluss/server/tablet/TabletServiceITCase.java
@@ -890,6 +890,7 @@ public class TabletServiceITCase {
.updateMetadata(
makeUpdateMetadataRequest(
coordinatorServerInfo,
+ null,
newTabletServerInfos,
Collections.emptyList(),
Collections.emptyList()))
diff --git a/fluss-server/src/test/java/org/apache/fluss/server/zk/ZooKeeperClientTest.java b/fluss-server/src/test/java/org/apache/fluss/server/zk/ZooKeeperClientTest.java
index c867b8dce..16f59ddd8 100644
--- a/fluss-server/src/test/java/org/apache/fluss/server/zk/ZooKeeperClientTest.java
+++ b/fluss-server/src/test/java/org/apache/fluss/server/zk/ZooKeeperClientTest.java
@@ -55,6 +55,7 @@ import org.apache.fluss.utils.types.Tuple2;
import org.junit.jupiter.api.AfterAll;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.RegisterExtension;
import org.junit.jupiter.params.ParameterizedTest;
@@ -85,6 +86,7 @@ class ZooKeeperClientTest {
private static ZooKeeperClient zookeeperClient;
private static String remoteDataDir;
+ private static ZkEpoch zkEpoch;
@BeforeAll
static void beforeAll() {
@@ -95,6 +97,12 @@ class ZooKeeperClientTest {
remoteDataDir = zookeeperClient.getDefaultRemoteDataDir();
}
+ @BeforeEach
+ void beforeEach() throws Exception {
+ // create epoch node in beforeEach(), because it will always be cleaned up in afterEach()
+ zkEpoch = zookeeperClient.fenceBecomeCoordinatorLeader("tmp");
+ }
+
@AfterEach
void afterEach() {
ZOO_KEEPER_EXTENSION_WRAPPER.getCustomExtension().cleanupRoot();
@@ -176,7 +184,8 @@ class ZooKeeperClientTest {
// test update
TableAssignment tableAssignment3 =
TableAssignment.builder().add(3, BucketAssignment.of(1, 5)).build();
- zookeeperClient.updateTableAssignment(tableId1, tableAssignment3);
+ zookeeperClient.updateTableAssignment(
+ tableId1, tableAssignment3, zkEpoch.getCoordinatorEpochZkVersion());
assertThat(zookeeperClient.getTableAssignment(tableId1)).contains(tableAssignment3);
// test delete
@@ -193,19 +202,22 @@ class ZooKeeperClientTest {
assertThat(zookeeperClient.getLeaderAndIsr(tableBucket2)).isEmpty();
// try to register bucket leaderAndIsr
- LeaderAndIsr leaderAndIsr1 = new LeaderAndIsr(1, 10, Arrays.asList(1, 2, 3), 100, 1000);
- LeaderAndIsr leaderAndIsr2 = new LeaderAndIsr(2, 10, Arrays.asList(4, 5, 6), 100, 1000);
+ LeaderAndIsr leaderAndIsr1 = new LeaderAndIsr(1, 10, Arrays.asList(1, 2, 3), 0, 1000);
+ LeaderAndIsr leaderAndIsr2 = new LeaderAndIsr(2, 10, Arrays.asList(4, 5, 6), 0, 1000);
- zookeeperClient.registerLeaderAndIsr(tableBucket1, leaderAndIsr1);
- zookeeperClient.registerLeaderAndIsr(tableBucket2, leaderAndIsr2);
+ zookeeperClient.registerLeaderAndIsr(
+ tableBucket1, leaderAndIsr1, zkEpoch.getCoordinatorEpochZkVersion());
+ zookeeperClient.registerLeaderAndIsr(
+ tableBucket2, leaderAndIsr2, zkEpoch.getCoordinatorEpochZkVersion());
assertThat(zookeeperClient.getLeaderAndIsr(tableBucket1)).hasValue(leaderAndIsr1);
assertThat(zookeeperClient.getLeaderAndIsr(tableBucket2)).hasValue(leaderAndIsr2);
assertThat(zookeeperClient.getLeaderAndIsrs(Arrays.asList(tableBucket1, tableBucket2)))
.containsValues(leaderAndIsr1, leaderAndIsr2);
// test update
- leaderAndIsr1 = new LeaderAndIsr(2, 20, Collections.emptyList(), 200, 2000);
- zookeeperClient.updateLeaderAndIsr(tableBucket1, leaderAndIsr1);
+ leaderAndIsr1 = new LeaderAndIsr(2, 20, Collections.emptyList(), 0, 2000);
+ zookeeperClient.updateLeaderAndIsr(
+ tableBucket1, leaderAndIsr1, zkEpoch.getCoordinatorEpochZkVersion());
assertThat(zookeeperClient.getLeaderAndIsr(tableBucket1)).hasValue(leaderAndIsr1);
// test delete
@@ -222,7 +234,7 @@ class ZooKeeperClientTest {
TableBucket tableBucket =
isPartitionTable ? new TableBucket(1, 2L, i) : new TableBucket(1, i);
LeaderAndIsr leaderAndIsr =
- new LeaderAndIsr(i, 10, Arrays.asList(i + 1, i + 2, i + 3), 100, 1000);
+ new LeaderAndIsr(i, 10, Arrays.asList(i + 1, i + 2, i + 3), 0, 1000);
leaderAndIsrList.add(leaderAndIsr);
RegisterTableBucketLeadAndIsrInfo info =
isPartitionTable
@@ -233,7 +245,8 @@ class ZooKeeperClientTest {
tableBucketInfo.add(info);
}
// batch create
- zookeeperClient.batchRegisterLeaderAndIsrForTablePartition(tableBucketInfo);
+ zookeeperClient.batchRegisterLeaderAndIsrForTablePartition(
+ tableBucketInfo, zkEpoch.getCoordinatorEpochZkVersion());
for (int i = 0; i < 100; i++) {
// each should register successful
@@ -263,7 +276,7 @@ class ZooKeeperClientTest {
entry.setValue(adjustLeaderAndIsr);
});
// batch update
- zookeeperClient.batchUpdateLeaderAndIsr(updateMap);
+ zookeeperClient.batchUpdateLeaderAndIsr(updateMap, zkEpoch.getCoordinatorEpochZkVersion());
for (int i = 0; i < 100; i++) {
// each should update successful
Optional<LeaderAndIsr> optionalLeaderAndIsr =
@@ -282,9 +295,10 @@ class ZooKeeperClientTest {
for (int i = 0; i < totalCount; i++) {
TableBucket tableBucket = new TableBucket(1, i);
LeaderAndIsr leaderAndIsr =
- new LeaderAndIsr(i, 10, Arrays.asList(i + 1, i + 2, i + 3), 100, 1000);
+ new LeaderAndIsr(i, 10, Arrays.asList(i + 1, i + 2, i + 3), 0, 1000);
leaderAndIsrList.put(tableBucket, leaderAndIsr);
- zookeeperClient.registerLeaderAndIsr(tableBucket, leaderAndIsr);
+ zookeeperClient.registerLeaderAndIsr(
+ tableBucket, leaderAndIsr, zkEpoch.getCoordinatorEpochZkVersion());
}
// try to batch update
@@ -299,10 +313,11 @@ class ZooKeeperClientTest {
old.leader() + 1,
old.leaderEpoch() + 1,
old.isr(),
- old.coordinatorEpoch() + 1,
+ old.coordinatorEpoch(),
old.bucketEpoch() + 1);
}));
- zookeeperClient.batchUpdateLeaderAndIsr(updateLeaderAndIsrList);
+ zookeeperClient.batchUpdateLeaderAndIsr(
+ updateLeaderAndIsrList, zkEpoch.getCoordinatorEpochZkVersion());
for (Map.Entry<TableBucket, LeaderAndIsr> entry : updateLeaderAndIsrList.entrySet()) {
TableBucket tableBucket = entry.getKey();
LeaderAndIsr leaderAndIsr = entry.getValue();