This is an automated email from the ASF dual-hosted git repository.
wuchong 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 01c13e875 [kv] Support elect standby replicas for primary key table
(#2829)
01c13e875 is described below
commit 01c13e875b1730bf3825de5fbb3383f90fbbb0ef
Author: yunhong <[email protected]>
AuthorDate: Tue May 5 12:36:30 2026 +0800
[kv] Support elect standby replicas for primary key table (#2829)
---
fluss-rpc/src/main/proto/FlussApi.proto | 3 +
.../coordinator/CoordinatorEventProcessor.java | 1 +
.../statemachine/ReplicaLeaderElection.java | 201 +++++++++---
.../statemachine/ReplicaStateMachine.java | 3 +-
.../statemachine/TableBucketStateMachine.java | 60 +++-
.../server/entity/NotifyLeaderAndIsrData.java | 8 +
.../org/apache/fluss/server/replica/IsrState.java | 27 +-
.../org/apache/fluss/server/replica/Replica.java | 73 ++++-
.../fluss/server/utils/ServerRpcMessageUtils.java | 24 +-
.../apache/fluss/server/zk/data/LeaderAndIsr.java | 40 ++-
.../server/zk/data/LeaderAndIsrJsonSerde.java | 25 +-
.../coordinator/CoordinatorEventProcessorTest.java | 5 +-
.../CoordinatorHighAvailabilityITCase.java | 4 +-
.../server/coordinator/TestCoordinatorGateway.java | 1 +
.../statemachine/ReplicaLeaderElectionTest.java | 350 ++++++++++++++++++++-
.../statemachine/ReplicaStateMachineTest.java | 30 +-
.../statemachine/TableBucketStateMachineTest.java | 93 +++++-
.../fluss/server/log/remote/RemoteLogTestBase.java | 8 +-
.../metadata/ZkBasedMetadataProviderTest.java | 18 +-
.../fluss/server/replica/AdjustIsrITCase.java | 1 +
.../fluss/server/replica/AdjustIsrManagerTest.java | 36 ++-
.../apache/fluss/server/replica/AdjustIsrTest.java | 8 +-
.../replica/HighWatermarkPersistenceTest.java | 1 +
.../replica/KvRecoverFromRemoteLogITCase.java | 2 +
.../fluss/server/replica/ReplicaManagerTest.java | 4 +
.../apache/fluss/server/replica/ReplicaTest.java | 2 +
.../fluss/server/replica/ReplicaTestBase.java | 2 +
.../replica/fetcher/ReplicaFetcherITCase.java | 1 +
.../replica/fetcher/ReplicaFetcherManagerTest.java | 1 +
.../replica/fetcher/ReplicaFetcherThreadTest.java | 2 +
.../server/tablet/TabletServerShutdownITCase.java | 4 +-
.../fluss/server/tablet/TabletServiceITCase.java | 2 +
.../server/testutils/FlussClusterExtension.java | 5 +
.../fluss/server/zk/ZooKeeperClientTest.java | 31 +-
.../server/zk/data/LeaderAndIsrJsonSerdeTest.java | 38 ++-
35 files changed, 979 insertions(+), 135 deletions(-)
diff --git a/fluss-rpc/src/main/proto/FlussApi.proto
b/fluss-rpc/src/main/proto/FlussApi.proto
index b0f56cffb..e8381215f 100644
--- a/fluss-rpc/src/main/proto/FlussApi.proto
+++ b/fluss-rpc/src/main/proto/FlussApi.proto
@@ -959,6 +959,7 @@ message PbAdjustIsrReqForBucket {
repeated int32 new_isr = 4 [packed = true];
required int32 coordinator_epoch = 5;
required int32 bucket_epoch = 6;
+ repeated int32 standby_replicas = 7 [packed = true];
}
message PbAdjustIsrRespForTable {
@@ -976,6 +977,7 @@ message PbAdjustIsrRespForBucket {
repeated int32 isr = 7 [packed = true];
optional int32 bucket_epoch = 8;
optional int32 coordinator_epoch = 9;
+ repeated int32 standby_replicas = 10 [packed = true];
}
message PbListOffsetsRespForBucket {
@@ -996,6 +998,7 @@ message PbNotifyLeaderAndIsrReqForBucket {
repeated int32 replicas = 5 [packed = true];
repeated int32 isr = 6 [packed = true];
required int32 bucket_epoch = 7;
+ repeated int32 standby_replicas = 8 [packed = true];
}
message PbNotifyLeaderAndIsrRespForBucket {
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 89c294ea9..96f39ef08 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
@@ -1686,6 +1686,7 @@ public class CoordinatorEventProcessor implements
EventProcessor {
// TODO: reject the request if there is a replica
in ISR is not online,
// see KIP-841.
tryAdjustLeaderAndIsr.isr(),
+ tryAdjustLeaderAndIsr.standbyReplicas(),
coordinatorContext.getCoordinatorEpoch(),
currentLeaderAndIsr.bucketEpoch() + 1);
newLeaderAndIsrList.put(tableBucket, newLeaderAndIsr);
diff --git
a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/statemachine/ReplicaLeaderElection.java
b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/statemachine/ReplicaLeaderElection.java
index e9b0d33cf..d63714423 100644
---
a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/statemachine/ReplicaLeaderElection.java
+++
b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/statemachine/ReplicaLeaderElection.java
@@ -21,6 +21,7 @@ import
org.apache.fluss.server.coordinator.statemachine.TableBucketStateMachine.
import org.apache.fluss.server.zk.data.LeaderAndIsr;
import java.util.ArrayList;
+import java.util.Collections;
import java.util.HashSet;
import java.util.List;
import java.util.Optional;
@@ -41,23 +42,26 @@ public abstract class ReplicaLeaderElection {
* @param assignments the assignments
* @param aliveReplicas the alive replicas
* @param leaderAndIsr the original leaderAndIsr
+ * @param isPrimaryKeyTable whether this table bucket is primary key
table
* @return the election result
*/
public Optional<ElectionResult> leaderElection(
- List<Integer> assignments, List<Integer> aliveReplicas,
LeaderAndIsr leaderAndIsr) {
- // currently, we always use the first replica in assignment, which
also in aliveReplicas
- // and
- // isr as the leader replica.
+ List<Integer> assignments,
+ List<Integer> aliveReplicas,
+ LeaderAndIsr leaderAndIsr,
+ boolean isPrimaryKeyTable) {
List<Integer> isr = leaderAndIsr.isr();
- for (int assignment : assignments) {
- if (aliveReplicas.contains(assignment) &&
isr.contains(assignment)) {
- return Optional.of(
- new TableBucketStateMachine.ElectionResult(
- aliveReplicas,
leaderAndIsr.newLeaderAndIsr(assignment, isr)));
- }
- }
-
- return Optional.empty();
+ // First we will filter out the assignment list to only contain
the alive replicas and
+ // isr.
+ List<Integer> availableReplicas =
+ assignments.stream()
+ .filter(
+ replica ->
+ aliveReplicas.contains(replica)
+ && isr.contains(replica))
+ .collect(Collectors.toList());
+ return electLeader(
+ availableReplicas, aliveReplicas, isr, leaderAndIsr,
isPrimaryKeyTable);
}
}
@@ -70,32 +74,41 @@ public abstract class ReplicaLeaderElection {
* @param aliveReplicas the alive replicas
* @param leaderAndIsr the original leaderAndIsr
* @param shutdownTabletServers the shutdown tabletServers
+ * @param isPrimaryKeyTable whether this table bucket is primary key
table
* @return the election result
*/
public Optional<ElectionResult> leaderElection(
List<Integer> assignments,
List<Integer> aliveReplicas,
LeaderAndIsr leaderAndIsr,
- Set<Integer> shutdownTabletServers) {
+ Set<Integer> shutdownTabletServers,
+ boolean isPrimaryKeyTable) {
List<Integer> originIsr = leaderAndIsr.isr();
Set<Integer> isrSet = new HashSet<>(originIsr);
- for (Integer id : assignments) {
- if (aliveReplicas.contains(id)
- && isrSet.contains(id)
- && !shutdownTabletServers.contains(id)) {
- Set<Integer> newAliveReplicas = new
HashSet<>(aliveReplicas);
- newAliveReplicas.removeAll(shutdownTabletServers);
- List<Integer> newIsr =
- originIsr.stream()
- .filter(replica ->
!shutdownTabletServers.contains(replica))
- .collect(Collectors.toList());
- return Optional.of(
- new ElectionResult(
- new ArrayList<>(newAliveReplicas),
- leaderAndIsr.newLeaderAndIsr(id, newIsr)));
- }
- }
- return Optional.empty();
+ // Filter out available replicas: alive, in ISR, and not shutting
down.
+ List<Integer> availableReplicas =
+ assignments.stream()
+ .filter(
+ replica ->
+ aliveReplicas.contains(replica)
+ && isrSet.contains(replica)
+ &&
!shutdownTabletServers.contains(replica))
+ .collect(Collectors.toList());
+
+ // Filter alive replicas and ISR to exclude shutting-down servers.
+ Set<Integer> newAliveReplicaSet = new HashSet<>(aliveReplicas);
+ newAliveReplicaSet.removeAll(shutdownTabletServers);
+ List<Integer> newIsr =
+ originIsr.stream()
+ .filter(replica ->
!shutdownTabletServers.contains(replica))
+ .collect(Collectors.toList());
+
+ return electLeader(
+ availableReplicas,
+ new ArrayList<>(newAliveReplicaSet),
+ newIsr,
+ leaderAndIsr,
+ isPrimaryKeyTable);
}
}
@@ -108,20 +121,126 @@ public abstract class ReplicaLeaderElection {
}
public Optional<ElectionResult> leaderElection(
- List<Integer> liveReplicas, LeaderAndIsr leaderAndIsr) {
- // currently, we always use the first replica in targetReplicas,
which also in
- // liveReplicas and isr as the leader replica. For bucket
reassignment, the first
- // replica is the target leader replica.
+ List<Integer> liveReplicas, LeaderAndIsr leaderAndIsr, boolean
isPrimaryKeyTable) {
+ // For bucket reassignment, the first replica in newReplicas is
the target leader
+ // encoded by GoalOptimizerUtils. We must always honor this target
instead of using
+ // standby-promotion logic, otherwise the elected leader may
differ from the rebalance
+ // plan and tryToCompleteRebalanceTask will remain stuck in
REBALANCING state.
List<Integer> isr = leaderAndIsr.isr();
- for (int assignment : newReplicas) {
- if (liveReplicas.contains(assignment) &&
isr.contains(assignment)) {
- return Optional.of(
- new ElectionResult(
- liveReplicas,
leaderAndIsr.newLeaderAndIsr(assignment, isr)));
- }
+ List<Integer> availableReplicas =
+ newReplicas.stream()
+ .filter(
+ replica ->
+ liveReplicas.contains(replica) &&
isr.contains(replica))
+ .collect(Collectors.toList());
+
+ if (availableReplicas.isEmpty()) {
+ return Optional.empty();
}
+ // Always use the first available replica as leader to honor the
rebalance plan.
+ int newLeader = availableReplicas.get(0);
+ List<Integer> standbyReplicas =
+ isPrimaryKeyTable
+ ? findNewStandby(availableReplicas, newLeader)
+ : Collections.emptyList();
+
+ return Optional.of(
+ new ElectionResult(
+ liveReplicas,
+ leaderAndIsr.newLeaderAndIsr(newLeader, isr,
standbyReplicas)));
+ }
+ }
+
+ // ------------------------------------------------------------------------
+ // Common election logic
+ // ------------------------------------------------------------------------
+
+ private static Optional<ElectionResult> electLeader(
+ List<Integer> availableReplicas,
+ List<Integer> liveReplicas,
+ List<Integer> newIsr,
+ LeaderAndIsr leaderAndIsr,
+ boolean isPrimaryKeyTable) {
+ if (availableReplicas.isEmpty()) {
return Optional.empty();
}
+
+ // For log table, simply use the first available replica as leader.
+ if (!isPrimaryKeyTable) {
+ return Optional.of(
+ new ElectionResult(
+ liveReplicas,
+ leaderAndIsr.newLeaderAndIsr(
+ availableReplicas.get(0), newIsr,
Collections.emptyList())));
+ }
+
+ // For PK table, elect leader and standby.
+ LeaderAndStandby leaderAndStandby =
+ electLeaderAndStandbyForPkTable(availableReplicas,
leaderAndIsr);
+ return Optional.of(
+ new ElectionResult(
+ liveReplicas,
+ leaderAndIsr.newLeaderAndIsr(
+ leaderAndStandby.leader,
+ newIsr,
+ leaderAndStandby.standbyReplicas)));
+ }
+
+ /**
+ * Elect leader and standby for PK table.
+ *
+ * <p>Election strategy:
+ *
+ * <ul>
+ * <li>If current standby exists and is available, promote it to leader
+ * <li>Otherwise, use the first available replica as leader
+ * <li>Select new standby from remaining available replicas (if any)
+ * </ul>
+ */
+ private static LeaderAndStandby electLeaderAndStandbyForPkTable(
+ List<Integer> availableReplicas, LeaderAndIsr leaderAndIsr) {
+ int currentStandby = getCurrentStandby(leaderAndIsr);
+ int newLeader;
+
+ if (currentStandby != -1 &&
availableReplicas.contains(currentStandby)) {
+ // Promote current standby to leader.
+ newLeader = currentStandby;
+ } else {
+ // Use first available replica as leader.
+ newLeader = availableReplicas.get(0);
+ }
+
+ // Find new standby from remaining replicas.
+ List<Integer> standbyReplicas = findNewStandby(availableReplicas,
newLeader);
+ return new LeaderAndStandby(newLeader, standbyReplicas);
+ }
+
+ /** Get current standby replica ID, returns -1 if no standby exists. */
+ private static int getCurrentStandby(LeaderAndIsr leaderAndIsr) {
+ return leaderAndIsr.standbyReplicas().isEmpty()
+ ? -1
+ : leaderAndIsr.standbyReplicas().get(0);
+ }
+
+ /** Find new standby from available replicas, excluding the leader. */
+ private static List<Integer> findNewStandby(
+ List<Integer> availableReplicas, int excludeLeader) {
+ return availableReplicas.stream()
+ .filter(replica -> replica != excludeLeader)
+ .findFirst()
+ .map(Collections::singletonList)
+ .orElse(Collections.emptyList());
+ }
+
+ /** Internal class to hold leader and standby election result. */
+ private static class LeaderAndStandby {
+ final int leader;
+ final List<Integer> standbyReplicas;
+
+ LeaderAndStandby(int leader, List<Integer> standbyReplicas) {
+ this.leader = leader;
+ this.standbyReplicas = standbyReplicas;
+ }
}
}
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 63f5dd965..d35bb7578 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
@@ -481,7 +481,8 @@ public class ReplicaStateMachine {
.collect(Collectors.toList());
LeaderAndIsr adjustLeaderAndIsr =
newLeader == LeaderAndIsr.NO_LEADER
- ? leaderAndIsr.newLeaderAndIsr(newLeader, newIsr)
+ ? leaderAndIsr.newLeaderAndIsr(
+ newLeader, newIsr,
leaderAndIsr.standbyReplicas())
: leaderAndIsr.newLeaderAndIsr(newIsr);
adjustedLeaderAndIsr.put(tableBucketReplica, adjustLeaderAndIsr);
toUpdateLeaderAndIsrList.put(tableBucket, adjustLeaderAndIsr);
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 300e98f08..6a556ddea 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
@@ -440,7 +440,12 @@ public class TableBucketStateMachine {
// servers as inSyncReplica set.
Optional<ElectionResult> resultOpt =
initReplicaLeaderElection(
- assignedServers, liveServers,
coordinatorContext.getCoordinatorEpoch());
+ assignedServers,
+ liveServers,
+ coordinatorContext.getCoordinatorEpoch(),
+ coordinatorContext
+ .getTableInfoById(tableBucket.getTableId())
+ .hasPrimaryKey());
if (!resultOpt.isPresent()) {
LOG.error(
"The leader election for table bucket {} is empty.",
@@ -620,10 +625,12 @@ public class TableBucketStateMachine {
}
Optional<ElectionResult> resultOpt = Optional.empty();
+ boolean isPkTable =
+
coordinatorContext.getTableInfoById(tableBucket.getTableId()).hasPrimaryKey();
if (electionStrategy instanceof DefaultLeaderElection) {
resultOpt =
((DefaultLeaderElection) electionStrategy)
- .leaderElection(assignment, liveReplicas,
leaderAndIsr);
+ .leaderElection(assignment, liveReplicas,
leaderAndIsr, isPkTable);
} else if (electionStrategy instanceof
ControlledShutdownLeaderElection) {
Set<Integer> shuttingDownTabletServers =
coordinatorContext.shuttingDownTabletServers();
resultOpt =
@@ -632,11 +639,12 @@ public class TableBucketStateMachine {
assignment,
liveReplicas,
leaderAndIsr,
- shuttingDownTabletServers);
+ shuttingDownTabletServers,
+ isPkTable);
} else if (electionStrategy instanceof ReassignmentLeaderElection) {
resultOpt =
((ReassignmentLeaderElection) electionStrategy)
- .leaderElection(liveReplicas, leaderAndIsr);
+ .leaderElection(liveReplicas, leaderAndIsr,
isPkTable);
}
if (!resultOpt.isPresent()) {
@@ -677,23 +685,45 @@ public class TableBucketStateMachine {
* @param assignments the assignments
* @param aliveReplicas the alive replicas
* @param coordinatorEpoch the coordinator epoch
+ * @param isPrimaryKeyTable whether this table bucket is primary key table
* @return the election result
*/
@VisibleForTesting
public static Optional<ElectionResult> initReplicaLeaderElection(
- List<Integer> assignments, List<Integer> aliveReplicas, int
coordinatorEpoch) {
- // currently, we always use the first replica in assignment, which
also in aliveReplicas and
- // isr as the leader replica.
- for (int assignment : assignments) {
- if (aliveReplicas.contains(assignment)) {
- return Optional.of(
- new ElectionResult(
- aliveReplicas,
- new LeaderAndIsr(
- assignment, 0, aliveReplicas,
coordinatorEpoch, 0)));
+ List<Integer> assignments,
+ List<Integer> aliveReplicas,
+ int coordinatorEpoch,
+ boolean isPrimaryKeyTable) {
+ // First we will filter out the assignment list to only contain the
alive replicas.
+ List<Integer> availableReplicas =
+
assignments.stream().filter(aliveReplicas::contains).collect(Collectors.toList());
+
+ // If the assignment list is empty, we return empty.
+ if (availableReplicas.isEmpty()) {
+ return Optional.empty();
+ }
+
+ // Then we will use the first replica in assignment as the leader
replica.
+ int leader = availableReplicas.get(0);
+
+ // If this table is primaryKey table, we will use the second replica
in assignment as the
+ // standby if exists.
+ List<Integer> standbyReplicas = new ArrayList<>();
+ if (isPrimaryKeyTable) {
+ if (availableReplicas.size() > 1) {
+ standbyReplicas.add(availableReplicas.get(1));
}
}
- return Optional.empty();
+ return Optional.of(
+ new ElectionResult(
+ availableReplicas,
+ new LeaderAndIsr(
+ leader,
+ 0,
+ availableReplicas,
+ standbyReplicas,
+ coordinatorEpoch,
+ 0)));
}
}
diff --git
a/fluss-server/src/main/java/org/apache/fluss/server/entity/NotifyLeaderAndIsrData.java
b/fluss-server/src/main/java/org/apache/fluss/server/entity/NotifyLeaderAndIsrData.java
index 8ba000208..e6702cdca 100644
---
a/fluss-server/src/main/java/org/apache/fluss/server/entity/NotifyLeaderAndIsrData.java
+++
b/fluss-server/src/main/java/org/apache/fluss/server/entity/NotifyLeaderAndIsrData.java
@@ -85,4 +85,12 @@ public final class NotifyLeaderAndIsrData {
public LeaderAndIsr getLeaderAndIsr() {
return leaderAndIsr;
}
+
+ public List<Integer> getStandbyReplicas() {
+ return leaderAndIsr.standbyReplicas();
+ }
+
+ public int[] getStandbyReplicasArray() {
+ return
leaderAndIsr.standbyReplicas().stream().mapToInt(Integer::intValue).toArray();
+ }
}
diff --git
a/fluss-server/src/main/java/org/apache/fluss/server/replica/IsrState.java
b/fluss-server/src/main/java/org/apache/fluss/server/replica/IsrState.java
index aac5df385..b8b3b0a82 100644
--- a/fluss-server/src/main/java/org/apache/fluss/server/replica/IsrState.java
+++ b/fluss-server/src/main/java/org/apache/fluss/server/replica/IsrState.java
@@ -40,15 +40,25 @@ public interface IsrState {
*/
List<Integer> maximalIsr();
+ /**
+ * Returns the list of standby replicas for this bucket. Standby replicas
are follower replicas
+ * designated to maintain a recent KV snapshot so that they can be quickly
promoted to leader
+ * without a full data rebuild. Currently, at most one standby replica is
supported per bucket,
+ * and it is only applicable to primary-key tables.
+ */
+ List<Integer> standbyReplicas();
+
/** Indicates if we have an AdjustIsr request inflight. */
boolean isInflight();
/** Class to represent the committed isr state of a {@link TableBucket}. */
class CommittedIsrState implements IsrState {
private final List<Integer> isr;
+ private final List<Integer> standbyReplicas;
- public CommittedIsrState(List<Integer> isr) {
+ public CommittedIsrState(List<Integer> isr, List<Integer>
standbyReplicas) {
this.isr = isr;
+ this.standbyReplicas = standbyReplicas;
}
@Override
@@ -56,6 +66,11 @@ public interface IsrState {
return isr;
}
+ @Override
+ public List<Integer> standbyReplicas() {
+ return standbyReplicas;
+ }
+
@Override
public List<Integer> maximalIsr() {
return isr;
@@ -135,6 +150,11 @@ public interface IsrState {
return lastCommittedState.isr();
}
+ @Override
+ public List<Integer> standbyReplicas() {
+ return lastCommittedState.standbyReplicas();
+ }
+
@Override
public List<Integer> maximalIsr() {
ArrayList<Integer> maximalIsr = new
ArrayList<>(lastCommittedState.isr());
@@ -205,6 +225,11 @@ public interface IsrState {
return lastCommittedState.isr();
}
+ @Override
+ public List<Integer> standbyReplicas() {
+ return lastCommittedState.standbyReplicas();
+ }
+
@Override
public List<Integer> maximalIsr() {
return lastCommittedState.isr();
diff --git
a/fluss-server/src/main/java/org/apache/fluss/server/replica/Replica.java
b/fluss-server/src/main/java/org/apache/fluss/server/replica/Replica.java
index 7a4881323..2259ec4c3 100644
--- a/fluss-server/src/main/java/org/apache/fluss/server/replica/Replica.java
+++ b/fluss-server/src/main/java/org/apache/fluss/server/replica/Replica.java
@@ -199,10 +199,12 @@ public final class Replica {
*/
private final Map<Integer, FollowerReplica> followerReplicasMap = new
ConcurrentHashMap<>();
- private volatile IsrState isrState = new
IsrState.CommittedIsrState(Collections.emptyList());
+ private volatile IsrState isrState =
+ new IsrState.CommittedIsrState(Collections.emptyList(),
Collections.emptyList());
private volatile int leaderEpoch = LeaderAndIsr.INITIAL_LEADER_EPOCH - 1;
private volatile int bucketEpoch = LeaderAndIsr.INITIAL_BUCKET_EPOCH;
private volatile int coordinatorEpoch =
CoordinatorContext.INITIAL_COORDINATOR_EPOCH;
+ private volatile boolean isStandbyReplica = false;
// null if table without pk or haven't become leader
private volatile @Nullable KvTablet kvTablet;
@@ -413,7 +415,11 @@ public final class Replica {
long currentTimeMs = clock.milliseconds();
// Updating the assignment and ISR state is safe
if the bucket epoch is
// larger or equal to the current bucket epoch.
- updateAssignmentAndIsr(data.getReplicas(), true,
data.getIsr());
+ updateAssignmentAndIsr(
+ data.getReplicas(),
+ true,
+ data.getIsr(),
+ data.getStandbyReplicas());
int requestLeaderEpoch = data.getLeaderEpoch();
if (requestLeaderEpoch > leaderEpoch) {
@@ -464,7 +470,11 @@ public final class Replica {
coordinatorEpoch = data.getCoordinatorEpoch();
- updateAssignmentAndIsr(Collections.emptyList(), false,
Collections.emptyList());
+ updateAssignmentAndIsr(
+ Collections.emptyList(),
+ false,
+ Collections.emptyList(),
+ Collections.emptyList());
int requestLeaderEpoch = data.getLeaderEpoch();
boolean isNewLeaderEpoch = requestLeaderEpoch >
leaderEpoch;
@@ -474,7 +484,10 @@ public final class Replica {
tableBucket,
requestLeaderEpoch,
logTablet.localLogEndOffset());
- onBecomeNewFollower();
+ // Currently, we only support one standby replica.
+ List<Integer> standbyReplicas =
data.getStandbyReplicas();
+ onBecomeNewFollower(
+ standbyReplicas.isEmpty() ? -1 :
standbyReplicas.get(0));
} else if (requestLeaderEpoch == leaderEpoch) {
LOG.info(
"Skipped the become-follower state change for
bucket {} since "
@@ -534,6 +547,9 @@ public final class Replica {
//
-------------------------------------------------------------------------------------------
private void onBecomeNewLeader() {
+ // Clear standby flag — a leader is never a standby replica.
+ isStandbyReplica = false;
+
updateLeaderEndOffsetSnapshot();
if (isDataLakeEnabled()) {
@@ -575,11 +591,28 @@ public final class Replica {
: logTablet.localMaxTimestamp() -
logTablet.getLakeMaxTimestamp());
}
- private void onBecomeNewFollower() {
+ private void onBecomeNewFollower(int standbyReplica) {
if (isKvTable()) {
+ boolean isNowStandby = (standbyReplica == localTabletServerId);
+ boolean wasLeader = isLeader();
+ boolean wasStandby = this.isStandbyReplica;
+ if (isNowStandby) {
+ // Mark as standby immediately to ensure coordinator's state
is consistent.
+ isStandbyReplica = true;
+ } else {
+ if (wasStandby || wasLeader) {
+ // standby -> follower or leader -> follower
+ if (wasStandby) {
+ isStandbyReplica = false;
+ }
+ }
+ // follower -> follower: do nothing.
+ }
+
// it should be from leader to follower, we need to destroy the kv
tablet
dropKv();
}
+
if (lakeTieringMetricGroup != null) {
lakeTieringMetricGroup.close();
}
@@ -1165,7 +1198,10 @@ public final class Replica {
}
private void updateAssignmentAndIsr(
- List<Integer> replicas, boolean isLeader, List<Integer> isr) {
+ List<Integer> replicas,
+ boolean isLeader,
+ List<Integer> isr,
+ List<Integer> standbyReplicas) {
if (isLeader) {
List<Integer> followers =
replicas.stream()
@@ -1188,7 +1224,7 @@ public final class Replica {
}
// update isr info.
- isrState = new IsrState.CommittedIsrState(isr);
+ isrState = new IsrState.CommittedIsrState(isr, standbyReplicas);
}
private void updateFollowerFetchState(
@@ -1719,9 +1755,19 @@ public final class Replica {
// TODO add server epoch to isr.
+ // TODO: For PK tables, if the current standby list is empty (e.g.,
the previous standby
+ // went down), we should consider assigning a standby from the
expanded ISR members.
+ // This also requires sending a NotifyLeaderAndIsr to the newly
assigned standby replica
+ // so it can start maintaining KV snapshots. Without this, buckets
that lose their standby
+ // will not regain one until the next leader election.
LeaderAndIsr newLeaderAndIsr =
new LeaderAndIsr(
- localTabletServerId, leaderEpoch, isrToSend,
coordinatorEpoch, bucketEpoch);
+ localTabletServerId,
+ leaderEpoch,
+ isrToSend,
+ currentState.standbyReplicas(),
+ coordinatorEpoch,
+ bucketEpoch);
IsrState.PendingExpandIsrState updatedState =
new IsrState.PendingExpandIsrState(
@@ -1743,7 +1789,12 @@ public final class Replica {
LeaderAndIsr newLeaderAndIsr =
new LeaderAndIsr(
- localTabletServerId, leaderEpoch, isrToSend,
coordinatorEpoch, bucketEpoch);
+ localTabletServerId,
+ leaderEpoch,
+ isrToSend,
+ currentState.standbyReplicas(),
+ coordinatorEpoch,
+ bucketEpoch);
IsrState.PendingShrinkIsrState updatedState =
new IsrState.PendingShrinkIsrState(
outOfSyncFollowerReplicas, newLeaderAndIsr,
currentState);
@@ -1838,7 +1889,9 @@ public final class Replica {
// proposed and actual state are the same.
// In both cases, we want to move from Pending to Committed state
to ensure new updates
// are processed.
- isrState = new IsrState.CommittedIsrState(leaderAndIsr.isr());
+ isrState =
+ new IsrState.CommittedIsrState(
+ leaderAndIsr.isr(),
leaderAndIsr.standbyReplicas());
bucketEpoch = leaderAndIsr.bucketEpoch();
LOG.info(
"ISR updated to {} and bucket epoch updated to {} for
bucket {}",
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 d37c36148..f7a89828a 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
@@ -684,7 +684,8 @@ public class ServerRpcMessageUtils {
new PbNotifyLeaderAndIsrReqForBucket()
.setLeader(notifyLeaderAndIsrData.getLeader())
.setLeaderEpoch(notifyLeaderAndIsrData.getLeaderEpoch())
-
.setBucketEpoch(notifyLeaderAndIsrData.getBucketEpoch());
+
.setBucketEpoch(notifyLeaderAndIsrData.getBucketEpoch())
+
.setStandbyReplicas(notifyLeaderAndIsrData.getStandbyReplicasArray());
TableBucket tb = notifyLeaderAndIsrData.getTableBucket();
PbTableBucket pbTableBucket =
@@ -720,6 +721,11 @@ public class ServerRpcMessageUtils {
isr.add(reqForBucket.getIsrAt(i));
}
+ List<Integer> standbyReplicas = new ArrayList<>();
+ for (int i = 0; i < reqForBucket.getStandbyReplicasCount(); i++) {
+ standbyReplicas.add(reqForBucket.getStandbyReplicaAt(i));
+ }
+
PbTableBucket pbTableBucket = reqForBucket.getTableBucket();
notifyLeaderAndIsrDataList.add(
new NotifyLeaderAndIsrData(
@@ -730,6 +736,7 @@ public class ServerRpcMessageUtils {
reqForBucket.getLeader(),
reqForBucket.getLeaderEpoch(),
isr,
+ standbyReplicas,
request.getCoordinatorEpoch(),
reqForBucket.getBucketEpoch())));
}
@@ -1295,6 +1302,7 @@ public class ServerRpcMessageUtils {
reqForBucket.setPartitionId(tb.getPartitionId());
}
leaderAndIsr.isr().forEach(reqForBucket::addNewIsr);
+
leaderAndIsr.standbyReplicas().forEach(reqForBucket::addStandbyReplica);
if (reqForBucketByTableId.containsKey(tb.getTableId())) {
reqForBucketByTableId.get(tb.getTableId()).add(reqForBucket);
} else {
@@ -1336,12 +1344,18 @@ public class ServerRpcMessageUtils {
for (int i = 0; i < reqForBucket.getNewIsrsCount(); i++) {
newIsr.add(reqForBucket.getNewIsrAt(i));
}
+ List<Integer> standbyReplicas = new ArrayList<>();
+ for (int i = 0; i < reqForBucket.getStandbyReplicasCount();
i++) {
+ standbyReplicas.add(reqForBucket.getStandbyReplicaAt(i));
+ }
+
leaderAndIsrMap.put(
tb,
new LeaderAndIsr(
leaderId,
reqForBucket.getLeaderEpoch(),
newIsr,
+ standbyReplicas,
reqForBucket.getCoordinatorEpoch(),
reqForBucket.getBucketEpoch()));
}
@@ -1368,7 +1382,8 @@ public class ServerRpcMessageUtils {
.setLeaderEpoch(leaderAndIsr.leaderEpoch())
.setCoordinatorEpoch(leaderAndIsr.coordinatorEpoch())
.setBucketEpoch(leaderAndIsr.bucketEpoch())
- .setIsrs(leaderAndIsr.isrArray());
+ .setIsrs(leaderAndIsr.isrArray())
+
.setStandbyReplicas(leaderAndIsr.standbyReplicasArray());
}
if (respMap.containsKey(tb.getTableId())) {
@@ -1417,6 +1432,10 @@ public class ServerRpcMessageUtils {
for (int i = 0; i < respForBucket.getIsrsCount(); i++) {
isr.add(respForBucket.getIsrAt(i));
}
+ List<Integer> standbyReplicas = new ArrayList<>();
+ for (int i = 0; i < respForBucket.getStandbyReplicasCount();
i++) {
+ standbyReplicas.add(respForBucket.getStandbyReplicaAt(i));
+ }
adjustIsrResult.put(
tb,
new AdjustIsrResultForBucket(
@@ -1425,6 +1444,7 @@ public class ServerRpcMessageUtils {
respForBucket.getLeaderId(),
respForBucket.getLeaderEpoch(),
isr,
+ standbyReplicas,
respForBucket.getCoordinatorEpoch(),
respForBucket.getBucketEpoch())));
}
diff --git
a/fluss-server/src/main/java/org/apache/fluss/server/zk/data/LeaderAndIsr.java
b/fluss-server/src/main/java/org/apache/fluss/server/zk/data/LeaderAndIsr.java
index 251248138..020601716 100644
---
a/fluss-server/src/main/java/org/apache/fluss/server/zk/data/LeaderAndIsr.java
+++
b/fluss-server/src/main/java/org/apache/fluss/server/zk/data/LeaderAndIsr.java
@@ -43,6 +43,8 @@ public class LeaderAndIsr {
/** The latest inSyncReplica collection. */
private final List<Integer> isr;
+ private final List<Integer> standbyReplicas;
+
/** The coordinator epoch. */
private final int coordinatorEpoch;
@@ -57,15 +59,22 @@ public class LeaderAndIsr {
leader,
INITIAL_LEADER_EPOCH,
new ArrayList<>(),
+ new ArrayList<>(),
coordinatorEpoch,
INITIAL_BUCKET_EPOCH);
}
public LeaderAndIsr(
- int leader, int leaderEpoch, List<Integer> isr, int
coordinatorEpoch, int bucketEpoch) {
+ int leader,
+ int leaderEpoch,
+ List<Integer> isr,
+ List<Integer> standbyReplicas,
+ int coordinatorEpoch,
+ int bucketEpoch) {
this.leader = leader;
this.leaderEpoch = leaderEpoch;
this.isr = checkNotNull(isr);
+ this.standbyReplicas = checkNotNull(standbyReplicas);
this.coordinatorEpoch = coordinatorEpoch;
this.bucketEpoch = bucketEpoch;
}
@@ -77,9 +86,15 @@ public class LeaderAndIsr {
* @param newIsr the new isr
* @return the new LeaderAndIsr
*/
- public LeaderAndIsr newLeaderAndIsr(int newLeader, List<Integer> newIsr) {
+ public LeaderAndIsr newLeaderAndIsr(
+ int newLeader, List<Integer> newIsr, List<Integer>
standbyReplicas) {
return new LeaderAndIsr(
- newLeader, leaderEpoch + 1, newIsr, coordinatorEpoch,
bucketEpoch + 1);
+ newLeader,
+ leaderEpoch + 1,
+ newIsr,
+ standbyReplicas,
+ coordinatorEpoch,
+ bucketEpoch + 1);
}
/**
@@ -90,7 +105,8 @@ public class LeaderAndIsr {
* @return the new LeaderAndIsr
*/
public LeaderAndIsr newLeaderAndIsr(List<Integer> newIsr) {
- return new LeaderAndIsr(leader, leaderEpoch, newIsr, coordinatorEpoch,
bucketEpoch + 1);
+ return new LeaderAndIsr(
+ leader, leaderEpoch, newIsr, standbyReplicas,
coordinatorEpoch, bucketEpoch + 1);
}
public int leader() {
@@ -109,10 +125,18 @@ public class LeaderAndIsr {
return isr;
}
+ public List<Integer> standbyReplicas() {
+ return standbyReplicas;
+ }
+
public int[] isrArray() {
return isr.stream().mapToInt(Integer::intValue).toArray();
}
+ public int[] standbyReplicasArray() {
+ return standbyReplicas.stream().mapToInt(Integer::intValue).toArray();
+ }
+
public int bucketEpoch() {
return bucketEpoch;
}
@@ -130,12 +154,14 @@ public class LeaderAndIsr {
&& leaderEpoch == that.leaderEpoch
&& coordinatorEpoch == that.coordinatorEpoch
&& bucketEpoch == that.bucketEpoch
- && Objects.equals(isr, that.isr);
+ && Objects.equals(isr, that.isr)
+ && Objects.equals(standbyReplicas, that.standbyReplicas);
}
@Override
public int hashCode() {
- return Objects.hash(leader, leaderEpoch, isr, coordinatorEpoch,
bucketEpoch);
+ return Objects.hash(
+ leader, leaderEpoch, isr, standbyReplicas, coordinatorEpoch,
bucketEpoch);
}
@Override
@@ -147,6 +173,8 @@ public class LeaderAndIsr {
+ leaderEpoch
+ ", isr="
+ isr
+ + ", standbyReplicas="
+ + standbyReplicas
+ ", coordinatorEpoch="
+ coordinatorEpoch
+ ", bucketEpoch="
diff --git
a/fluss-server/src/main/java/org/apache/fluss/server/zk/data/LeaderAndIsrJsonSerde.java
b/fluss-server/src/main/java/org/apache/fluss/server/zk/data/LeaderAndIsrJsonSerde.java
index 9d76330f2..65c69d1c4 100644
---
a/fluss-server/src/main/java/org/apache/fluss/server/zk/data/LeaderAndIsrJsonSerde.java
+++
b/fluss-server/src/main/java/org/apache/fluss/server/zk/data/LeaderAndIsrJsonSerde.java
@@ -35,18 +35,19 @@ public class LeaderAndIsrJsonSerde
public static final LeaderAndIsrJsonSerde INSTANCE = new
LeaderAndIsrJsonSerde();
private static final String VERSION_KEY = "version";
- private static final int VERSION = 1;
+ private static final int CURRENT_VERSION = 2;
private static final String LEADER = "leader";
private static final String LEADER_EPOCH = "leader_epoch";
private static final String ISR = "isr";
private static final String COORDINATOR_EPOCH = "coordinator_epoch";
private static final String BUCKET_EPOCH = "bucket_epoch";
+ private static final String STANDBY_REPLICA = "standby_replicas";
@Override
public void serialize(LeaderAndIsr leaderAndIsr, JsonGenerator generator)
throws IOException {
generator.writeStartObject();
- generator.writeNumberField(VERSION_KEY, VERSION);
+ generator.writeNumberField(VERSION_KEY, CURRENT_VERSION);
generator.writeNumberField(LEADER, leaderAndIsr.leader());
generator.writeNumberField(LEADER_EPOCH, leaderAndIsr.leaderEpoch());
generator.writeArrayFieldStart(ISR);
@@ -57,11 +58,19 @@ public class LeaderAndIsrJsonSerde
generator.writeNumberField(COORDINATOR_EPOCH,
leaderAndIsr.coordinatorEpoch());
generator.writeNumberField(BUCKET_EPOCH, leaderAndIsr.bucketEpoch());
+ // write standby_replicas
+ generator.writeArrayFieldStart(STANDBY_REPLICA);
+ for (Integer replica : leaderAndIsr.standbyReplicas()) {
+ generator.writeNumber(replica);
+ }
+ generator.writeEndArray();
+
generator.writeEndObject();
}
@Override
public LeaderAndIsr deserialize(JsonNode node) {
+ int version = node.get(VERSION_KEY).asInt();
int leader = node.get(LEADER).asInt();
int leaderEpoch = node.get(LEADER_EPOCH).asInt();
int coordinatorEpoch = node.get(COORDINATOR_EPOCH).asInt();
@@ -71,6 +80,16 @@ public class LeaderAndIsrJsonSerde
while (isrNodes.hasNext()) {
isr.add(isrNodes.next().asInt());
}
- return new LeaderAndIsr(leader, leaderEpoch, isr, coordinatorEpoch,
bucketEpoch);
+
+ List<Integer> standbyList = new ArrayList<>();
+ if (version >= CURRENT_VERSION) {
+ Iterator<JsonNode> hotStandbyListNodes =
node.get(STANDBY_REPLICA).elements();
+ while (hotStandbyListNodes.hasNext()) {
+ standbyList.add(hotStandbyListNodes.next().asInt());
+ }
+ }
+
+ return new LeaderAndIsr(
+ leader, leaderEpoch, isr, standbyList, coordinatorEpoch,
bucketEpoch);
}
}
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 e4f60a938..0694e9787 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
@@ -777,6 +777,7 @@ class CoordinatorEventProcessorTest {
leader,
leaderAndIsr.leaderEpoch(),
newIsr,
+ Collections.emptyList(),
coordinatorEpoch,
bucketLeaderEpoch));
return null;
@@ -1035,7 +1036,9 @@ class CoordinatorEventProcessorTest {
CompletableFuture<AdjustIsrResponse> respCallback = new
CompletableFuture<>();
// This isr list equals originReplicas + addingReplicas. the bucket
epoch is 1.
- leaderAndIsrMap.put(tb0, new LeaderAndIsr(0, 0, Arrays.asList(0, 1, 2,
3), 0, 1));
+ leaderAndIsrMap.put(
+ tb0,
+ new LeaderAndIsr(0, 0, Arrays.asList(0, 1, 2, 3),
Collections.emptyList(), 0, 1));
eventProcessor
.getCoordinatorEventManager()
.put(new AdjustIsrReceivedEvent(leaderAndIsrMap,
respCallback));
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 68e46e359..23b8a70bf 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
@@ -56,6 +56,7 @@ import org.junit.jupiter.api.io.TempDir;
import java.nio.file.Path;
import java.time.Duration;
import java.util.Arrays;
+import java.util.Collections;
import java.util.List;
import java.util.Objects;
import java.util.concurrent.CompletionException;
@@ -332,7 +333,8 @@ class CoordinatorHighAvailabilityITCase {
.isEqualTo(standby.getServerId());
TableBucket tableBucket = new TableBucket(1, 1);
- LeaderAndIsr leaderAndIsr = new LeaderAndIsr(0, 0, Arrays.asList(2,
3), 0, 0);
+ LeaderAndIsr leaderAndIsr =
+ new LeaderAndIsr(0, 0, Arrays.asList(2, 3),
Collections.emptyList(), 0, 0);
int newLeaderEpochZkVersion =
standby.getCoordinatorEventProcessor()
diff --git
a/fluss-server/src/test/java/org/apache/fluss/server/coordinator/TestCoordinatorGateway.java
b/fluss-server/src/test/java/org/apache/fluss/server/coordinator/TestCoordinatorGateway.java
index f55846a7f..98e295f9a 100644
---
a/fluss-server/src/test/java/org/apache/fluss/server/coordinator/TestCoordinatorGateway.java
+++
b/fluss-server/src/test/java/org/apache/fluss/server/coordinator/TestCoordinatorGateway.java
@@ -324,6 +324,7 @@ public class TestCoordinatorGateway implements
CoordinatorGateway {
leaderAndIsr.leader(),
currentLeaderEpoch,
leaderAndIsr.isr(),
+ leaderAndIsr.standbyReplicas(),
leaderAndIsr.coordinatorEpoch(),
leaderAndIsr.bucketEpoch() +
1));
}
diff --git
a/fluss-server/src/test/java/org/apache/fluss/server/coordinator/statemachine/ReplicaLeaderElectionTest.java
b/fluss-server/src/test/java/org/apache/fluss/server/coordinator/statemachine/ReplicaLeaderElectionTest.java
index d1dc77d61..313114d69 100644
---
a/fluss-server/src/test/java/org/apache/fluss/server/coordinator/statemachine/ReplicaLeaderElectionTest.java
+++
b/fluss-server/src/test/java/org/apache/fluss/server/coordinator/statemachine/ReplicaLeaderElectionTest.java
@@ -41,30 +41,90 @@ public class ReplicaLeaderElectionTest {
void testDefaultReplicaLeaderElection() {
List<Integer> assignments = Arrays.asList(2, 4);
List<Integer> liveReplicas = Arrays.asList(2, 4);
- LeaderAndIsr originLeaderAndIsr = new LeaderAndIsr(4, 0,
Arrays.asList(2, 4), 0, 0);
+ LeaderAndIsr originLeaderAndIsr =
+ new LeaderAndIsr(4, 0, Arrays.asList(2, 4),
Collections.emptyList(), 0, 0);
DefaultLeaderElection defaultLeaderElection = new
DefaultLeaderElection();
Optional<ElectionResult> leaderElectionResultOpt =
- defaultLeaderElection.leaderElection(assignments,
liveReplicas, originLeaderAndIsr);
+ defaultLeaderElection.leaderElection(
+ assignments, liveReplicas, originLeaderAndIsr, false);
assertThat(leaderElectionResultOpt.isPresent()).isTrue();
ElectionResult leaderElectionResult = leaderElectionResultOpt.get();
assertThat(leaderElectionResult.getLiveReplicas()).containsExactlyInAnyOrder(2,
4);
assertThat(leaderElectionResult.getLeaderAndIsr().leader()).isEqualTo(2);
assertThat(leaderElectionResult.getLeaderAndIsr().isr()).containsExactlyInAnyOrder(2,
4);
+
assertThat(leaderElectionResult.getLeaderAndIsr().standbyReplicas()).isEmpty();
+ }
+
+ @Test
+ void testDefaultReplicaLeaderElectionForPkTable() {
+ List<Integer> assignments = Arrays.asList(2, 3, 4);
+ List<Integer> liveReplicas = Arrays.asList(3, 4);
+ LeaderAndIsr originLeaderAndIsr =
+ new LeaderAndIsr(2, 0, Arrays.asList(2, 3, 4),
Collections.emptyList(), 0, 0);
+
+ // first, test origin leaderAndIsr don't have standby replica.
+ DefaultLeaderElection defaultLeaderElection = new
DefaultLeaderElection();
+ Optional<ElectionResult> leaderElectionResultOpt =
+ defaultLeaderElection.leaderElection(
+ assignments, liveReplicas, originLeaderAndIsr, true);
+ assertThat(leaderElectionResultOpt.isPresent()).isTrue();
+ ElectionResult leaderElectionResult = leaderElectionResultOpt.get();
+
assertThat(leaderElectionResult.getLiveReplicas()).containsExactlyInAnyOrder(3,
4);
+
assertThat(leaderElectionResult.getLeaderAndIsr().leader()).isEqualTo(3);
+
assertThat(leaderElectionResult.getLeaderAndIsr().isr()).containsExactlyInAnyOrder(2,
3, 4);
+ assertThat(leaderElectionResult.getLeaderAndIsr().standbyReplicas())
+ .containsExactlyInAnyOrder(4);
+
+ // second. test origin leaderAndIsr has standby replica.
+ originLeaderAndIsr =
+ new LeaderAndIsr(2, 0, Arrays.asList(2, 3, 4),
Collections.singletonList(4), 0, 0);
+ defaultLeaderElection = new DefaultLeaderElection();
+ leaderElectionResultOpt =
+ defaultLeaderElection.leaderElection(
+ assignments, liveReplicas, originLeaderAndIsr, true);
+ assertThat(leaderElectionResultOpt.isPresent()).isTrue();
+ leaderElectionResult = leaderElectionResultOpt.get();
+
assertThat(leaderElectionResult.getLiveReplicas()).containsExactlyInAnyOrder(3,
4);
+
assertThat(leaderElectionResult.getLeaderAndIsr().leader()).isEqualTo(4);
+
assertThat(leaderElectionResult.getLeaderAndIsr().isr()).containsExactlyInAnyOrder(2,
3, 4);
+ assertThat(leaderElectionResult.getLeaderAndIsr().standbyReplicas())
+ .containsExactlyInAnyOrder(3);
+
+ // third. test no enough live replicas.
+ assignments = Arrays.asList(2, 3, 4);
+ liveReplicas = Collections.singletonList(4);
+ originLeaderAndIsr =
+ new LeaderAndIsr(2, 0, Arrays.asList(2, 3, 4),
Collections.emptyList(), 0, 0);
+ defaultLeaderElection = new DefaultLeaderElection();
+ leaderElectionResultOpt =
+ defaultLeaderElection.leaderElection(
+ assignments, liveReplicas, originLeaderAndIsr, true);
+ assertThat(leaderElectionResultOpt.isPresent()).isTrue();
+ leaderElectionResult = leaderElectionResultOpt.get();
+
assertThat(leaderElectionResult.getLiveReplicas()).containsExactlyInAnyOrder(4);
+
assertThat(leaderElectionResult.getLeaderAndIsr().leader()).isEqualTo(4);
+
assertThat(leaderElectionResult.getLeaderAndIsr().isr()).containsExactlyInAnyOrder(2,
3, 4);
+
assertThat(leaderElectionResult.getLeaderAndIsr().standbyReplicas()).isEmpty();
}
@Test
void testControlledShutdownReplicaLeaderElection() {
List<Integer> assignments = Arrays.asList(2, 4);
List<Integer> liveReplicas = Arrays.asList(2, 4);
- LeaderAndIsr originLeaderAndIsr = new LeaderAndIsr(2, 0,
Arrays.asList(2, 4), 0, 0);
+ LeaderAndIsr originLeaderAndIsr =
+ new LeaderAndIsr(2, 0, Arrays.asList(2, 4),
Collections.emptyList(), 0, 0);
Set<Integer> shutdownTabletServers = Collections.singleton(2);
ControlledShutdownLeaderElection controlledShutdownLeaderElection =
new ControlledShutdownLeaderElection();
Optional<ElectionResult> leaderElectionResultOpt =
controlledShutdownLeaderElection.leaderElection(
- assignments, liveReplicas, originLeaderAndIsr,
shutdownTabletServers);
+ assignments,
+ liveReplicas,
+ originLeaderAndIsr,
+ shutdownTabletServers,
+ false);
assertThat(leaderElectionResultOpt.isPresent()).isTrue();
ElectionResult leaderElectionResult = leaderElectionResultOpt.get();
assertThat(leaderElectionResult.getLiveReplicas()).containsExactlyInAnyOrder(4);
@@ -77,57 +137,317 @@ public class ReplicaLeaderElectionTest {
List<Integer> assignments = Arrays.asList(2, 4);
List<Integer> liveReplicas = Arrays.asList(2, 4);
LeaderAndIsr originLeaderAndIsr =
- new LeaderAndIsr(2, 0, Collections.singletonList(2), 0, 0);
+ new LeaderAndIsr(2, 0, Collections.singletonList(2),
Collections.emptyList(), 0, 0);
Set<Integer> shutdownTabletServers = Collections.singleton(2);
ControlledShutdownLeaderElection controlledShutdownLeaderElection =
new ControlledShutdownLeaderElection();
Optional<ElectionResult> leaderElectionResultOpt =
controlledShutdownLeaderElection.leaderElection(
- assignments, liveReplicas, originLeaderAndIsr,
shutdownTabletServers);
+ assignments,
+ liveReplicas,
+ originLeaderAndIsr,
+ shutdownTabletServers,
+ false);
assertThat(leaderElectionResultOpt).isEmpty();
}
@Test
- void
testControlledShutdownPartitionLeaderElectionAllIsrSimultaneouslyShutdown() {
+ void testControlledShutdownLeaderElectionAllIsrSimultaneouslyShutdown() {
List<Integer> assignments = Arrays.asList(2, 4);
List<Integer> liveReplicas = Arrays.asList(2, 4);
- LeaderAndIsr originLeaderAndIsr = new LeaderAndIsr(2, 0,
Arrays.asList(2, 4), 0, 0);
+ LeaderAndIsr originLeaderAndIsr =
+ new LeaderAndIsr(2, 0, Arrays.asList(2, 4),
Collections.emptyList(), 0, 0);
Set<Integer> shutdownTabletServers = new HashSet<>(Arrays.asList(2,
4));
ControlledShutdownLeaderElection controlledShutdownLeaderElection =
new ControlledShutdownLeaderElection();
Optional<ElectionResult> leaderElectionResultOpt =
controlledShutdownLeaderElection.leaderElection(
- assignments, liveReplicas, originLeaderAndIsr,
shutdownTabletServers);
+ assignments,
+ liveReplicas,
+ originLeaderAndIsr,
+ shutdownTabletServers,
+ false);
assertThat(leaderElectionResultOpt).isEmpty();
}
+ @Test
+ void testControlledShutdownReplicaLeaderElectionForPkTable() {
+ List<Integer> assignments = Arrays.asList(2, 3, 4);
+ List<Integer> liveReplicas = Arrays.asList(2, 3, 4);
+ LeaderAndIsr originLeaderAndIsr =
+ new LeaderAndIsr(2, 0, Arrays.asList(2, 3, 4),
Collections.emptyList(), 0, 0);
+ Set<Integer> shutdownTabletServers = Collections.singleton(2);
+ ControlledShutdownLeaderElection controlledShutdownLeaderElection =
+ new ControlledShutdownLeaderElection();
+ Optional<ElectionResult> leaderElectionResultOpt =
+ controlledShutdownLeaderElection.leaderElection(
+ assignments, liveReplicas, originLeaderAndIsr,
shutdownTabletServers, true);
+ assertThat(leaderElectionResultOpt.isPresent()).isTrue();
+ ElectionResult leaderElectionResult = leaderElectionResultOpt.get();
+
assertThat(leaderElectionResult.getLiveReplicas()).containsExactlyInAnyOrder(3,
4);
+
assertThat(leaderElectionResult.getLeaderAndIsr().leader()).isEqualTo(3);
+
assertThat(leaderElectionResult.getLeaderAndIsr().isr()).containsExactlyInAnyOrder(3,
4);
+ assertThat(leaderElectionResult.getLeaderAndIsr().standbyReplicas())
+ .containsExactlyInAnyOrder(4);
+
+ originLeaderAndIsr =
+ new LeaderAndIsr(2, 0, Arrays.asList(2, 3, 4),
Collections.singletonList(4), 0, 0);
+ controlledShutdownLeaderElection = new
ControlledShutdownLeaderElection();
+ leaderElectionResultOpt =
+ controlledShutdownLeaderElection.leaderElection(
+ assignments, liveReplicas, originLeaderAndIsr,
shutdownTabletServers, true);
+ assertThat(leaderElectionResultOpt.isPresent()).isTrue();
+ leaderElectionResult = leaderElectionResultOpt.get();
+
assertThat(leaderElectionResult.getLiveReplicas()).containsExactlyInAnyOrder(3,
4);
+
assertThat(leaderElectionResult.getLeaderAndIsr().leader()).isEqualTo(4);
+
assertThat(leaderElectionResult.getLeaderAndIsr().isr()).containsExactlyInAnyOrder(3,
4);
+ assertThat(leaderElectionResult.getLeaderAndIsr().standbyReplicas())
+ .containsExactlyInAnyOrder(3);
+ }
+
@Test
void testReassignBucketLeaderElection() {
List<Integer> targetReplicas = Arrays.asList(1, 2, 3);
ReassignmentLeaderElection reassignmentLeaderElection =
new ReassignmentLeaderElection(targetReplicas);
List<Integer> liveReplicas = Arrays.asList(1, 2, 3);
- LeaderAndIsr leaderAndIsr = new LeaderAndIsr(1, 0, Arrays.asList(1, 2,
3), 0, 0);
+ LeaderAndIsr leaderAndIsr =
+ new LeaderAndIsr(1, 0, Arrays.asList(1, 2, 3),
Collections.emptyList(), 0, 0);
Optional<ElectionResult> leaderOpt =
- reassignmentLeaderElection.leaderElection(liveReplicas,
leaderAndIsr);
+ reassignmentLeaderElection.leaderElection(liveReplicas,
leaderAndIsr, false);
assertThat(leaderOpt).isPresent();
assertThat(leaderOpt.get().getLeaderAndIsr().leader()).isEqualTo(1);
targetReplicas = Arrays.asList(1, 2, 3);
reassignmentLeaderElection = new
ReassignmentLeaderElection(targetReplicas);
liveReplicas = Arrays.asList(2, 3);
- leaderAndIsr = new LeaderAndIsr(1, 0, Arrays.asList(2, 3), 0, 0);
- leaderOpt = reassignmentLeaderElection.leaderElection(liveReplicas,
leaderAndIsr);
+ leaderAndIsr = new LeaderAndIsr(1, 0, Arrays.asList(2, 3),
Collections.emptyList(), 0, 0);
+ leaderOpt = reassignmentLeaderElection.leaderElection(liveReplicas,
leaderAndIsr, false);
assertThat(leaderOpt).isPresent();
assertThat(leaderOpt.get().getLeaderAndIsr().leader()).isEqualTo(2);
targetReplicas = Arrays.asList(1, 2, 3);
reassignmentLeaderElection = new
ReassignmentLeaderElection(targetReplicas);
liveReplicas = Arrays.asList(1, 2);
- leaderAndIsr = new LeaderAndIsr(2, 1, Collections.emptyList(), 0, 1);
- leaderOpt = reassignmentLeaderElection.leaderElection(liveReplicas,
leaderAndIsr);
+ leaderAndIsr =
+ new LeaderAndIsr(2, 1, Collections.emptyList(),
Collections.emptyList(), 0, 1);
+ leaderOpt = reassignmentLeaderElection.leaderElection(liveReplicas,
leaderAndIsr, false);
assertThat(leaderOpt).isNotPresent();
}
+
+ @Test
+ void testDefaultLeaderElectionWithSingleReplica() {
+ // Test that single replica election doesn't throw
IndexOutOfBoundsException
+ List<Integer> assignments = Collections.singletonList(1);
+ List<Integer> liveReplicas = Collections.singletonList(1);
+ LeaderAndIsr originLeaderAndIsr =
+ new LeaderAndIsr(1, 0, Collections.singletonList(1),
Collections.emptyList(), 0, 0);
+
+ DefaultLeaderElection defaultLeaderElection = new
DefaultLeaderElection();
+ Optional<ElectionResult> leaderElectionResultOpt =
+ defaultLeaderElection.leaderElection(
+ assignments, liveReplicas, originLeaderAndIsr, true);
+ assertThat(leaderElectionResultOpt.isPresent()).isTrue();
+ ElectionResult leaderElectionResult = leaderElectionResultOpt.get();
+
assertThat(leaderElectionResult.getLeaderAndIsr().leader()).isEqualTo(1);
+ // With only one replica, standby should be empty
+
assertThat(leaderElectionResult.getLeaderAndIsr().standbyReplicas()).isEmpty();
+ }
+
+ @Test
+ void testControlledShutdownWithSingleRemainingReplica() {
+ // Test controlled shutdown when only one replica remains available
+ List<Integer> assignments = Arrays.asList(1, 2);
+ List<Integer> liveReplicas = Arrays.asList(1, 2);
+ LeaderAndIsr originLeaderAndIsr =
+ new LeaderAndIsr(1, 0, Arrays.asList(1, 2),
Collections.singletonList(2), 0, 0);
+ Set<Integer> shutdownTabletServers = Collections.singleton(1);
+
+ ControlledShutdownLeaderElection controlledShutdownLeaderElection =
+ new ControlledShutdownLeaderElection();
+ Optional<ElectionResult> leaderElectionResultOpt =
+ controlledShutdownLeaderElection.leaderElection(
+ assignments, liveReplicas, originLeaderAndIsr,
shutdownTabletServers, true);
+ assertThat(leaderElectionResultOpt.isPresent()).isTrue();
+ ElectionResult leaderElectionResult = leaderElectionResultOpt.get();
+ // Standby (2) should be promoted to leader
+
assertThat(leaderElectionResult.getLeaderAndIsr().leader()).isEqualTo(2);
+ // With only one remaining replica, standby should be empty
+
assertThat(leaderElectionResult.getLeaderAndIsr().standbyReplicas()).isEmpty();
+ }
+
+ @Test
+ void testReassignBucketLeaderElectionForPkTable() {
+ // Test reassignment election for PK table
+ List<Integer> targetReplicas = Arrays.asList(1, 2, 3);
+ ReassignmentLeaderElection reassignmentLeaderElection =
+ new ReassignmentLeaderElection(targetReplicas);
+ List<Integer> liveReplicas = Arrays.asList(1, 2, 3);
+ LeaderAndIsr leaderAndIsr =
+ new LeaderAndIsr(2, 0, Arrays.asList(1, 2, 3),
Collections.singletonList(3), 0, 0);
+ Optional<ElectionResult> leaderOpt =
+ reassignmentLeaderElection.leaderElection(liveReplicas,
leaderAndIsr, true);
+ assertThat(leaderOpt).isPresent();
+ // First available replica (1) should be leader to honor the rebalance
plan
+ assertThat(leaderOpt.get().getLeaderAndIsr().leader()).isEqualTo(1);
+ // A new standby should be assigned from remaining available replicas
+ assertThat(leaderOpt.get().getLeaderAndIsr().standbyReplicas())
+ .containsExactlyInAnyOrder(2);
+ }
+
+ @Test
+ void testReassignBucketLeaderElectionForPkTableWithSingleReplica() {
+ // Test reassignment election for PK table with only one replica
+ List<Integer> targetReplicas = Collections.singletonList(1);
+ ReassignmentLeaderElection reassignmentLeaderElection =
+ new ReassignmentLeaderElection(targetReplicas);
+ List<Integer> liveReplicas = Collections.singletonList(1);
+ LeaderAndIsr leaderAndIsr =
+ new LeaderAndIsr(1, 0, Collections.singletonList(1),
Collections.emptyList(), 0, 0);
+ Optional<ElectionResult> leaderOpt =
+ reassignmentLeaderElection.leaderElection(liveReplicas,
leaderAndIsr, true);
+ assertThat(leaderOpt).isPresent();
+ assertThat(leaderOpt.get().getLeaderAndIsr().leader()).isEqualTo(1);
+ // With only one replica, standby should be empty
+
assertThat(leaderOpt.get().getLeaderAndIsr().standbyReplicas()).isEmpty();
+ }
+
+ @Test
+ void testDefaultLeaderElectionWithStandbyUnavailable() {
+ // Test when current standby is not in available replicas
+ List<Integer> assignments = Arrays.asList(1, 2, 3);
+ List<Integer> liveReplicas = Arrays.asList(1, 2); // standby (3) is
not alive
+ LeaderAndIsr originLeaderAndIsr =
+ new LeaderAndIsr(1, 0, Arrays.asList(1, 2, 3),
Collections.singletonList(3), 0, 0);
+
+ DefaultLeaderElection defaultLeaderElection = new
DefaultLeaderElection();
+ Optional<ElectionResult> leaderElectionResultOpt =
+ defaultLeaderElection.leaderElection(
+ assignments, liveReplicas, originLeaderAndIsr, true);
+ assertThat(leaderElectionResultOpt.isPresent()).isTrue();
+ ElectionResult leaderElectionResult = leaderElectionResultOpt.get();
+ // Should select first available replica as leader since standby is
unavailable
+
assertThat(leaderElectionResult.getLeaderAndIsr().leader()).isEqualTo(1);
+ // New standby should be selected from remaining available replicas
+ assertThat(leaderElectionResult.getLeaderAndIsr().standbyReplicas())
+ .containsExactlyInAnyOrder(2);
+ }
+
+ @Test
+ void testDefaultLeaderElectionWithNoAvailableReplicas() {
+ // Test when no replicas are available
+ List<Integer> assignments = Arrays.asList(1, 2, 3);
+ List<Integer> liveReplicas = Collections.emptyList();
+ LeaderAndIsr originLeaderAndIsr =
+ new LeaderAndIsr(1, 0, Arrays.asList(1, 2, 3),
Collections.emptyList(), 0, 0);
+
+ DefaultLeaderElection defaultLeaderElection = new
DefaultLeaderElection();
+ Optional<ElectionResult> leaderElectionResultOpt =
+ defaultLeaderElection.leaderElection(
+ assignments, liveReplicas, originLeaderAndIsr, true);
+ assertThat(leaderElectionResultOpt).isEmpty();
+ }
+
+ @Test
+ void testControlledShutdownForPkTableWithStandbyShuttingDown() {
+ // Test when the current standby is shutting down
+ List<Integer> assignments = Arrays.asList(1, 2, 3);
+ List<Integer> liveReplicas = Arrays.asList(1, 2, 3);
+ LeaderAndIsr originLeaderAndIsr =
+ new LeaderAndIsr(1, 0, Arrays.asList(1, 2, 3),
Collections.singletonList(2), 0, 0);
+ Set<Integer> shutdownTabletServers = Collections.singleton(2); //
standby is shutting down
+
+ ControlledShutdownLeaderElection controlledShutdownLeaderElection =
+ new ControlledShutdownLeaderElection();
+ Optional<ElectionResult> leaderElectionResultOpt =
+ controlledShutdownLeaderElection.leaderElection(
+ assignments, liveReplicas, originLeaderAndIsr,
shutdownTabletServers, true);
+ assertThat(leaderElectionResultOpt.isPresent()).isTrue();
+ ElectionResult leaderElectionResult = leaderElectionResultOpt.get();
+ // Leader (1) should remain leader since standby (2) is shutting down
+
assertThat(leaderElectionResult.getLeaderAndIsr().leader()).isEqualTo(1);
+ // New standby should be selected from remaining replicas
+ assertThat(leaderElectionResult.getLeaderAndIsr().standbyReplicas())
+ .containsExactlyInAnyOrder(3);
+ }
+
+ @Test
+ void
testControlledShutdownForPkTableWithLeaderAndStandbyBothShuttingDown() {
+ // Test when both leader and standby are shutting down simultaneously
+ List<Integer> assignments = Arrays.asList(1, 2, 3);
+ List<Integer> liveReplicas = Arrays.asList(1, 2, 3);
+ LeaderAndIsr originLeaderAndIsr =
+ new LeaderAndIsr(1, 0, Arrays.asList(1, 2, 3),
Collections.singletonList(2), 0, 0);
+ // Both leader (1) and standby (2) are shutting down
+ Set<Integer> shutdownTabletServers = new HashSet<>(Arrays.asList(1,
2));
+
+ ControlledShutdownLeaderElection controlledShutdownLeaderElection =
+ new ControlledShutdownLeaderElection();
+ Optional<ElectionResult> leaderElectionResultOpt =
+ controlledShutdownLeaderElection.leaderElection(
+ assignments, liveReplicas, originLeaderAndIsr,
shutdownTabletServers, true);
+ assertThat(leaderElectionResultOpt.isPresent()).isTrue();
+ ElectionResult leaderElectionResult = leaderElectionResultOpt.get();
+ // Replica 3 should become leader since both leader and standby are
shutting down
+
assertThat(leaderElectionResult.getLeaderAndIsr().leader()).isEqualTo(3);
+
assertThat(leaderElectionResult.getLiveReplicas()).containsExactlyInAnyOrder(3);
+
assertThat(leaderElectionResult.getLeaderAndIsr().isr()).containsExactlyInAnyOrder(3);
+ // No remaining replicas for standby
+
assertThat(leaderElectionResult.getLeaderAndIsr().standbyReplicas()).isEmpty();
+ }
+
+ @Test
+ void testControlledShutdownAllIsrShutdownForPkTable() {
+ // Test when all ISR replicas are shutting down for PK table
+ List<Integer> assignments = Arrays.asList(1, 2, 3);
+ List<Integer> liveReplicas = Arrays.asList(1, 2, 3);
+ LeaderAndIsr originLeaderAndIsr =
+ new LeaderAndIsr(1, 0, Arrays.asList(1, 2, 3),
Collections.singletonList(2), 0, 0);
+ Set<Integer> shutdownTabletServers = new HashSet<>(Arrays.asList(1, 2,
3));
+
+ ControlledShutdownLeaderElection controlledShutdownLeaderElection =
+ new ControlledShutdownLeaderElection();
+ Optional<ElectionResult> leaderElectionResultOpt =
+ controlledShutdownLeaderElection.leaderElection(
+ assignments, liveReplicas, originLeaderAndIsr,
shutdownTabletServers, true);
+ // No available replicas, should return empty
+ assertThat(leaderElectionResultOpt).isEmpty();
+ }
+
+ @Test
+ void testReassignBucketLeaderElectionForPkTableWithStandbyUnavailable() {
+ // Test reassignment election for PK table when current standby is not
in new replicas
+ List<Integer> targetReplicas = Arrays.asList(4, 5, 6);
+ ReassignmentLeaderElection reassignmentLeaderElection =
+ new ReassignmentLeaderElection(targetReplicas);
+ List<Integer> liveReplicas = Arrays.asList(4, 5, 6);
+ // Current standby (3) is not in the new target replicas
+ LeaderAndIsr leaderAndIsr =
+ new LeaderAndIsr(1, 0, Arrays.asList(4, 5, 6),
Collections.singletonList(3), 0, 0);
+ Optional<ElectionResult> leaderOpt =
+ reassignmentLeaderElection.leaderElection(liveReplicas,
leaderAndIsr, true);
+ assertThat(leaderOpt).isPresent();
+ // First available replica should be leader since old standby (3) is
not available
+ assertThat(leaderOpt.get().getLeaderAndIsr().leader()).isEqualTo(4);
+ // New standby should be selected from remaining available replicas
+ assertThat(leaderOpt.get().getLeaderAndIsr().standbyReplicas())
+ .containsExactlyInAnyOrder(5);
+ }
+
+ @Test
+ void testReassignBucketLeaderElectionForPkTableNoAvailableReplicas() {
+ // Test reassignment election for PK table when no replicas are
available
+ List<Integer> targetReplicas = Arrays.asList(1, 2, 3);
+ ReassignmentLeaderElection reassignmentLeaderElection =
+ new ReassignmentLeaderElection(targetReplicas);
+ List<Integer> liveReplicas = Arrays.asList(4, 5); // None of the
targets are live
+ LeaderAndIsr leaderAndIsr =
+ new LeaderAndIsr(1, 0, Arrays.asList(4, 5),
Collections.singletonList(2), 0, 0);
+ Optional<ElectionResult> leaderOpt =
+ reassignmentLeaderElection.leaderElection(liveReplicas,
leaderAndIsr, true);
+ // No available replicas should return empty
+ assertThat(leaderOpt).isEmpty();
+ }
}
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 f1ad2fe1c..4d1946ab4 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
@@ -219,7 +219,8 @@ class ReplicaStateMachineTest {
replicas.add(replica);
}
// put leader and isr
- LeaderAndIsr leaderAndIsr = new LeaderAndIsr(0, 0, Arrays.asList(0, 1,
2), 0, 0);
+ LeaderAndIsr leaderAndIsr =
+ new LeaderAndIsr(0, 0, Arrays.asList(0, 1, 2),
Collections.emptyList(), 0, 0);
zookeeperClient.registerLeaderAndIsr(
tableBucket, leaderAndIsr,
ZkVersion.MATCH_ANY_VERSION.getVersion());
coordinatorContext.updateBucketReplicaAssignment(tableBucket,
Arrays.asList(0, 1, 2));
@@ -230,7 +231,14 @@ class ReplicaStateMachineTest {
replicaStateMachine.handleStateChanges(replicas, OfflineReplica);
leaderAndIsr =
coordinatorContext.getBucketLeaderAndIsr(tableBucket).get();
assertThat(leaderAndIsr)
- .isEqualTo(new LeaderAndIsr(LeaderAndIsr.NO_LEADER, 3,
Arrays.asList(2), 0, 3));
+ .isEqualTo(
+ new LeaderAndIsr(
+ LeaderAndIsr.NO_LEADER,
+ 3,
+ Collections.singletonList(2),
+ Collections.emptyList(),
+ 0,
+ 3));
}
@Test
@@ -257,7 +265,8 @@ class ReplicaStateMachineTest {
coordinatorContext.putReplicaState(replica, OnlineReplica);
}
// put leader and isr
- LeaderAndIsr leaderAndIsr = new LeaderAndIsr(0, 0, Arrays.asList(0, 1,
2), 0, 0);
+ LeaderAndIsr leaderAndIsr =
+ new LeaderAndIsr(0, 0, Arrays.asList(0, 1, 2),
Collections.emptyList(), 0, 0);
zookeeperClient.registerLeaderAndIsr(
tableBucket, leaderAndIsr,
ZkVersion.MATCH_ANY_VERSION.getVersion());
coordinatorContext.updateBucketReplicaAssignment(tableBucket,
Arrays.asList(0, 1, 2));
@@ -267,14 +276,18 @@ class ReplicaStateMachineTest {
replicaStateMachine.handleStateChanges(
Collections.singleton(new TableBucketReplica(tableBucket, 1)),
OfflineReplica);
leaderAndIsr =
coordinatorContext.getBucketLeaderAndIsr(tableBucket).get();
- assertThat(leaderAndIsr).isEqualTo(new LeaderAndIsr(0, 0,
Arrays.asList(0, 2), 0, 1));
+ assertThat(leaderAndIsr)
+ .isEqualTo(
+ new LeaderAndIsr(0, 0, Arrays.asList(0, 2),
Collections.emptyList(), 0, 1));
// set replica 2 to offline
replicaStateMachine.handleStateChanges(
Collections.singleton(new TableBucketReplica(tableBucket, 2)),
OfflineReplica);
leaderAndIsr =
coordinatorContext.getBucketLeaderAndIsr(tableBucket).get();
assertThat(leaderAndIsr)
- .isEqualTo(new LeaderAndIsr(0, 0,
Collections.singletonList(0), 0, 2));
+ .isEqualTo(
+ new LeaderAndIsr(
+ 0, 0, Collections.singletonList(0),
Collections.emptyList(), 0, 2));
// set replica 0 to offline, isr shouldn't be empty, leader should be
NO_LEADER
replicaStateMachine.handleStateChanges(
@@ -283,7 +296,12 @@ class ReplicaStateMachineTest {
assertThat(leaderAndIsr)
.isEqualTo(
new LeaderAndIsr(
- LeaderAndIsr.NO_LEADER, 1,
Collections.singletonList(0), 0, 3));
+ LeaderAndIsr.NO_LEADER,
+ 1,
+ Collections.singletonList(0),
+ Collections.emptyList(),
+ 0,
+ 3));
}
private void toReplicaDeletionStartedState(
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 2ea3718ff..7d80fe41c 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
@@ -150,6 +150,24 @@ class TableBucketStateMachineTest {
TableBucket t2b0 = new TableBucket(t2Id, 0);
coordinatorContext.putTablePath(t1Id, TablePath.of("db1", "t1"));
coordinatorContext.putTablePath(t2Id, TablePath.of("db1", "t2"));
+ coordinatorContext.putTableInfo(
+ TableInfo.of(
+ TablePath.of("db1", "t1"),
+ t1Id,
+ 0,
+ DATA1_TABLE_DESCRIPTOR,
+ DEFAULT_REMOTE_DATA_DIR,
+ System.currentTimeMillis(),
+ System.currentTimeMillis()));
+ coordinatorContext.putTableInfo(
+ TableInfo.of(
+ TablePath.of("db1", "t2"),
+ t2Id,
+ 0,
+ DATA1_TABLE_DESCRIPTOR,
+ DEFAULT_REMOTE_DATA_DIR,
+ System.currentTimeMillis(),
+ System.currentTimeMillis()));
coordinatorContext.setLiveTabletServers(createServers(Arrays.asList(0,
1, 3)));
makeSendLeaderAndStopRequestAlwaysSuccess(
@@ -162,11 +180,11 @@ 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 LeaderAndIsr(0, 0, Arrays.asList(0, 1),
Collections.emptyList(), 0, 0),
ZkVersion.MATCH_ANY_VERSION.getVersion());
zookeeperClient.registerLeaderAndIsr(
new TableBucket(t1Id, 1),
- new LeaderAndIsr(2, 0, Arrays.asList(2, 3), 0, 0),
+ new LeaderAndIsr(2, 0, Arrays.asList(2, 3),
Collections.emptyList(), 0, 0),
ZkVersion.MATCH_ANY_VERSION.getVersion());
// update the LeaderAndIsr to context
coordinatorContext.putBucketLeaderAndIsr(
@@ -311,6 +329,16 @@ class TableBucketStateMachineTest {
tableId = 5;
final TableBucket tableBucket1 = new TableBucket(tableId, 0);
coordinatorContext.putTablePath(tableId, fakeTablePath);
+ coordinatorContext.putTableInfo(
+ TableInfo.of(
+ fakeTablePath,
+ tableId,
+ 0,
+ DATA1_TABLE_DESCRIPTOR,
+ DEFAULT_REMOTE_DATA_DIR,
+ System.currentTimeMillis(),
+ System.currentTimeMillis()));
+ coordinatorContext.putTablePath(tableId, fakeTablePath);
coordinatorContext.updateBucketReplicaAssignment(tableBucket1,
Arrays.asList(0, 1, 2));
coordinatorContext.putBucketState(tableBucket1, NewBucket);
tableBucketStateMachine.handleStateChange(
@@ -418,12 +446,71 @@ class TableBucketStateMachineTest {
List<Integer> liveReplicas = Collections.singletonList(4);
Optional<ElectionResult> leaderElectionResultOpt =
- initReplicaLeaderElection(assignments, liveReplicas, 0);
+ initReplicaLeaderElection(assignments, liveReplicas, 0, false);
assertThat(leaderElectionResultOpt.isPresent()).isTrue();
ElectionResult leaderElectionResult = leaderElectionResultOpt.get();
assertThat(leaderElectionResult.getLiveReplicas()).containsExactlyInAnyOrder(4);
assertThat(leaderElectionResult.getLeaderAndIsr().leader()).isEqualTo(4);
assertThat(leaderElectionResult.getLeaderAndIsr().isr()).containsExactlyInAnyOrder(4);
+
assertThat(leaderElectionResult.getLeaderAndIsr().standbyReplicas()).isEmpty();
+ }
+
+ @Test
+ void testInitReplicaLeaderElectionForPkTable() {
+ // PK table with multiple available replicas: first as leader, second
as standby
+ List<Integer> assignments = Arrays.asList(1, 2, 3);
+ List<Integer> liveReplicas = Arrays.asList(1, 2, 3);
+
+ Optional<ElectionResult> resultOpt =
+ initReplicaLeaderElection(assignments, liveReplicas, 0, true);
+ assertThat(resultOpt).isPresent();
+ ElectionResult result = resultOpt.get();
+ assertThat(result.getLeaderAndIsr().leader()).isEqualTo(1);
+
assertThat(result.getLeaderAndIsr().isr()).containsExactlyInAnyOrder(1, 2, 3);
+
assertThat(result.getLeaderAndIsr().standbyReplicas()).containsExactly(2);
+ }
+
+ @Test
+ void testInitReplicaLeaderElectionForPkTableWithSingleAvailableReplica() {
+ // PK table with only one available replica: leader only, no standby
+ List<Integer> assignments = Arrays.asList(1, 2, 3);
+ List<Integer> liveReplicas = Collections.singletonList(2);
+
+ Optional<ElectionResult> resultOpt =
+ initReplicaLeaderElection(assignments, liveReplicas, 0, true);
+ assertThat(resultOpt).isPresent();
+ ElectionResult result = resultOpt.get();
+ assertThat(result.getLeaderAndIsr().leader()).isEqualTo(2);
+
assertThat(result.getLeaderAndIsr().isr()).containsExactlyInAnyOrder(2);
+ assertThat(result.getLeaderAndIsr().standbyReplicas()).isEmpty();
+ }
+
+ @Test
+ void testInitReplicaLeaderElectionForPkTableWithNoAliveReplica() {
+ // PK table with no alive replicas: should return empty
+ List<Integer> assignments = Arrays.asList(1, 2, 3);
+ List<Integer> liveReplicas = Collections.emptyList();
+
+ Optional<ElectionResult> resultOpt =
+ initReplicaLeaderElection(assignments, liveReplicas, 0, true);
+ assertThat(resultOpt).isEmpty();
+ }
+
+ @Test
+ void testInitReplicaLeaderElectionForPkTableWithPartiallyAliveReplicas() {
+ // PK table where first assigned replica is not alive,
+ // second and third are alive
+ List<Integer> assignments = Arrays.asList(1, 2, 3);
+ List<Integer> liveReplicas = Arrays.asList(2, 3);
+
+ Optional<ElectionResult> resultOpt =
+ initReplicaLeaderElection(assignments, liveReplicas, 0, true);
+ assertThat(resultOpt).isPresent();
+ ElectionResult result = resultOpt.get();
+ // First available (2) should be leader, second available (3) should
be standby
+ assertThat(result.getLeaderAndIsr().leader()).isEqualTo(2);
+
assertThat(result.getLeaderAndIsr().isr()).containsExactlyInAnyOrder(2, 3);
+
assertThat(result.getLeaderAndIsr().standbyReplicas()).containsExactly(3);
}
private TableBucketStateMachine createTableBucketStateMachine() {
diff --git
a/fluss-server/src/test/java/org/apache/fluss/server/log/remote/RemoteLogTestBase.java
b/fluss-server/src/test/java/org/apache/fluss/server/log/remote/RemoteLogTestBase.java
index 13ab2f7ba..ef6b2875d 100644
---
a/fluss-server/src/test/java/org/apache/fluss/server/log/remote/RemoteLogTestBase.java
+++
b/fluss-server/src/test/java/org/apache/fluss/server/log/remote/RemoteLogTestBase.java
@@ -87,7 +87,13 @@ public class RemoteLogTestBase extends ReplicaTestBase {
physicalTablePath,
tb,
Collections.singletonList(0),
- new LeaderAndIsr(0, 0, Collections.singletonList(0),
0, 0)));
+ new LeaderAndIsr(
+ 0,
+ 0,
+ Collections.singletonList(0),
+ Collections.emptyList(),
+ 0,
+ 0)));
addMultiSegmentsToLogTablet(replica.getLogTablet(), segmentSize);
return replica;
}
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 5597bb01b..6ef53b62a 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
@@ -108,8 +108,10 @@ class ZkBasedMetadataProviderTest {
TableBucket tableBucket0 = new TableBucket(tableId, 0);
TableBucket tableBucket1 = new TableBucket(tableId, 1);
- 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);
+ LeaderAndIsr leaderAndIsr0 =
+ new LeaderAndIsr(1, 10, Arrays.asList(1, 2, 3),
Collections.emptyList(), 100, 1000);
+ LeaderAndIsr leaderAndIsr1 =
+ new LeaderAndIsr(2, 20, Arrays.asList(2, 3, 4),
Collections.emptyList(), 200, 2000);
zookeeperClient.registerLeaderAndIsr(
tableBucket0, leaderAndIsr0,
ZkVersion.MATCH_ANY_VERSION.getVersion());
@@ -176,8 +178,10 @@ class ZkBasedMetadataProviderTest {
TableBucket partitionBucket0 = new TableBucket(tableId, partitionId,
0);
TableBucket partitionBucket1 = new TableBucket(tableId, partitionId,
1);
- LeaderAndIsr leaderAndIsr0 = new LeaderAndIsr(1, 10, Arrays.asList(1,
2), 100, 1000);
- LeaderAndIsr leaderAndIsr1 = new LeaderAndIsr(2, 20, Arrays.asList(2,
3), 200, 2000);
+ LeaderAndIsr leaderAndIsr0 =
+ new LeaderAndIsr(1, 10, Arrays.asList(1, 2),
Collections.emptyList(), 100, 1000);
+ LeaderAndIsr leaderAndIsr1 =
+ new LeaderAndIsr(2, 20, Arrays.asList(2, 3),
Collections.emptyList(), 200, 2000);
zookeeperClient.registerLeaderAndIsr(
partitionBucket0, leaderAndIsr0,
ZkVersion.MATCH_ANY_VERSION.getVersion());
@@ -281,15 +285,15 @@ class ZkBasedMetadataProviderTest {
zookeeperClient.registerLeaderAndIsr(
bucket1,
- new LeaderAndIsr(1, 10, Arrays.asList(1, 2), 100, 1000),
+ new LeaderAndIsr(1, 10, Arrays.asList(1, 2),
Collections.emptyList(), 100, 1000),
ZkVersion.MATCH_ANY_VERSION.getVersion());
zookeeperClient.registerLeaderAndIsr(
bucket2,
- new LeaderAndIsr(2, 20, Arrays.asList(2, 3), 200, 2000),
+ new LeaderAndIsr(2, 20, Arrays.asList(2, 3),
Collections.emptyList(), 200, 2000),
ZkVersion.MATCH_ANY_VERSION.getVersion());
zookeeperClient.registerLeaderAndIsr(
bucket3,
- new LeaderAndIsr(1, 30, Arrays.asList(1, 3), 300, 3000),
+ new LeaderAndIsr(1, 30, Arrays.asList(1, 3),
Collections.emptyList(), 300, 3000),
ZkVersion.MATCH_ANY_VERSION.getVersion());
// Test getPartitionsMetadataFromZK
diff --git
a/fluss-server/src/test/java/org/apache/fluss/server/replica/AdjustIsrITCase.java
b/fluss-server/src/test/java/org/apache/fluss/server/replica/AdjustIsrITCase.java
index 2931ca546..846a1f5f5 100644
---
a/fluss-server/src/test/java/org/apache/fluss/server/replica/AdjustIsrITCase.java
+++
b/fluss-server/src/test/java/org/apache/fluss/server/replica/AdjustIsrITCase.java
@@ -135,6 +135,7 @@ public class AdjustIsrITCase {
currentLeaderAndIsr.leader(),
currentLeaderAndIsr.leaderEpoch() + 1,
isr,
+ currentLeaderAndIsr.standbyReplicas(),
currentLeaderAndIsr.coordinatorEpoch(),
currentLeaderAndIsr.bucketEpoch());
isr.add(stopFollower);
diff --git
a/fluss-server/src/test/java/org/apache/fluss/server/replica/AdjustIsrManagerTest.java
b/fluss-server/src/test/java/org/apache/fluss/server/replica/AdjustIsrManagerTest.java
index 248e7929e..7bdffcafe 100644
---
a/fluss-server/src/test/java/org/apache/fluss/server/replica/AdjustIsrManagerTest.java
+++
b/fluss-server/src/test/java/org/apache/fluss/server/replica/AdjustIsrManagerTest.java
@@ -26,6 +26,7 @@ import org.apache.fluss.utils.concurrent.FlussScheduler;
import org.junit.jupiter.api.Test;
import java.util.Arrays;
+import java.util.Collections;
import java.util.List;
import static org.assertj.core.api.Assertions.assertThat;
@@ -46,17 +47,32 @@ class AdjustIsrManagerTest {
// shrink isr
TableBucket tb = new TableBucket(150001L, 0);
List<Integer> currentIsr = Arrays.asList(1, 2);
- LeaderAndIsr adjustIsr = new LeaderAndIsr(tabletServerId, 0,
currentIsr, 0, 0);
+ LeaderAndIsr adjustIsr =
+ new LeaderAndIsr(tabletServerId, 0, currentIsr,
Collections.emptyList(), 0, 0);
LeaderAndIsr result = adjustIsrManager.submit(tb, adjustIsr).get();
assertThat(result)
- .isEqualTo(new LeaderAndIsr(tabletServerId, 0,
Arrays.asList(1, 2), 0, 1));
+ .isEqualTo(
+ new LeaderAndIsr(
+ tabletServerId,
+ 0,
+ Arrays.asList(1, 2),
+ Collections.emptyList(),
+ 0,
+ 1));
// expand isr
currentIsr = Arrays.asList(1, 2, 3);
- adjustIsr = new LeaderAndIsr(tabletServerId, 0, currentIsr, 0, 1);
+ adjustIsr = new LeaderAndIsr(tabletServerId, 0, currentIsr,
Collections.emptyList(), 0, 1);
result = adjustIsrManager.submit(tb, adjustIsr).get();
assertThat(result)
- .isEqualTo(new LeaderAndIsr(tabletServerId, 0,
Arrays.asList(1, 2, 3), 0, 2));
+ .isEqualTo(
+ new LeaderAndIsr(
+ tabletServerId,
+ 0,
+ Arrays.asList(1, 2, 3),
+ Collections.emptyList(),
+ 0,
+ 2));
}
@Test
@@ -71,7 +87,8 @@ class AdjustIsrManagerTest {
// The RPC-level exception is propagated to the submit future.
TableBucket tb = new TableBucket(150001L, 0);
List<Integer> currentIsr = Arrays.asList(1, 2);
- LeaderAndIsr adjustIsr = new LeaderAndIsr(tabletServerId, 0,
currentIsr, 0, 0);
+ LeaderAndIsr adjustIsr =
+ new LeaderAndIsr(tabletServerId, 0, currentIsr,
Collections.emptyList(), 0, 0);
assertThatThrownBy(() -> adjustIsrManager.submit(tb, adjustIsr).get())
.rootCause()
.isInstanceOf(NetworkException.class)
@@ -83,6 +100,13 @@ class AdjustIsrManagerTest {
// submit.
LeaderAndIsr result = adjustIsrManager.submit(tb, adjustIsr).get();
assertThat(result)
- .isEqualTo(new LeaderAndIsr(tabletServerId, 0,
Arrays.asList(1, 2), 0, 1));
+ .isEqualTo(
+ new LeaderAndIsr(
+ tabletServerId,
+ 0,
+ Arrays.asList(1, 2),
+ Collections.emptyList(),
+ 0,
+ 1));
}
}
diff --git
a/fluss-server/src/test/java/org/apache/fluss/server/replica/AdjustIsrTest.java
b/fluss-server/src/test/java/org/apache/fluss/server/replica/AdjustIsrTest.java
index 1779b96f8..6ab4ca69a 100644
---
a/fluss-server/src/test/java/org/apache/fluss/server/replica/AdjustIsrTest.java
+++
b/fluss-server/src/test/java/org/apache/fluss/server/replica/AdjustIsrTest.java
@@ -133,7 +133,8 @@ public class AdjustIsrTest extends ReplicaTestBase {
// To mock we prepare an isr shrink in Replica#maybeShrinkIsr();
IsrState.PendingShrinkIsrState pendingShrinkIsrState =
replica.prepareIsrShrink(
- new IsrState.CommittedIsrState(Arrays.asList(1, 2, 3)),
+ new IsrState.CommittedIsrState(
+ Arrays.asList(1, 2, 3),
Collections.emptyList()),
Arrays.asList(1, 2),
Collections.singletonList(3));
@@ -161,7 +162,8 @@ public class AdjustIsrTest extends ReplicaTestBase {
// To mock we prepare an isr shrink in Replica#maybeShrinkIsr();
IsrState.PendingShrinkIsrState pendingShrinkIsrState =
replica.prepareIsrShrink(
- new IsrState.CommittedIsrState(Arrays.asList(1, 2, 3)),
+ new IsrState.CommittedIsrState(
+ Arrays.asList(1, 2, 3),
Collections.emptyList()),
Arrays.asList(1, 2),
Collections.singletonList(3));
@@ -176,6 +178,6 @@ public class AdjustIsrTest extends ReplicaTestBase {
.hasMessage(
"Rejecting adjustIsr request for table bucket "
+ "TableBucket{tableId=150001, bucket=1}
because it specified ineligible replicas [2] "
- + "in the new ISR LeaderAndIsr{leader=1,
leaderEpoch=0, isr=[1, 2], coordinatorEpoch=0, bucketEpoch=0}");
+ + "in the new ISR LeaderAndIsr{leader=1,
leaderEpoch=0, isr=[1, 2], standbyReplicas=[], coordinatorEpoch=0,
bucketEpoch=0}");
}
}
diff --git
a/fluss-server/src/test/java/org/apache/fluss/server/replica/HighWatermarkPersistenceTest.java
b/fluss-server/src/test/java/org/apache/fluss/server/replica/HighWatermarkPersistenceTest.java
index b093787f0..a520f9dad 100644
---
a/fluss-server/src/test/java/org/apache/fluss/server/replica/HighWatermarkPersistenceTest.java
+++
b/fluss-server/src/test/java/org/apache/fluss/server/replica/HighWatermarkPersistenceTest.java
@@ -105,6 +105,7 @@ final class HighWatermarkPersistenceTest extends
ReplicaTestBase {
TABLET_SERVER_ID,
LeaderAndIsr.INITIAL_LEADER_EPOCH,
Collections.singletonList(TABLET_SERVER_ID),
+ Collections.emptyList(),
INITIAL_COORDINATOR_EPOCH,
LeaderAndIsr.INITIAL_BUCKET_EPOCH))),
result -> {});
diff --git
a/fluss-server/src/test/java/org/apache/fluss/server/replica/KvRecoverFromRemoteLogITCase.java
b/fluss-server/src/test/java/org/apache/fluss/server/replica/KvRecoverFromRemoteLogITCase.java
index a070870eb..47c5ec88f 100644
---
a/fluss-server/src/test/java/org/apache/fluss/server/replica/KvRecoverFromRemoteLogITCase.java
+++
b/fluss-server/src/test/java/org/apache/fluss/server/replica/KvRecoverFromRemoteLogITCase.java
@@ -43,6 +43,7 @@ import org.junit.jupiter.api.extension.RegisterExtension;
import java.time.Duration;
import java.util.ArrayList;
+import java.util.Collections;
import java.util.List;
import java.util.Optional;
import java.util.concurrent.CompletableFuture;
@@ -215,6 +216,7 @@ class KvRecoverFromRemoteLogITCase {
followerToPromote,
currentLeaderAndIsr.leaderEpoch() + 1,
currentLeaderAndIsr.isr(),
+ Collections.emptyList(),
currentLeaderAndIsr.coordinatorEpoch(),
currentLeaderAndIsr.bucketEpoch() + 1);
diff --git
a/fluss-server/src/test/java/org/apache/fluss/server/replica/ReplicaManagerTest.java
b/fluss-server/src/test/java/org/apache/fluss/server/replica/ReplicaManagerTest.java
index 0d9f5bc56..5120bf6a6 100644
---
a/fluss-server/src/test/java/org/apache/fluss/server/replica/ReplicaManagerTest.java
+++
b/fluss-server/src/test/java/org/apache/fluss/server/replica/ReplicaManagerTest.java
@@ -1458,6 +1458,7 @@ class ReplicaManagerTest extends ReplicaTestBase {
TABLET_SERVER_ID,
1,
Arrays.asList(1, 2, 3),
+ Collections.emptyList(),
INITIAL_COORDINATOR_EPOCH,
INITIAL_BUCKET_EPOCH))),
future::complete);
@@ -1478,6 +1479,7 @@ class ReplicaManagerTest extends ReplicaTestBase {
TABLET_SERVER_ID,
INITIAL_LEADER_EPOCH,
Arrays.asList(1, 2, 3),
+ Collections.emptyList(),
INITIAL_COORDINATOR_EPOCH,
INITIAL_BUCKET_EPOCH))),
future::complete);
@@ -1512,6 +1514,7 @@ class ReplicaManagerTest extends ReplicaTestBase {
TABLET_SERVER_ID,
1,
Arrays.asList(1, 2, 3),
+ Collections.emptyList(),
INITIAL_COORDINATOR_EPOCH,
INITIAL_BUCKET_EPOCH))),
future::complete);
@@ -1543,6 +1546,7 @@ class ReplicaManagerTest extends ReplicaTestBase {
TABLET_SERVER_ID,
2,
Arrays.asList(1, 2, 3),
+ Collections.emptyList(),
INITIAL_COORDINATOR_EPOCH,
INITIAL_BUCKET_EPOCH))),
future::complete);
diff --git
a/fluss-server/src/test/java/org/apache/fluss/server/replica/ReplicaTest.java
b/fluss-server/src/test/java/org/apache/fluss/server/replica/ReplicaTest.java
index 04331fb6a..4eefc0e2f 100644
---
a/fluss-server/src/test/java/org/apache/fluss/server/replica/ReplicaTest.java
+++
b/fluss-server/src/test/java/org/apache/fluss/server/replica/ReplicaTest.java
@@ -801,6 +801,7 @@ final class ReplicaTest extends ReplicaTestBase {
TABLET_SERVER_ID,
leaderEpoch,
Collections.singletonList(TABLET_SERVER_ID),
+ Collections.emptyList(),
INITIAL_COORDINATOR_EPOCH,
// we also use the leader epoch as bucket epoch
leaderEpoch)));
@@ -818,6 +819,7 @@ final class ReplicaTest extends ReplicaTestBase {
TABLET_SERVER_ID,
leaderEpoch,
Collections.singletonList(TABLET_SERVER_ID),
+ Collections.emptyList(),
INITIAL_COORDINATOR_EPOCH,
// we also use the leader epoch as bucket epoch
leaderEpoch)));
diff --git
a/fluss-server/src/test/java/org/apache/fluss/server/replica/ReplicaTestBase.java
b/fluss-server/src/test/java/org/apache/fluss/server/replica/ReplicaTestBase.java
index 8c2f772c3..f6b2e01c8 100644
---
a/fluss-server/src/test/java/org/apache/fluss/server/replica/ReplicaTestBase.java
+++
b/fluss-server/src/test/java/org/apache/fluss/server/replica/ReplicaTestBase.java
@@ -396,6 +396,7 @@ public class ReplicaTestBase {
TABLET_SERVER_ID,
INITIAL_LEADER_EPOCH,
isr,
+ Collections.emptyList(),
INITIAL_COORDINATOR_EPOCH,
INITIAL_BUCKET_EPOCH))));
}
@@ -437,6 +438,7 @@ public class ReplicaTestBase {
TABLET_SERVER_ID,
leaderEpoch,
isr,
+ Collections.emptyList(),
INITIAL_COORDINATOR_EPOCH,
// use leader epoch as bucket epoch
leaderEpoch))));
diff --git
a/fluss-server/src/test/java/org/apache/fluss/server/replica/fetcher/ReplicaFetcherITCase.java
b/fluss-server/src/test/java/org/apache/fluss/server/replica/fetcher/ReplicaFetcherITCase.java
index e2bb80f96..f94db7184 100644
---
a/fluss-server/src/test/java/org/apache/fluss/server/replica/fetcher/ReplicaFetcherITCase.java
+++
b/fluss-server/src/test/java/org/apache/fluss/server/replica/fetcher/ReplicaFetcherITCase.java
@@ -315,6 +315,7 @@ public class ReplicaFetcherITCase {
currentLeaderAndIsr.leader(),
currentLeaderAndIsr.leaderEpoch() + 1,
currentLeaderAndIsr.isr(),
+ Collections.emptyList(),
currentLeaderAndIsr.coordinatorEpoch(),
currentLeaderAndIsr.bucketEpoch());
FLUSS_CLUSTER_EXTENSION.notifyLeaderAndIsr(
diff --git
a/fluss-server/src/test/java/org/apache/fluss/server/replica/fetcher/ReplicaFetcherManagerTest.java
b/fluss-server/src/test/java/org/apache/fluss/server/replica/fetcher/ReplicaFetcherManagerTest.java
index 4329ff6b3..15be1f2b0 100644
---
a/fluss-server/src/test/java/org/apache/fluss/server/replica/fetcher/ReplicaFetcherManagerTest.java
+++
b/fluss-server/src/test/java/org/apache/fluss/server/replica/fetcher/ReplicaFetcherManagerTest.java
@@ -91,6 +91,7 @@ class ReplicaFetcherManagerTest extends ReplicaTestBase {
leader.id(),
LeaderAndIsr.INITIAL_LEADER_EPOCH,
Arrays.asList(leader.id(),
TABLET_SERVER_ID),
+ Collections.emptyList(),
INITIAL_COORDINATOR_EPOCH,
LeaderAndIsr.INITIAL_BUCKET_EPOCH))),
result -> {});
diff --git
a/fluss-server/src/test/java/org/apache/fluss/server/replica/fetcher/ReplicaFetcherThreadTest.java
b/fluss-server/src/test/java/org/apache/fluss/server/replica/fetcher/ReplicaFetcherThreadTest.java
index 556c5bea4..ff089cb4a 100644
---
a/fluss-server/src/test/java/org/apache/fluss/server/replica/fetcher/ReplicaFetcherThreadTest.java
+++
b/fluss-server/src/test/java/org/apache/fluss/server/replica/fetcher/ReplicaFetcherThreadTest.java
@@ -477,6 +477,7 @@ public class ReplicaFetcherThreadTest {
leaderServerId,
INITIAL_LEADER_EPOCH,
Arrays.asList(leaderServerId,
followerServerId),
+ Collections.emptyList(),
INITIAL_COORDINATOR_EPOCH,
INITIAL_BUCKET_EPOCH))),
result -> {});
@@ -491,6 +492,7 @@ public class ReplicaFetcherThreadTest {
leaderServerId,
INITIAL_LEADER_EPOCH,
Arrays.asList(leaderServerId,
followerServerId),
+ Collections.emptyList(),
INITIAL_COORDINATOR_EPOCH,
INITIAL_BUCKET_EPOCH))),
result -> {});
diff --git
a/fluss-server/src/test/java/org/apache/fluss/server/tablet/TabletServerShutdownITCase.java
b/fluss-server/src/test/java/org/apache/fluss/server/tablet/TabletServerShutdownITCase.java
index 68ec0d911..95e66f252 100644
---
a/fluss-server/src/test/java/org/apache/fluss/server/tablet/TabletServerShutdownITCase.java
+++
b/fluss-server/src/test/java/org/apache/fluss/server/tablet/TabletServerShutdownITCase.java
@@ -217,7 +217,9 @@ public class TabletServerShutdownITCase {
// should be no leader
LeaderAndIsr expectedLeaderAndIsr2 =
expectedLeaderAndIsr1.newLeaderAndIsr(
- LeaderAndIsr.NO_LEADER,
Collections.singletonList(leader));
+ LeaderAndIsr.NO_LEADER,
+ Collections.singletonList(leader),
+ Collections.emptyList());
retry(
Duration.ofMinutes(1),
() ->
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 7bdd78107..17b436e1a 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
@@ -866,6 +866,7 @@ public class TabletServiceITCase {
leader,
1,
originLeaderAndIsr.isr(),
+
originLeaderAndIsr.standbyReplicas(),
originLeaderAndIsr.coordinatorEpoch(),
originLeaderAndIsr.bucketEpoch())))
.get();
@@ -909,6 +910,7 @@ public class TabletServiceITCase {
leader,
2,
originLeaderAndIsr.isr(),
+
originLeaderAndIsr.standbyReplicas(),
originLeaderAndIsr.coordinatorEpoch(),
originLeaderAndIsr.bucketEpoch())))
.get();
diff --git
a/fluss-server/src/test/java/org/apache/fluss/server/testutils/FlussClusterExtension.java
b/fluss-server/src/test/java/org/apache/fluss/server/testutils/FlussClusterExtension.java
index 74a041596..00a2344a6 100644
---
a/fluss-server/src/test/java/org/apache/fluss/server/testutils/FlussClusterExtension.java
+++
b/fluss-server/src/test/java/org/apache/fluss/server/testutils/FlussClusterExtension.java
@@ -921,6 +921,11 @@ public final class FlussClusterExtension
return waitLeaderAndIsrReady(tb).leader();
}
+ public int waitAndGetStandby(TableBucket tb) {
+ List<Integer> standbyReplicas =
waitLeaderAndIsrReady(tb).standbyReplicas();
+ return standbyReplicas.get(0);
+ }
+
public LeaderAndIsr waitLeaderAndIsrReady(TableBucket tb) {
ZooKeeperClient zkClient = getZooKeeperClient();
return waitValue(
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 16f59ddd8..a94cf8e0d 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
@@ -202,8 +202,11 @@ class ZooKeeperClientTest {
assertThat(zookeeperClient.getLeaderAndIsr(tableBucket2)).isEmpty();
// try to register bucket leaderAndIsr
- 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);
+ LeaderAndIsr leaderAndIsr1 =
+ new LeaderAndIsr(
+ 1, 10, Arrays.asList(1, 2, 3),
Collections.singletonList(2), 0, 1000);
+ LeaderAndIsr leaderAndIsr2 =
+ new LeaderAndIsr(2, 10, Arrays.asList(4, 5, 6),
Collections.emptyList(), 0, 1000);
zookeeperClient.registerLeaderAndIsr(
tableBucket1, leaderAndIsr1,
zkEpoch.getCoordinatorEpochZkVersion());
@@ -215,7 +218,9 @@ class ZooKeeperClientTest {
.containsValues(leaderAndIsr1, leaderAndIsr2);
// test update
- leaderAndIsr1 = new LeaderAndIsr(2, 20, Collections.emptyList(), 0,
2000);
+ leaderAndIsr1 =
+ new LeaderAndIsr(
+ 2, 20, Collections.emptyList(),
Collections.singletonList(3), 0, 2000);
zookeeperClient.updateLeaderAndIsr(
tableBucket1, leaderAndIsr1,
zkEpoch.getCoordinatorEpochZkVersion());
assertThat(zookeeperClient.getLeaderAndIsr(tableBucket1)).hasValue(leaderAndIsr1);
@@ -234,7 +239,13 @@ 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), 0, 1000);
+ new LeaderAndIsr(
+ i,
+ 10,
+ Arrays.asList(i + 1, i + 2, i + 3),
+ Collections.emptyList(),
+ 0,
+ 1000);
leaderAndIsrList.add(leaderAndIsr);
RegisterTableBucketLeadAndIsrInfo info =
isPartitionTable
@@ -271,7 +282,8 @@ class ZooKeeperClientTest {
LeaderAndIsr adjustLeaderAndIsr =
originalLeaderAndIsr.newLeaderAndIsr(
LeaderAndIsr.NO_LEADER,
-
originalLeaderAndIsr.isr().subList(0, 1));
+
originalLeaderAndIsr.isr().subList(0, 1),
+ Collections.emptyList());
leaderAndIsrUpdateList.add(adjustLeaderAndIsr);
entry.setValue(adjustLeaderAndIsr);
});
@@ -295,7 +307,13 @@ 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), 0, 1000);
+ new LeaderAndIsr(
+ i,
+ 10,
+ Arrays.asList(i + 1, i + 2, i + 3),
+ Collections.emptyList(),
+ 0,
+ 1000);
leaderAndIsrList.put(tableBucket, leaderAndIsr);
zookeeperClient.registerLeaderAndIsr(
tableBucket, leaderAndIsr,
zkEpoch.getCoordinatorEpochZkVersion());
@@ -313,6 +331,7 @@ class ZooKeeperClientTest {
old.leader() + 1,
old.leaderEpoch() + 1,
old.isr(),
+ Collections.emptyList(),
old.coordinatorEpoch(),
old.bucketEpoch() + 1);
}));
diff --git
a/fluss-server/src/test/java/org/apache/fluss/server/zk/data/LeaderAndIsrJsonSerdeTest.java
b/fluss-server/src/test/java/org/apache/fluss/server/zk/data/LeaderAndIsrJsonSerdeTest.java
index 128b14a18..f71a98d98 100644
---
a/fluss-server/src/test/java/org/apache/fluss/server/zk/data/LeaderAndIsrJsonSerdeTest.java
+++
b/fluss-server/src/test/java/org/apache/fluss/server/zk/data/LeaderAndIsrJsonSerdeTest.java
@@ -17,12 +17,17 @@
package org.apache.fluss.server.zk.data;
+import
org.apache.fluss.shaded.jackson2.com.fasterxml.jackson.databind.JsonNode;
+import
org.apache.fluss.shaded.jackson2.com.fasterxml.jackson.databind.ObjectMapper;
import org.apache.fluss.utils.json.JsonSerdeTestBase;
+import org.junit.jupiter.api.Test;
+
+import java.nio.charset.StandardCharsets;
import java.util.Arrays;
import java.util.Collections;
-/** Test for {@link org.apache.fluss.server.zk.data.LeaderAndIsrJsonSerde}. */
+/** Test for {@link LeaderAndIsrJsonSerde}. */
public class LeaderAndIsrJsonSerdeTest extends JsonSerdeTestBase<LeaderAndIsr>
{
LeaderAndIsrJsonSerdeTest() {
@@ -31,16 +36,37 @@ public class LeaderAndIsrJsonSerdeTest extends
JsonSerdeTestBase<LeaderAndIsr> {
@Override
protected LeaderAndIsr[] createObjects() {
- LeaderAndIsr leaderAndIsr1 = new LeaderAndIsr(1, 10, Arrays.asList(1,
2, 3), 100, 1000);
- LeaderAndIsr leaderAndIsr2 = new LeaderAndIsr(2, 20,
Collections.emptyList(), 200, 2000);
- return new LeaderAndIsr[] {leaderAndIsr1, leaderAndIsr2};
+ LeaderAndIsr leaderAndIsr1 =
+ new LeaderAndIsr(1, 10, Arrays.asList(1, 2, 3),
Collections.emptyList(), 100, 1000);
+ LeaderAndIsr leaderAndIsr2 =
+ new LeaderAndIsr(
+ 2, 20, Collections.emptyList(),
Collections.emptyList(), 200, 2000);
+ LeaderAndIsr leaderAndIsr3 =
+ new LeaderAndIsr(
+ 2, 20, Arrays.asList(1, 2, 3),
Collections.singletonList(3), 200, 2000);
+ return new LeaderAndIsr[] {leaderAndIsr1, leaderAndIsr2,
leaderAndIsr3};
}
@Override
protected String[] expectedJsons() {
return new String[] {
-
"{\"version\":1,\"leader\":1,\"leader_epoch\":10,\"isr\":[1,2,3],\"coordinator_epoch\":100,\"bucket_epoch\":1000}",
-
"{\"version\":1,\"leader\":2,\"leader_epoch\":20,\"isr\":[],\"coordinator_epoch\":200,\"bucket_epoch\":2000}"
+
"{\"version\":2,\"leader\":1,\"leader_epoch\":10,\"isr\":[1,2,3],\"coordinator_epoch\":100,\"bucket_epoch\":1000,\"standby_replicas\":[]}",
+
"{\"version\":2,\"leader\":2,\"leader_epoch\":20,\"isr\":[],\"coordinator_epoch\":200,\"bucket_epoch\":2000,\"standby_replicas\":[]}",
+
"{\"version\":2,\"leader\":2,\"leader_epoch\":20,\"isr\":[1,2,3],\"coordinator_epoch\":200,\"bucket_epoch\":2000,\"standby_replicas\":[3]}"
};
}
+
+ @Test
+ void testCompatibility() throws Exception {
+ // test compatibility with version 1
+ JsonNode jsonInVersion1 =
+ new ObjectMapper()
+ .readTree(
+
"{\"version\":1,\"leader\":1,\"leader_epoch\":10,\"isr\":[1,2,3],\"coordinator_epoch\":100,\"bucket_epoch\":1000}"
+ .getBytes(StandardCharsets.UTF_8));
+ LeaderAndIsr leaderAndIsr =
LeaderAndIsrJsonSerde.INSTANCE.deserialize(jsonInVersion1);
+ LeaderAndIsr expectedLeaderAndIsr =
+ new LeaderAndIsr(1, 10, Arrays.asList(1, 2, 3),
Collections.emptyList(), 100, 1000);
+ assertEquals(leaderAndIsr, expectedLeaderAndIsr);
+ }
}
