RATIS-2806. Preserve existing listeners in setConfiguration ADD mode

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<RaftPeer> peersInNewConf) {
+  PendingRequest startSetConfiguration(SetConfigurationRequest request,
+      List<RaftPeer> peersInNewConf, List<RaftPeer> 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<RaftPeer> listenersInNewConf = request.getArguments().getPeersInNewConf(RaftPeerRole.LISTENER);
     final Collection<RaftPeer> peersToBootStrap = server.getRaftConf().filterNotContainedInConf(peersInNewConf);
     final Collection<RaftPeer> listenersToBootStrap= server.getRaftConf().filterNotContainedInConf(listenersInNewConf);
 

ratis-server/src/main/java/org/apache/ratis/server/impl/RaftServerImpl.java

@@ -1413,7 +1413,7 @@ public CompletableFuture<RaftClientReply> 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();
   }

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<Object> infoIndexChange = s -> LOG.info("{}: {}", getName(), s);
   private final Consumer<Object> 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

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<RaftPeer> currentPeers = leader.getRaftConf().getCurrentPeers();
+    final Collection<RaftPeer> listeners = leader.getRaftConf().getCurrentPeers(RaftPeerRole.LISTENER);
+
+    Assertions.assertEquals(1, listeners.size());
+
+    final PeerChanges change = cluster.addNewPeers(1, true);
+    final List<RaftPeer> addedPeers = change.getAddedPeers();
+    final List<RaftPeer> 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);

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<RaftPeer> expectedPeers = expectedConf.getCurrentPeers();
+    final Collection<RaftPeer> 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();
   }