From f4a3c4b7211c7f19d55a8cba4d0f876cad382206 Mon Sep 17 00:00:00 2001 From: Guangcun Li Date: Sat, 7 Nov 2026 22:30:29 +0800 Subject: [PATCH] RATIS-2806. Preserve existing listeners in ADD mode --- .../ratis/server/impl/LeaderStateImpl.java | 4 +- .../ratis/server/impl/RaftServerImpl.java | 2 +- .../ratis/server/raftlog/RaftLogBase.java | 4 -- .../impl/RaftReconfigurationBaseTest.java | 40 +++++++++++++++++++ .../ratis/server/impl/RaftServerTestUtil.java | 35 ++++++++++++++++ 5 files changed, 78 insertions(+), 7 deletions(-) diff --git a/ratis-server/src/main/java/org/apache/ratis/server/impl/LeaderStateImpl.java b/ratis-server/src/main/java/org/apache/ratis/server/impl/LeaderStateImpl.java index 09a9e12ded..ed41dcd910 100644 --- a/ratis-server/src/main/java/org/apache/ratis/server/impl/LeaderStateImpl.java +++ b/ratis-server/src/main/java/org/apache/ratis/server/impl/LeaderStateImpl.java @@ -519,12 +519,12 @@ public boolean onFollowerTerm(FollowerInfo follower, long followerTerm) { /** * Start bootstrapping new peers */ - PendingRequest startSetConfiguration(SetConfigurationRequest request, List peersInNewConf) { + PendingRequest startSetConfiguration(SetConfigurationRequest request, + List peersInNewConf, List listenersInNewConf) { LOG.info("{}: startSetConfiguration {}", this, request); Preconditions.assertTrue(isRunning(), () -> this + " is not running."); Preconditions.assertTrue(!inStagingState(), () -> this + " is already in staging state " + stagingState); - final List listenersInNewConf = request.getArguments().getPeersInNewConf(RaftPeerRole.LISTENER); final Collection peersToBootStrap = server.getRaftConf().filterNotContainedInConf(peersInNewConf); final Collection listenersToBootStrap= server.getRaftConf().filterNotContainedInConf(listenersInNewConf); diff --git a/ratis-server/src/main/java/org/apache/ratis/server/impl/RaftServerImpl.java b/ratis-server/src/main/java/org/apache/ratis/server/impl/RaftServerImpl.java index 18ccdf9f6e..73b6ce7f15 100644 --- a/ratis-server/src/main/java/org/apache/ratis/server/impl/RaftServerImpl.java +++ b/ratis-server/src/main/java/org/apache/ratis/server/impl/RaftServerImpl.java @@ -1413,7 +1413,7 @@ public CompletableFuture setConfigurationAsync(SetConfiguration getRaftServer().addRaftPeers(serversInNewConf); getRaftServer().addRaftPeers(listenersInNewConf); // add staging state into the leaderState - pending = leaderState.startSetConfiguration(request, serversInNewConf); + pending = leaderState.startSetConfiguration(request, serversInNewConf, listenersInNewConf); } return pending.getFuture(); } diff --git a/ratis-server/src/main/java/org/apache/ratis/server/raftlog/RaftLogBase.java b/ratis-server/src/main/java/org/apache/ratis/server/raftlog/RaftLogBase.java index 48b410147c..23c9052526 100644 --- a/ratis-server/src/main/java/org/apache/ratis/server/raftlog/RaftLogBase.java +++ b/ratis-server/src/main/java/org/apache/ratis/server/raftlog/RaftLogBase.java @@ -54,10 +54,6 @@ public abstract class RaftLogBase implements RaftLog { private final Consumer infoIndexChange = s -> LOG.info("{}: {}", getName(), s); private final Consumer traceIndexChange = s -> LOG.trace("{}: {}", getName(), s); - /** The least valid log index, i.e. the index used when writing to an empty log. */ - public static final long LEAST_VALID_LOG_INDEX = 0L; - public static final long INVALID_LOG_INDEX = LEAST_VALID_LOG_INDEX - 1; - private final String name; /** * The largest committed index. Note the last committed log may be included diff --git a/ratis-server/src/test/java/org/apache/ratis/server/impl/RaftReconfigurationBaseTest.java b/ratis-server/src/test/java/org/apache/ratis/server/impl/RaftReconfigurationBaseTest.java index efdb6b98f0..1f32e32a46 100644 --- a/ratis-server/src/test/java/org/apache/ratis/server/impl/RaftReconfigurationBaseTest.java +++ b/ratis-server/src/test/java/org/apache/ratis/server/impl/RaftReconfigurationBaseTest.java @@ -23,6 +23,7 @@ import org.apache.ratis.client.RaftClient; import org.apache.ratis.client.RaftClientRpc; import org.apache.ratis.proto.RaftProtos.LogEntryProto; +import org.apache.ratis.proto.RaftProtos.RaftPeerRole; import org.apache.ratis.protocol.RaftClientReply; import org.apache.ratis.protocol.RaftClientRequest; import org.apache.ratis.protocol.RaftGroup; @@ -54,6 +55,7 @@ import java.io.IOException; import java.util.ArrayList; import java.util.Arrays; +import java.util.Collection; import java.util.Collections; import java.util.List; import java.util.concurrent.CountDownLatch; @@ -65,6 +67,7 @@ import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicReference; +import static org.apache.ratis.server.impl.RaftServerTestUtil.waitAndCheckAddedPeers; import static org.apache.ratis.server.impl.RaftServerTestUtil.waitAndCheckNewConf; import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertFalse; @@ -280,6 +283,43 @@ private void runTestSetConfigurationInAddMode(CLUSTER cluster) throws Exception cluster.close(); } + @Test + public void testAddModePreservesExistingListeners() throws Exception { + runWithNewCluster(2, 1, this::runTestAddModePreservesExistingListeners); + } + + private void runTestAddModePreservesExistingListeners(CLUSTER cluster) throws Exception { + final RaftServer.Division leader = RaftTestUtil.waitForLeader(cluster); + final Collection currentPeers = leader.getRaftConf().getCurrentPeers(); + final Collection listeners = leader.getRaftConf().getCurrentPeers(RaftPeerRole.LISTENER); + + Assertions.assertEquals(1, listeners.size()); + + final PeerChanges change = cluster.addNewPeers(1, true); + final List addedPeers = change.getAddedPeers(); + final List newPeers = new ArrayList<>(currentPeers); + newPeers.addAll(addedPeers); + + try (final RaftClient client = cluster.createClient(leader.getId())) { + for (int i = 0; i < 10; i++) { + final RaftClientReply reply = client.io().send(new SimpleMessage("m" + i)); + Assertions.assertTrue(reply.isSuccess()); + } + + final RaftClientReply reply = client.admin().setConfiguration( + SetConfigurationRequest.Arguments.newBuilder() + .setServersInNewConf(addedPeers) + .setMode(SetConfigurationRequest.Mode.ADD) + .build()); + Assertions.assertTrue(reply.isSuccess()); + + final RaftConfigurationImpl expectedConf = RaftConfigurationImpl.newBuilder() + .setConf(newPeers, listeners) + .build(); + waitAndCheckAddedPeers(cluster, expectedConf, 1); + } + } + @Test public void testSetConfigurationInCasMode() throws Exception { runWithNewCluster(2, this::runTestSetConfigurationInCasMode); diff --git a/ratis-server/src/test/java/org/apache/ratis/server/impl/RaftServerTestUtil.java b/ratis-server/src/test/java/org/apache/ratis/server/impl/RaftServerTestUtil.java index 4c9d1d7bfb..3e2982e0f3 100644 --- a/ratis-server/src/test/java/org/apache/ratis/server/impl/RaftServerTestUtil.java +++ b/ratis-server/src/test/java/org/apache/ratis/server/impl/RaftServerTestUtil.java @@ -130,6 +130,41 @@ private static void waitAndCheckNewConf(MiniRaftCluster cluster, Assertions.assertEquals(peers.size(), numIncluded + deadIncluded); } + static void waitAndCheckAddedPeers(MiniRaftCluster cluster, + RaftConfigurationImpl expectedConf, int numOfNewPeers) throws Exception { + final TimeDuration sleepTime = cluster.getTimeoutMax().apply(n -> n * (numOfNewPeers + 2)); + JavaUtils.attempt(() -> waitAndCheckAddedPeers(cluster, expectedConf), + 10, sleepTime, "waitAndCheckAddedPeers", LOG); + } + + private static void waitAndCheckAddedPeers(MiniRaftCluster cluster, + RaftConfigurationImpl expectedConf) { + Assertions.assertNotNull(cluster.getLeader()); + + final Collection expectedPeers = expectedConf.getCurrentPeers(); + final Collection expectedListeners = + expectedConf.getCurrentPeers(RaftProtos.RaftPeerRole.LISTENER); + int numIncluded = 0; + + for (RaftServer.Division d : cluster.iterateDivisions()) { + final RaftServerImpl server = (RaftServerImpl)d; + final RaftConfigurationImpl actual = server.getState().getRaftConf(); + + if (expectedConf.containsInConf( + server.getId(), RaftProtos.RaftPeerRole.FOLLOWER, RaftProtos.RaftPeerRole.LISTENER)) { + Assertions.assertTrue(actual.isStable(), + () -> server.getId() + ": configuration is transitional: " + actual); + Assertions.assertTrue(actual.hasNoChange(expectedPeers, expectedListeners), + () -> "configuration mismatch for " + server.getId() + + ": expected=" + expectedConf + ", actual=" + actual); + numIncluded++; + } + } + + Assertions.assertEquals(expectedPeers.size() + expectedListeners.size(), numIncluded, + "Number of expected members checked"); + } + public static long getNextIndex(RaftServer.Division server) { return ((RaftServerImpl)server).getState().getNextIndex(); }