/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.apache.ratis.server.impl;

import org.apache.ratis.BaseTest;
import org.apache.ratis.RaftTestUtil;
import org.apache.ratis.RaftTestUtil.SimpleMessage;
import org.apache.ratis.client.RaftClient;
import org.apache.ratis.client.RaftClientRpc;
import org.apache.ratis.proto.RaftProtos.LogEntryProto;
import org.apache.ratis.protocol.RaftClientReply;
import org.apache.ratis.protocol.RaftClientRequest;
import org.apache.ratis.protocol.RaftGroup;
import org.apache.ratis.protocol.RaftGroupId;
import org.apache.ratis.protocol.RaftPeer;
import org.apache.ratis.protocol.RaftPeerId;
import org.apache.ratis.protocol.SetConfigurationRequest;
import org.apache.ratis.protocol.exceptions.SetConfigurationException;
import org.apache.ratis.protocol.exceptions.LeaderNotReadyException;
import org.apache.ratis.protocol.exceptions.ReconfigurationInProgressException;
import org.apache.ratis.protocol.exceptions.ReconfigurationTimeoutException;
import org.apache.ratis.server.RaftConfiguration;
import org.apache.ratis.server.RaftServer;
import org.apache.ratis.server.RaftServerConfigKeys;
import org.apache.ratis.server.impl.MiniRaftCluster.PeerChanges;
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.RaftStorageTestUtils;
import org.apache.ratis.util.JavaUtils;
import org.apache.ratis.util.LifeCycle;
import org.apache.ratis.util.Slf4jUtils;
import org.apache.ratis.util.TimeDuration;
import org.junit.Assert;
import org.junit.Test;
import org.slf4j.event.Level;

import java.io.IOException;
import java.util.*;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicReference;

import static java.util.Arrays.asList;
import static org.apache.ratis.server.impl.RaftServerTestUtil.waitAndCheckNewConf;
import static org.junit.Assert.assertThrows;

public abstract class RaftReconfigurationBaseTest<CLUSTER extends MiniRaftCluster>
    extends BaseTest
    implements MiniRaftCluster.Factory.Get<CLUSTER> {
  static {
    Slf4jUtils.setLogLevel(RaftServer.Division.LOG, Level.DEBUG);
  }

  private static final DelayLocalExecutionInjection logSyncDelay = RaftServerTestUtil.getLogSyncDelay();
  private static final DelayLocalExecutionInjection leaderPlaceHolderDelay =
      new DelayLocalExecutionInjection(LeaderStateImpl.APPEND_PLACEHOLDER);

  static final int STAGING_CATCHUP_GAP = 10;

  {
    RaftServerConfigKeys.setStagingCatchupGap(getProperties(), STAGING_CATCHUP_GAP);
  }

  private void checkPriority(CLUSTER cluster, RaftGroupId groupId, List<RaftPeer> peersWithPriority)
      throws InterruptedException {
    RaftTestUtil.waitForLeader(cluster, groupId);

    for (int i = 0; i < peersWithPriority.size(); i ++) {
      RaftPeerId peerId = peersWithPriority.get(i).getId();
      final RaftServer.Division server = cluster.getDivision(peerId, groupId);
      final RaftConfiguration conf = server.getRaftConf();

      for (int j = 0; j < peersWithPriority.size(); j ++) {
        int priorityInConf = conf.getPeer(peersWithPriority.get(j).getId()).getPriority();
        Assert.assertEquals(priorityInConf, peersWithPriority.get(j).getPriority());
      }
    }
  }

  @Test
  public void testRestorePriority() throws Exception {
    runWithNewCluster(3, cluster -> {
      // Add groups
      List<RaftPeer> peers = cluster.getPeers();

      List<RaftPeer> peersWithPriority = new ArrayList<>();
      for (int i = 0; i < peers.size(); i++) {
        RaftPeer peer = peers.get(i);
        peersWithPriority.add(
            RaftPeer.newBuilder(peer).setPriority(i).build());
      }

      final RaftGroup newGroup = RaftGroup.valueOf(RaftGroupId.randomId(), peersWithPriority);
      LOG.info("add new group: " + newGroup);
      try (final RaftClient client = cluster.createClient(newGroup)) {
        for (RaftPeer p : newGroup.getPeers()) {
          client.getGroupManagementApi(p.getId()).add(newGroup);
        }
      }

      RaftGroupId groupId = newGroup.getGroupId();

      checkPriority(cluster, groupId, peersWithPriority);

      cluster.restart(false);

      checkPriority(cluster, groupId, peersWithPriority);
    });
  }

  /**
   * add 2 new peers (3 peers -> 5 peers), no leader change
   */
  @Test
  public void testAddPeers() throws Exception {
    runWithNewCluster(3, cluster -> {
      RaftTestUtil.waitForLeader(cluster);

      // add new peers
      RaftPeer[] allPeers = cluster.addNewPeers(2, true).allPeersInNewConf;

      // trigger setConfiguration
      cluster.setConfiguration(allPeers);

      // wait for the new configuration to take effect
      waitAndCheckNewConf(cluster, allPeers, 0, null);
    });
  }

  /**
   * Test leader election when changing cluster from single mode to HA mode.
   */
  @Test
  public void testLeaderElectionWhenChangeFromSingleToHA() throws Exception {
    runWithNewCluster(1, cluster -> {
      RaftTestUtil.waitForLeader(cluster);

      RaftGroupId groupId = cluster.getGroup().getGroupId();
      RaftPeer curPeer = cluster.getGroup().getPeers().iterator().next();
      RaftPeer newPeer = cluster.addNewPeers(1, true, true).newPeers[0];

      RaftServerProxy leaderServer = cluster.getServer(curPeer.getId());

      // Update leader conf to transitional single mode.
      RaftConfigurationImpl oldNewConf = RaftConfigurationImpl.newBuilder()
          .setOldConf(new PeerConfiguration(Arrays.asList(curPeer)))
          .setConf(new PeerConfiguration(Arrays.asList(curPeer, newPeer)))
          .setLogEntryIndex(Long.MAX_VALUE / 2)
          .build();
      Assert.assertTrue(oldNewConf.isSingleMode(curPeer.getId()));
      RaftServerTestUtil.setRaftConf(leaderServer, groupId, oldNewConf);
      try(RaftClient client = cluster.createClient()) {
        client.admin().transferLeadership(null, leaderServer.getId(), 1000);
      }

      final RaftServer.Division newLeader = RaftTestUtil.waitForLeader(cluster);
      Assert.assertEquals(leaderServer.getId(), newLeader.getId());
      Assert.assertEquals(oldNewConf, newLeader.getRaftConf());
    });
  }

  @Test
  public void testChangeMajority() throws Exception {
    runWithNewCluster(1, cluster -> {
      RaftTestUtil.waitForLeader(cluster);
      final RaftPeerId leaderId = cluster.getLeader().getId();

      try (final RaftClient client = cluster.createClient(leaderId)) {
        final PeerChanges c1 = cluster.addNewPeers(2, true);

        SetConfigurationRequest.Arguments arguments = SetConfigurationRequest.Arguments.newBuilder()
            .setServersInCurrentConf(cluster.getPeers())
            .setServersInNewConf(c1.allPeersInNewConf)
            .setMode(SetConfigurationRequest.Mode.COMPARE_AND_SET)
            .build();
        assertThrows("Expect change majority error.", SetConfigurationException.class,
            () -> client.admin().setConfiguration(arguments));
      }
    });
  }

  /**
   * remove 2 peers (5 peers -> 3 peers), no leader change
   */
  @Test
  public void testRemovePeers() throws Exception {
    runWithNewCluster(5, cluster -> {
      RaftTestUtil.waitForLeader(cluster);

      // remove peers, leader still included in the new conf
      RaftPeer[] allPeers = cluster
          .removePeers(2, false, Collections.emptyList()).allPeersInNewConf;

      // trigger setConfiguration
      cluster.setConfiguration(allPeers);

      // wait for the new configuration to take effect
      waitAndCheckNewConf(cluster, allPeers, 2, 2, null);
    });
  }

  /**
   * 5 peers -> 5 peers, remove 2 old, add 2 new, no leader change
   */
  @Test
  public void testAddRemovePeers() throws Exception {
    runWithNewCluster(5, cluster -> runTestAddRemovePeers(false, cluster));
  }

  @Test
  public void testLeaderStepDown() throws Exception {
    runWithNewCluster(5, cluster -> runTestAddRemovePeers(true, cluster));
  }

  private void runTestAddRemovePeers(boolean leaderStepdown, CLUSTER cluster) throws Exception {
      RaftTestUtil.waitForLeader(cluster);

      PeerChanges change = cluster.addNewPeers(2, true);
      RaftPeer[] allPeers = cluster.removePeers(2, leaderStepdown,
          asList(change.newPeers)).allPeersInNewConf;

      // trigger setConfiguration
      cluster.setConfiguration(allPeers);

      // wait for the new configuration to take effect
      waitAndCheckNewConf(cluster, allPeers, 2, null);
  }

  @Test
  public void testSetConfigurationInAddMode() throws Exception {
    runWithNewCluster(2, this::runTestSetConfigurationInAddMode);
  }

  private void runTestSetConfigurationInAddMode(CLUSTER cluster) throws Exception {
    final RaftServer.Division leader = RaftTestUtil.waitForLeader(cluster);

    PeerChanges change = cluster.addNewPeers(1, true);
    List<RaftPeer> peers = Arrays.asList(change.newPeers);

    try (final RaftClient client = cluster.createClient(leader.getId())) {
      for (int i = 0; i < 10; i++) {
        RaftClientReply reply = client.io().send(new SimpleMessage("m" + i));
        Assert.assertTrue(reply.isSuccess());
      }
      RaftClientReply reply = client.admin().setConfiguration(
          SetConfigurationRequest.Arguments.newBuilder()
          .setServersInNewConf(peers)
          .setMode(SetConfigurationRequest.Mode.ADD).build());
      Assert.assertTrue(reply.isSuccess());
      waitAndCheckNewConf(cluster, change.allPeersInNewConf, 0, null);
    }
    cluster.close();
  }

  @Test
  public void testSetConfigurationInCasMode() throws Exception {
    runWithNewCluster(2, this::runTestSetConfigurationInCasMode);
  }

  private void runTestSetConfigurationInCasMode(CLUSTER cluster) throws Exception {
    final RaftServer.Division leader = RaftTestUtil.waitForLeader(cluster);
    List<RaftPeer> oldPeers = cluster.getPeers();

    PeerChanges change = cluster.addNewPeers(1, true);
    List<RaftPeer> peers = Arrays.asList(change.allPeersInNewConf);

    try (final RaftClient client = cluster.createClient(leader.getId())) {
      for (int i = 0; i < 10; i++) {
        RaftClientReply reply = client.io().send(new SimpleMessage("m" + i));
        Assert.assertTrue(reply.isSuccess());
      }

      testFailureCase("Can't set configuration in CAS mode ",
          () -> client.admin().setConfiguration(SetConfigurationRequest.Arguments.newBuilder()
              .setServersInNewConf(peers)
              .setServersInCurrentConf(peers)
              .setMode(SetConfigurationRequest.Mode.COMPARE_AND_SET)
              .build()), SetConfigurationException.class);

      Collections.shuffle(oldPeers);
      RaftClientReply reply = client.admin().setConfiguration(
          SetConfigurationRequest.Arguments.newBuilder()
              .setServersInNewConf(peers)
              .setServersInCurrentConf(oldPeers)
              .setMode(SetConfigurationRequest.Mode.COMPARE_AND_SET)
              .build());
      Assert.assertTrue(reply.isSuccess());
      waitAndCheckNewConf(cluster, change.allPeersInNewConf, 0, null);
    }
    cluster.close();
  }


  @Test(timeout = 30000)
  public void testReconfTwice() throws Exception {
    runWithNewCluster(3, this::runTestReconfTwice);
  }

  void runTestReconfTwice(CLUSTER cluster) throws Exception {
      final RaftPeerId leaderId = RaftTestUtil.waitForLeader(cluster).getId();
      try (final RaftClient client = cluster.createClient(leaderId)) {

        // submit some msgs before reconf
        for (int i = 0; i < STAGING_CATCHUP_GAP * 2; i++) {
          RaftClientReply reply = client.io().send(new SimpleMessage("m" + i));
          Assert.assertTrue(reply.isSuccess());
        }

        final AtomicBoolean reconf1 = new AtomicBoolean(false);
        final AtomicBoolean reconf2 = new AtomicBoolean(false);
        final AtomicReference<RaftPeer[]> finalPeers = new AtomicReference<>(null);
        final AtomicReference<RaftPeer[]> deadPeers = new AtomicReference<>(null);
        CountDownLatch latch = new CountDownLatch(1);
        Thread clientThread = new Thread(() -> {
          try {
            PeerChanges c1 = cluster.addNewPeers(2, true, true);
            LOG.info("Start changing the configuration: {}",
                    asList(c1.allPeersInNewConf));

            RaftClientReply reply = client.admin().setConfiguration(c1.allPeersInNewConf);
            reconf1.set(reply.isSuccess());

            PeerChanges c2 = cluster.removePeers(2, true, asList(c1.newPeers));
            finalPeers.set(c2.allPeersInNewConf);
            deadPeers.set(c2.removedPeers);

            LOG.info("Start changing the configuration again: {}",
                    asList(c2.allPeersInNewConf));
            reply = client.admin().setConfiguration(c2.allPeersInNewConf);
            reconf2.set(reply.isSuccess());

            latch.countDown();
          } catch (Exception ignored) {
            LOG.warn("{} is ignored", JavaUtils.getClassSimpleName(ignored.getClass()), ignored);
          }
        });
        clientThread.start();

        latch.await();
        Assert.assertTrue(reconf1.get());
        Assert.assertTrue(reconf2.get());
        waitAndCheckNewConf(cluster, finalPeers.get(), 2, null);
        final RaftPeerId leader2 = RaftTestUtil.waitForLeader(cluster).getId();

        // check configuration manager's internal state
        // each reconf will generate two configurations: (old, new) and (new)
        cluster.getServerAliveStream().forEach(server -> {
          final ConfigurationManager confManager = RaftServerTestUtil.getConfigurationManager(server);
          // each reconf will generate two configurations: (old, new) and (new)
          // each leader change generates one configuration.
          // expectedConf = 1 (init) + 2*2 (two conf changes) + #leader
          final int expectedConf = leader2.equals(leaderId) ? 6 : 7;
          Assert.assertEquals(server.getId() + ": " + confManager, expectedConf, confManager.numOfConf());
        });
      }
  }

  @Test
  public void testReconfTimeout() throws Exception {
    // originally 3 peers
    runWithNewCluster(3, this::runTestReconfTimeout);
  }

  void runTestReconfTimeout(CLUSTER cluster) throws Exception {
    final RaftPeerId leaderId = RaftTestUtil.waitForLeader(cluster).getId();

    try (final RaftClient client = cluster.createClient(leaderId)) {
      PeerChanges c1 = cluster.addNewPeers(2, false);

      LOG.info("Start changing the configuration: {}",
          asList(c1.allPeersInNewConf));
      Assert.assertFalse(((RaftConfigurationImpl)cluster.getLeader().getRaftConf()).isTransitional());

      final RaftClientRpc sender = client.getClientRpc();
      final SetConfigurationRequest request = cluster.newSetConfigurationRequest(
          client.getId(), leaderId, c1.allPeersInNewConf);
      try {
        RaftClientReply reply = sender.sendRequest(request);
        Assert.fail("did not get expected exception " + reply.toString());
      } catch (IOException e) {
        Assert.assertTrue("Got exception " + e,
            e instanceof ReconfigurationTimeoutException);
      }

      // the two new peers have not started yet, the bootstrapping must timeout
      LOG.info(cluster.printServers());

      // resend the same request, make sure the server has correctly reset its
      // state so that we still get timeout instead of in-progress exception
      try {
        sender.sendRequest(request);
        Assert.fail("did not get expected exception");
      } catch (IOException e) {
        Assert.assertTrue("Got exception " + e,
            e instanceof ReconfigurationTimeoutException);
      }

      // start the two new peers
      LOG.info("Start new peers");
      for (RaftPeer np : c1.newPeers) {
        cluster.restartServer(np.getId(), false);
      }
      Assert.assertTrue(client.admin().setConfiguration(c1.allPeersInNewConf).isSuccess());
    }
  }

  @Test
  public void testBootstrapReconfWithSingleNodeAddOne() throws Exception {
    // originally 1 peer, add 1 more
    runWithNewCluster(1, cluster -> runTestBootstrapReconf(1, true, cluster));
  }

  @Test
  public void testBootstrapReconfWithSingleNodeAddTwo() throws Exception {
    // originally 1 peer, add 2 more
    runWithNewCluster(1, cluster -> {
      RaftTestUtil.waitForLeader(cluster);
      final RaftPeerId leaderId = cluster.getLeader().getId();

      try (final RaftClient client = cluster.createClient(leaderId)) {
        final PeerChanges c1 = cluster.addNewPeers(2, true);

        assertThrows("Expect change majority error.", SetConfigurationException.class,
            () -> client.admin().setConfiguration(c1.allPeersInNewConf));
      }
    });
  }

  @Test
  public void testBootstrapReconf() throws Exception {
    // originally 3 peers, add 2 more
    runWithNewCluster(3, cluster -> runTestBootstrapReconf(2, true, cluster));
  }

  void runTestBootstrapReconf(int numNewPeer, boolean startNewPeer, CLUSTER cluster) throws Exception {
      LOG.info("Originally {} peer(s), add {} more, startNewPeer={}",
          cluster.getNumServers(), numNewPeer, startNewPeer);
      RaftTestUtil.waitForLeader(cluster);
      final RaftPeerId leaderId = cluster.getLeader().getId();
      try (final RaftClient client = cluster.createClient(leaderId)) {

        // submit some msgs before reconf
        for (int i = 0; i < STAGING_CATCHUP_GAP * 2; i++) {
          RaftClientReply reply = client.io().send(new SimpleMessage("m" + i));
          Assert.assertTrue(reply.isSuccess());
        }

        final PeerChanges c1 = cluster.addNewPeers(numNewPeer, startNewPeer);
        LOG.info("Start changing the configuration: {}",
                asList(c1.allPeersInNewConf));
        final AtomicReference<Boolean> success = new AtomicReference<>();
        Thread clientThread = new Thread(() -> {
          try {
            RaftClientReply reply = client.admin().setConfiguration(c1.allPeersInNewConf);
            success.set(reply.isSuccess());
          } catch (IOException ioe) {
            LOG.error("FAILED", ioe);
          }
        });
        clientThread.start();

        if (startNewPeer) {
          // Make sure that set configuration is run inside the thread
          RaftTestUtil.waitFor(() -> clientThread.isAlive(), 300, 5000);
          ONE_SECOND.sleep();
          LOG.info("start new peer(s): {}", c1.newPeers);
          for(RaftPeer p : c1.newPeers) {
            cluster.restartServer(p.getId(), false);
          }
        }

        RaftTestUtil.waitFor(() -> success.get() != null && success.get(), 300, 15000);
        LOG.info(cluster.printServers());

        RaftTestUtil.waitFor(() -> cluster.getLeader() != null, 300, 5000);
        final RaftLog leaderLog = cluster.getLeader().getRaftLog();
        for (RaftPeer newPeer : c1.newPeers) {
          final RaftServer.Division d = cluster.getDivision(newPeer.getId());
          RaftTestUtil.waitFor(() -> leaderLog.getEntries(0, Long.MAX_VALUE).length
                  == d.getRaftLog().getEntries(0, Long.MAX_VALUE).length, 300, 15000);
          Assert.assertArrayEquals(leaderLog.getEntries(0, Long.MAX_VALUE),
              d.getRaftLog().getEntries(0, Long.MAX_VALUE));
        }
      }
  }

  /**
   * kill the leader before reconfiguration finishes. Make sure the client keeps
   * retrying.
   */
  @Test
  public void testKillLeaderDuringReconf() throws Exception {
    // originally 3 peers
    runWithNewCluster(3, this::runTestKillLeaderDuringReconf);
  }

  void runTestKillLeaderDuringReconf(CLUSTER cluster) throws Exception {
    final AtomicBoolean clientRunning = new AtomicBoolean(true);
    Thread clientThread = null;
    try {
      final RaftPeerId leaderId = RaftTestUtil.waitForLeader(cluster).getId();

      PeerChanges c1 = cluster.addNewPeers(1, false);
      PeerChanges c2 = cluster.removePeers(1, false, asList(c1.newPeers));

      LOG.info("Start setConf: {}", asList(c2.allPeersInNewConf));
      LOG.info(cluster.printServers());

      final CompletableFuture<Void> setConf = new CompletableFuture<>();
      clientThread = new Thread(() -> {
        try(final RaftClient client = cluster.createClient(leaderId)) {
          for(int i = 0; clientRunning.get() && !setConf.isDone(); i++) {
            final RaftClientReply reply = client.admin().setConfiguration(c2.allPeersInNewConf);
            if (reply.isSuccess()) {
              setConf.complete(null);
              return;
            }
            LOG.info("setConf attempt #{} failed, {}", i, cluster.printServers());
          }
        } catch(Exception e) {
          LOG.error("Failed to setConf", e);
          setConf.completeExceptionally(e);
        }
      });
      clientThread.start();

      // the leader cannot generate the (old, new) conf, and it will keep
      // bootstrapping the 1 new peer since it has not started yet.
      Assert.assertFalse(((RaftConfigurationImpl)cluster.getLeader().getRaftConf()).isTransitional());

      // (0) the first conf entry, (1) the 1st setConf entry, (2) a metadata entry
      // (3) new current conf entry  (4) a metadata entry
      {
        final RaftLog leaderLog = cluster.getLeader().getRaftLog();
        for(LogEntryProto e : RaftTestUtil.getLogEntryProtos(leaderLog)) {
          LOG.info("{}", LogProtoUtils.toLogEntryString(e));
        }
        final long commitIndex = leaderLog.getLastCommittedIndex();
        Assert.assertTrue("commitIndex = " + commitIndex + " > 2", commitIndex <= 2);
      }

      final RaftPeerId killed = RaftTestUtil.waitAndKillLeader(cluster);
      Assert.assertEquals(leaderId, killed);
      final RaftPeerId newLeaderId = RaftTestUtil.waitForLeader(cluster).getId();
      LOG.info("newLeaderId: {}", newLeaderId);
      TimeDuration.valueOf(1500, TimeUnit.MILLISECONDS).sleep();

      LOG.info("start new peers: {}", Arrays.asList(c1.newPeers));
      for (RaftPeer np : c1.newPeers) {
        cluster.restartServer(np.getId(), false);
      }

      try {
        setConf.get(10, TimeUnit.SECONDS);
      } catch(TimeoutException ignored) {
      }

      RaftServerProxy newServer = cluster.getServer(c1.newPeers[0].getId());
      if (newServer.getLifeCycleState() == LifeCycle.State.CLOSED) {
        LOG.info("New peer {} is shutdown. Skip the check", c1.newPeers[0].getId());
      } else {
        // the client fails with the first leader, and then retry the same setConfiguration request
        waitAndCheckNewConf(cluster, c2.allPeersInNewConf, 1, Collections.singletonList(leaderId));
        setConf.get(2, TimeUnit.SECONDS);
      }
    } finally {
      if (clientThread != null) {
        clientRunning.set(false);
        clientThread.interrupt();
      }
    }
  }

  /**
   * When a request's new configuration is the same with the current one, make
   * sure we return success immediately and no log entry is recorded.
   */
  @Test
  public void testNoChangeRequest() throws Exception {
    // originally 3 peers
    runWithNewCluster(3, this::runTestNoChangeRequest);
  }

  void runTestNoChangeRequest(CLUSTER cluster) throws Exception {
    final RaftServer.Division leader = RaftTestUtil.waitForLeader(cluster);
    try(final RaftClient client = cluster.createClient(leader.getId())) {
      client.io().send(new SimpleMessage("m"));

      final RaftLog leaderLog = leader.getRaftLog();
      final long committedIndex = leaderLog.getLastCommittedIndex();
      final RaftConfiguration confBefore = cluster.getLeader().getRaftConf();

      // no real configuration change in the request
      final RaftClientReply reply = client.admin().setConfiguration(cluster.getPeers().toArray(RaftPeer.emptyArray()));
      Assert.assertTrue(reply.isSuccess());
      final long newCommittedIndex = leaderLog.getLastCommittedIndex();
      for(long i = committedIndex + 1; i <= newCommittedIndex; i++) {
        final LogEntryProto e = leaderLog.get(i);
        Assert.assertTrue(e.hasMetadataEntry());
      }
      Assert.assertSame(confBefore, cluster.getLeader().getRaftConf());
    }
  }

  /**
   * Make sure a setConfiguration request is rejected if a configuration change
   * is still in progress (i.e., has not been committed yet).
   */
  @Test
  public void testOverlappedSetConfRequests() throws Exception {
    // Make sure leadership check won't affect the test.
    // logSyncDelay.setDelayMs() blocks RaftServerImpl::appendEntriesAsync(),
    // which will disable heartbeats.
    final TimeDuration oldTimeoutMin = RaftServerConfigKeys.Rpc.timeoutMin(getProperties());
    RaftServerConfigKeys.Rpc.setTimeoutMin(getProperties(), TimeDuration.valueOf(2, TimeUnit.SECONDS));
    final TimeDuration oldTimeoutMax = RaftServerConfigKeys.Rpc.timeoutMax(getProperties());
    RaftServerConfigKeys.Rpc.setTimeoutMax(getProperties(), TimeDuration.valueOf(4, TimeUnit.SECONDS));

    // originally 3 peers
    runWithNewCluster(3, this::runTestOverlappedSetConfRequests);

    // reset for the other tests
    RaftServerConfigKeys.Rpc.setTimeoutMin(getProperties(), oldTimeoutMin);
    RaftServerConfigKeys.Rpc.setTimeoutMax(getProperties(), oldTimeoutMax);
  }

  void runTestOverlappedSetConfRequests(CLUSTER cluster) throws Exception {
    try {
      RaftTestUtil.waitForLeader(cluster);

      final RaftPeerId leaderId = cluster.getLeader().getId();

      RaftPeer[] newPeers = cluster.addNewPeers(2, true).allPeersInNewConf;

      // delay every peer's logSync so that the setConf request is delayed
      cluster.getPeers()
          .forEach(peer -> logSyncDelay.setDelayMs(peer.getId().toString(), 1000));

      final CountDownLatch latch = new CountDownLatch(1);
      final RaftPeer[] peersInRequest2 = cluster.getPeers().toArray(new RaftPeer[0]);
      AtomicBoolean caughtException = new AtomicBoolean(false);
      new Thread(() -> {
        try(final RaftClient client2 = cluster.createClient(leaderId)) {
          latch.await();
          LOG.info("client2 starts to change conf");
          final RaftClientRpc sender2 = client2.getClientRpc();
          sender2.sendRequest(cluster.newSetConfigurationRequest(
              client2.getId(), leaderId, peersInRequest2));
        } catch (ReconfigurationInProgressException e) {
          caughtException.set(true);
        } catch (Exception e) {
          LOG.warn("Got unexpected exception when client2 changes conf", e);
        }
      }).start();

      AtomicBoolean confChanged = new AtomicBoolean(false);
      new Thread(() -> {
        try(final RaftClient client1 = cluster.createClient(leaderId)) {
          LOG.info("client1 starts to change conf");
          confChanged.set(client1.admin().setConfiguration(newPeers).isSuccess());
        } catch (IOException e) {
          LOG.warn("Got unexpected exception when client1 changes conf", e);
        }
      }).start();
      Thread.sleep(100);
      latch.countDown();

      for (int i = 0; i < 10 && !confChanged.get(); i++) {
        Thread.sleep(1000);
      }
      Assert.assertTrue(confChanged.get());
      Assert.assertTrue(caughtException.get());
    } finally {
      logSyncDelay.clear();
    }
  }

  /**
   * Test a scenario where the follower truncates its log entries which causes
   * configuration change.
   */
  @Test
  public void testRevertConfigurationChange() throws Exception {
    runWithNewCluster(5, this::runTestRevertConfigurationChange);
  }

  void runTestRevertConfigurationChange(CLUSTER cluster) throws Exception {
    RaftLogBase log2 = null;
    try {
      RaftTestUtil.waitForLeader(cluster);

      final RaftServer.Division leader = cluster.getLeader();
      final RaftPeerId leaderId = leader.getId();

      final RaftLog log = leader.getRaftLog();
      log2 = (RaftLogBase) log;
      Thread.sleep(1000);

      // we block the incoming msg for the leader and block its requests to
      // followers, so that we force the leader change and the old leader will
      // not know
      LOG.info("start blocking the leader");
      BlockRequestHandlingInjection.getInstance().blockReplier(leaderId.toString());
      cluster.setBlockRequestsFrom(leaderId.toString(), true);

      PeerChanges change = cluster.removePeers(1, false, new ArrayList<>());

      AtomicBoolean gotNotLeader = new AtomicBoolean(false);
      final Thread clientThread = new Thread(() -> {
        try(final RaftClient client = cluster.createClient(leaderId)) {
          LOG.info("client starts to change conf");
          final RaftClientRpc sender = client.getClientRpc();
          RaftClientReply reply = sender.sendRequest(cluster.newSetConfigurationRequest(
              client.getId(), leaderId, change.allPeersInNewConf));
          if (reply.getNotLeaderException() != null) {
            gotNotLeader.set(true);
          }
        } catch (IOException e) {
          LOG.warn("Got unexpected exception when client1 changes conf", e);
        }
      });
      clientThread.start();

      // find ConfigurationEntry
      final TimeDuration sleepTime = TimeDuration.valueOf(500, TimeUnit.MILLISECONDS);
      final long confIndex = JavaUtils.attemptRepeatedly(() -> {
        final long last = log.getLastEntryTermIndex().getIndex();
        for (long i = last; i >= 1; i--) {
          if (log.get(i).hasConfigurationEntry()) {
            return i;
          }
        }
        throw new Exception("ConfigurationEntry not found: last=" + last);
      }, 10, sleepTime, "confIndex", LOG);

      // wait till the old leader persist the new conf
      JavaUtils.attemptRepeatedly(() -> {
        Assert.assertTrue(log.getFlushIndex() >= confIndex);
        return null;
      }, 10, sleepTime, "FLUSH", LOG);
      final long committed = log.getLastCommittedIndex();
      Assert.assertTrue(committed < confIndex);

      // unblock the old leader
      BlockRequestHandlingInjection.getInstance().unblockReplier(leaderId.toString());
      cluster.setBlockRequestsFrom(leaderId.toString(), false);

      // the client should get NotLeaderException
      clientThread.join(5000);
      Assert.assertTrue(gotNotLeader.get());

      // the old leader should have truncated the setConf from the log
      JavaUtils.attemptRepeatedly(() -> {
        Assert.assertTrue(log.getLastCommittedIndex() >= confIndex);
        return null;
      }, 10, ONE_SECOND, "COMMIT", LOG);
      Assert.assertTrue(log.get(confIndex).hasConfigurationEntry());
      log2 = null;
    } finally {
      RaftStorageTestUtils.printLog(log2, s -> LOG.info(s));
    }
  }

  /**
   * Delay the commit of the leader placeholder log entry and see if the client
   * can correctly receive and handle the LeaderNotReadyException.
   */
  @Test
  public void testLeaderNotReadyException() throws Exception {
    LOG.info("Start testLeaderNotReadyException");
    final MiniRaftCluster cluster = newCluster(1).initServers();
    try {
      // delay 1s for each logSync call
      cluster.getServers().forEach(
          peer -> leaderPlaceHolderDelay.setDelayMs(peer.getId().toString(), 2000));
      cluster.start();

      AtomicBoolean caughtNotReady = new AtomicBoolean(false);
      AtomicBoolean success = new AtomicBoolean(false);
      final RaftPeerId leaderId = cluster.getPeers().iterator().next().getId();
      new Thread(() -> {
        try (final RaftClient client = cluster.createClient(leaderId)) {
          final RaftClientRpc sender = client.getClientRpc();
          final RaftClientRequest request = cluster.newRaftClientRequest(
                  client.getId(), leaderId, new SimpleMessage("test"));
          while (!success.get()) {
            try {
              final RaftClientReply reply = sender.sendRequest(request);
              success.set(reply.isSuccess());
              if (reply.getException() != null && reply.getException() instanceof LeaderNotReadyException) {
                caughtNotReady.set(true);
              }
            } catch (IOException e) {
              LOG.info("Hit other IOException", e);
            }
            if (!success.get()) {
              try {
                Thread.sleep(200);
              } catch (InterruptedException ignored) {
                Thread.currentThread().interrupt();
              }
            }
          }
        } catch (Exception ignored) {
          LOG.warn("{} is ignored", JavaUtils.getClassSimpleName(ignored.getClass()), ignored);
        }
      }).start();

      RaftTestUtil.waitForLeader(cluster);
      for (int i = 0; !success.get() && i < 5; i++) {
        Thread.sleep(1000);
      }
      Assert.assertTrue(success.get());
      Assert.assertTrue(caughtNotReady.get());
    } finally {
      leaderPlaceHolderDelay.clear();
      cluster.shutdown();
    }
  }
}