RATIS-2082. setConfiguration should not ignore address changes of existing peers

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

@@ -302,7 +302,18 @@ public String toString() {
     return "conf: {index: " + logEntryIndex + ", cur=" + conf + ", old=" + oldConf + "}";
   }
 
+  /** @return true if the ids and the priorities are the same; the addresses are not compared. */
   boolean hasNoChange(Collection<RaftPeer> newMembers, Collection<RaftPeer> newListeners) {
+    return hasNoChange(newMembers, newListeners, false);
+  }
+
+  /** @return true if the ids, the priorities and the addresses are the same. */
+  boolean hasNoChangeIncludingAddresses(Collection<RaftPeer> newMembers, Collection<RaftPeer> newListeners) {
+    return hasNoChange(newMembers, newListeners, true);
+  }
+
+  private boolean hasNoChange(Collection<RaftPeer> newMembers, Collection<RaftPeer> newListeners,
+      boolean checkAddresses) {
     if (!isStable() || conf.size() != newMembers.size()
         || conf.getPeers(RaftPeerRole.LISTENER).size() != newListeners.size()) {
       return false;
@@ -315,6 +326,9 @@ boolean hasNoChange(Collection<RaftPeer> newMembers, Collection<RaftPeer> newLis
       if (inConf.getPriority() != peer.getPriority()) {
         return false;
       }
+      if (checkAddresses && !hasSameAddresses(inConf, peer)) {
+        return false;
+      }
     }
     for (RaftPeer peer : newListeners) {
       final RaftPeer inConf = conf.getPeer(peer.getId(), RaftPeerRole.LISTENER);
@@ -324,10 +338,25 @@ boolean hasNoChange(Collection<RaftPeer> newMembers, Collection<RaftPeer> newLis
       if (inConf.getPriority() != peer.getPriority()) {
         return false;
       }
+      if (checkAddresses && !hasSameAddresses(inConf, peer)) {
+        return false;
+      }
     }
     return true;
   }
 
+  private static boolean hasSameAddresses(RaftPeer left, RaftPeer right) {
+    return isSameAddress(left.getAddress(), right.getAddress())
+        && isSameAddress(left.getAdminAddress(), right.getAdminAddress())
+        && isSameAddress(left.getClientAddress(), right.getClientAddress())
+        && isSameAddress(left.getDataStreamAddress(), right.getDataStreamAddress());
+  }
+
+  /** An unset address is null in a newly built peer but empty in a peer converted from a proto. */
+  private static boolean isSameAddress(String left, String right) {
+    return (left == null ? "" : left).equals(right == null ? "" : right);
+  }
+
   @Override
   public long getLogEntryIndex() {
     return logEntryIndex;

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

@@ -1492,7 +1492,7 @@ public CompletableFuture<RaftClientReply> setConfigurationAsync(SetConfiguration
       }
 
       // return success with a null message if the new conf is the same as the current
-      if (current.hasNoChange(serversInNewConf, listenersInNewConf)) {
+      if (current.hasNoChangeIncludingAddresses(serversInNewConf, listenersInNewConf)) {
         pending = new PendingRequest(request);
         pending.setReply(newSuccessReply(request));
         return pending.getFuture();

ratis-server/src/test/java/org/apache/ratis/server/impl/RaftReconfigurationBaseTest.java

@@ -40,10 +40,12 @@
 import org.apache.ratis.server.raftlog.LogProtoUtils;
 import org.apache.ratis.server.raftlog.RaftLog;
 import org.apache.ratis.server.raftlog.RaftLogBase;
+import org.apache.ratis.server.storage.RaftStorageImpl;
 import org.apache.ratis.server.storage.RaftStorageTestUtils;
 import org.apache.ratis.util.ConcurrentUtils;
 import org.apache.ratis.util.JavaUtils;
 import org.apache.ratis.util.LifeCycle;
+import org.apache.ratis.util.NetUtils;
 import org.apache.ratis.util.Slf4jUtils;
 import org.apache.ratis.util.TimeDuration;
 import org.junit.jupiter.api.Assertions;
@@ -64,6 +66,7 @@
 import java.util.concurrent.TimeoutException;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicReference;
+import java.util.stream.Collectors;
 
 import static org.apache.ratis.server.impl.RaftServerTestUtil.waitAndCheckNewConf;
 import static org.junit.jupiter.api.Assertions.assertEquals;
@@ -652,7 +655,7 @@ void runTestNoChangeRequest(CLUSTER cluster) throws Exception {
       final RaftConfiguration confBefore = cluster.getLeader().getRaftConf();
 
       // no real configuration change in the request
-      final RaftClientReply reply = client.admin().setConfiguration(cluster.getPeers().toArray(RaftPeer.emptyArray()));
+      final RaftClientReply reply = client.admin().setConfiguration(new ArrayList<>(confBefore.getCurrentPeers()));
       Assertions.assertTrue(reply.isSuccess());
       final long newCommittedIndex = leaderLog.getLastCommittedIndex();
       for(long i = committedIndex + 1; i <= newCommittedIndex; i++) {
@@ -663,6 +666,51 @@ void runTestNoChangeRequest(CLUSTER cluster) throws Exception {
     }
   }
 
+  /**
+   * When a request changes only the addresses of the existing peers,
+   * make sure the new configuration is committed and persisted.
+   */
+  @Test
+  public void testAddressChangeRequest() throws Exception {
+    runWithNewCluster(3, this::runTestAddressChangeRequest);
+  }
+
+  void runTestAddressChangeRequest(CLUSTER cluster) throws Exception {
+    final RaftServer.Division leader = RaftTestUtil.waitForLeader(cluster);
+    final long confIndex = leader.getRaftConf().getLogEntryIndex();
+
+    // the same peers with an equivalent address
+    final List<RaftPeer> newPeers = leader.getRaftConf().getCurrentPeers().stream()
+        .map(p -> RaftPeer.newBuilder(p).setAddress(p.getAddress().replace(NetUtils.LOCALHOST, "127.0.0.1")).build())
+        .collect(Collectors.toList());
+    try (RaftClient client = cluster.createClient(leader.getId())) {
+      assertTrue(client.admin().setConfiguration(newPeers).isSuccess());
+      assertTrue(leader.getRaftConf().getLogEntryIndex() > confIndex);
+      assertAddresses(newPeers, cluster);
+
+      // the same request again has no change
+      final RaftConfiguration conf = leader.getRaftConf();
+      assertTrue(client.admin().setConfiguration(newPeers).isSuccess());
+      Assertions.assertSame(conf, leader.getRaftConf());
+    }
+
+    cluster.restart(false);
+    RaftTestUtil.waitForLeader(cluster);
+    assertAddresses(newPeers, cluster);
+  }
+
+  void assertAddresses(List<RaftPeer> expected, MiniRaftCluster cluster) throws Exception {
+    JavaUtils.attempt(() -> {
+      for (RaftServer.Division d : cluster.iterateDivisions()) {
+        final RaftConfiguration persisted = ((RaftStorageImpl) d.getRaftStorage()).readRaftConfiguration();
+        for (RaftPeer p : expected) {
+          assertEquals(p.getAddress(), d.getRaftConf().getPeer(p.getId()).getAddress());
+          assertEquals(p.getAddress(), persisted.getPeer(p.getId()).getAddress());
+        }
+      }
+    }, 10, ONE_SECOND, "assertAddresses", LOG);
+  }
+
   /**
    * Make sure a setConfiguration request is rejected if a configuration change
    * is still in progress (i.e., has not been committed yet).

ratis-test/src/test/java/org/apache/ratis/server/impl/TestRaftConfiguration.java

@@ -26,6 +26,9 @@
 
 import java.util.Arrays;
 import java.util.Collection;
+import java.util.Collections;
+import java.util.List;
+import java.util.function.UnaryOperator;
 import java.util.stream.Collectors;
 
 import static org.junit.jupiter.api.Assertions.assertFalse;
@@ -144,4 +147,35 @@ public void testChangeMajority() {
         .build();
     assertFalse(config.changeMajority(raftPeersWithPriority(1, 2, 3, 4, 6, 7)));
   }
-}
+
+  @Test
+  public void testHasNoChangeIncludingAddresses() {
+    final RaftPeer peer = RaftPeer.newBuilder().setId("s0").setAddress("h:1").build();
+    final RaftPeer listener = RaftPeer.newBuilder().setId("s1").setAddress("h:2").build();
+    final List<RaftPeer> peers = Collections.singletonList(peer);
+    final List<RaftPeer> listeners = Collections.singletonList(listener);
+    final RaftConfigurationImpl config = RaftConfigurationImpl.newBuilder()
+        .setConf(new PeerConfiguration(peers, listeners))
+        .build();
+    assertTrue(config.hasNoChangeIncludingAddresses(peers, listeners));
+
+    // an unset address is the same as an empty address
+    final RaftPeer empty = RaftPeer.newBuilder(peer)
+        .setAdminAddress("").setClientAddress("").setDataStreamAddress("").build();
+    assertTrue(config.hasNoChangeIncludingAddresses(Collections.singletonList(empty), listeners));
+
+    final List<UnaryOperator<RaftPeer.Builder>> changes = Arrays.asList(
+        b -> b.setAddress("h:9"), b -> b.setAdminAddress("h:9"),
+        b -> b.setClientAddress("h:9"), b -> b.setDataStreamAddress("h:9"));
+    for (UnaryOperator<RaftPeer.Builder> change : changes) {
+      final List<RaftPeer> newPeers = Collections.singletonList(change.apply(RaftPeer.newBuilder(peer)).build());
+      assertTrue(config.hasNoChange(newPeers, listeners));
+      assertFalse(config.hasNoChangeIncludingAddresses(newPeers, listeners));
+
+      final List<RaftPeer> newListeners = Collections.singletonList(
+          change.apply(RaftPeer.newBuilder(listener)).build());
+      assertTrue(config.hasNoChange(peers, newListeners));
+      assertFalse(config.hasNoChangeIncludingAddresses(peers, newListeners));
+    }
+  }
+}