public final class QuorumConfiguration {

  public final boolean isTransitional;

  private final Set<Long> allPeers;
  private final Set<Long> prevPeers;
  private final Set<Long> nextPeers;

  public static final QuorumConfiguration EMPTY = new QuorumConfiguration(new HashSet<>());

  public static QuorumConfiguration of(Collection<Long> peerCollection) {
    return new QuorumConfiguration(peerCollection);
  }

  public static QuorumConfiguration fromProtostuff(c5db.replication.generated.QuorumConfigurationMessage message) {
    if (message.getTransitional()) {
      return new QuorumConfiguration(message.getPrevPeersList(), message.getNextPeersList());
    } else {
      return new QuorumConfiguration(message.getAllPeersList());
    }
  }

  public QuorumConfiguration getTransitionalConfiguration(Collection<Long> newPeerCollection) {
    if (isTransitional) {
      return new QuorumConfiguration(prevPeers, newPeerCollection);
    } else {
      return new QuorumConfiguration(allPeers, newPeerCollection);
    }
  }

  public QuorumConfiguration getCompletedConfiguration() {
    assert isTransitional;

    return new QuorumConfiguration(nextPeers);
  }

  public QuorumConfigurationMessage toProtostuff() {
    return new QuorumConfigurationMessage(
        isTransitional,
        Lists.newArrayList(allPeers),
        Lists.newArrayList(prevPeers),
        Lists.newArrayList(nextPeers));
  }

  public Set<Long> allPeers() {
    return allPeers;
  }

  public Set<Long> prevPeers() {
    return prevPeers;
  }

  public Set<Long> nextPeers() {
    return nextPeers;
  }

  public boolean isEmpty() {
    return allPeers.size() == 0
        && prevPeers.size() == 0
        && nextPeers.size() == 0;
  }

  /**
   * Determine if the peers in sourceSet include a majority of the peers in this configuration.
   */
  public boolean setContainsMajority(Set<Long> sourceSet) {
    if (isTransitional) {
      return setComprisesMajorityOfAnotherSet(sourceSet, prevPeers)
          && setComprisesMajorityOfAnotherSet(sourceSet, nextPeers);
    } else {
      return setComprisesMajorityOfAnotherSet(sourceSet, allPeers);
    }
  }

  /**
   * Given a map which tells the last acknowledged entry index for different peers, find the maximum
   * index value which is less than or equal to a majority of this configuration's peers' indexes.
   */
  public long calculateCommittedIndex(Map<Long, Long> peersLastAckedIndex) {
    if (isTransitional) {
      return Math.min(
          getGreatestIndexCommittedByMajority(prevPeers, peersLastAckedIndex),
          getGreatestIndexCommittedByMajority(nextPeers, peersLastAckedIndex));
    } else {
      return getGreatestIndexCommittedByMajority(allPeers, peersLastAckedIndex);
    }
  }


  private QuorumConfiguration(Collection<Long> peers) {
    this.isTransitional = false;
    allPeers = ImmutableSet.copyOf(peers);
    prevPeers = nextPeers = ImmutableSet.of();
  }

  private QuorumConfiguration(Collection<Long> prevPeers, Collection<Long> nextPeers) {
    this.isTransitional = true;
    this.prevPeers = ImmutableSet.copyOf(prevPeers);
    this.nextPeers = ImmutableSet.copyOf(nextPeers);
    this.allPeers = Sets.union(this.prevPeers, this.nextPeers).immutableCopy();
  }


  private static long getGreatestIndexCommittedByMajority(Set<Long> peers, Map<Long, Long> peersLastAckedIndex) {
    SortedMultiset<Long> committedIndexes = TreeMultiset.create();
    committedIndexes.addAll(peers.stream().map(peerId
        -> peersLastAckedIndex.getOrDefault(peerId, 0L)).collect(Collectors.toList()));
    return Iterables.get(committedIndexes.descendingMultiset(), calculateNumericalMajority(peers.size()) - 1);
  }

  private static <T> boolean setComprisesMajorityOfAnotherSet(Set<T> sourceSet, Set<T> destinationSet) {
    return Sets.intersection(sourceSet, destinationSet).size() >= calculateNumericalMajority(destinationSet.size());
  }

  private static int calculateNumericalMajority(int setSize) {
    return (setSize / 2) + 1;
  }

  @Override
  public String toString() {
    return "QuorumConfiguration{" +
        "isTransitional=" + isTransitional +
        ", allPeers=" + allPeers +
        ", prevPeers=" + prevPeers +
        ", nextPeers=" + nextPeers +
        '}';
  }

  @Override
  public boolean equals(Object o) {
    if (this == o) {
      return true;
    }
    if (o == null || getClass() != o.getClass()) {
      return false;
    }

    QuorumConfiguration that = (QuorumConfiguration) o;

    return isTransitional == that.isTransitional
        && allPeers.equals(that.allPeers)
        && nextPeers.equals(that.nextPeers)
        && prevPeers.equals(that.prevPeers);
  }

  @Override
  public int hashCode() {
    int result = (isTransitional ? 1 : 0);
    result = 31 * result + allPeers.hashCode();
    result = 31 * result + prevPeers.hashCode();
    result = 31 * result + nextPeers.hashCode();
    return result;
  }
} public interface Replicator {
  /**
   * Return the ID of the quorum (that is, group of coordinating replicators) that this
   * replicator belongs to. For a given Replicator instance, this will always return the
   * same value.
   */
  String getQuorumId();

  /**
   * Return a future containing the Replicator's current quorum configuration; that is,
   * the set of peers it recognizes as being part of its quorum. This set can change
   * if directed to locally (for instance using the method changeQuorum) or if a remote
   * leader initiates a quorum change.
   */
  ListenableFuture<QuorumConfiguration> getQuorumConfiguration();

  /**
   * Change the members of the quorum to a new collection of peers (which may include peers in
   * the current quorum).
   *
   * @param newPeers The collection of peer IDs in the new quorum.
   * @return a future which will return the replicator receipt for logging the transitional
   * quorum configuration entry, when the entry's index known. The transitional quorum
   * configuration combines the current group of peers with the given collection of new peers.
   * When that transitional configuration is committed, the quorum configuration is guaranteed
   * to go through; prior to that commit, it is possible that a fault will cancel the quorum
   * change operation.
   * <p>
   * The actual completion of the quorum change will be signaled by a ReplicatorInstanceEvent
   * indicating the commitment of the quorum configuration consisting of the given peers,
   */
  ListenableFuture<ReplicatorReceipt> changeQuorum(Collection<Long> newPeers) throws InterruptedException;

  /**
   * Submit data to be replicated.
   * TODO we may want a variation of this method which does not block
   *
   * @param data Some data to log.
   * @return a listenable for a receipt for the log request, OR null if we aren't the leader.
   * The receipt gives information about the replication request that can be used, in conjunction
   * with commit notices, to determine if and when the request was successful.
   */
  ListenableFuture<ReplicatorReceipt> logData(List<ByteBuffer> data) throws InterruptedException;

  /**
   * @return The numerical ID for the server, or node, on which this Replicator resides. More
   * than one Replicator may have the same node ID, but any two Replicators operating at the same
   * time with the same ID will have different quorum IDs. And, considering all the Replicators
   * operating at some time within a single quorum, all must have different node IDs.
   */
  long getId();

  /**
   * Each time the Replicator changes State, it emits that State from this Subscriber.
   */
  Subscriber<State> getStateChannel();

  // What state is this instance in?
  public enum State {
    FOLLOWER,
    CANDIDATE,
    LEADER,
  }

  /**
   * The Replicator issues events from this Subscriber on various conditions. See the comments on
   * {@link c5db.interfaces.replication.ReplicatorInstanceEvent} for more information.
   */
  Subscriber<ReplicatorInstanceEvent> getEventChannel();

  /**
   * Get the Replicator's commit notice channel. By matching these issued IndexCommitNotices
   * against the ReplicatorReceipts returned when logging entries or changing quorums, users
   * of the Replicator can determine whether those submissions were successfully replicated
   * to the quorum.
   */
  Subscriber<IndexCommitNotice> getCommitNoticeChannel();
} public interface ReplicatorLog {
  /**
   * Log entries to the log.  The entries are in order, and should start from getLastIndex() + 1.
   * <p>
   * The implementation should feel free to verify this.
   * <p>
   * After this call returns, the log implementation should mark these entries as "to be committed"
   * and calls to 'getLastIndex()' should return the last entry in "entries".  The implementation will
   * then return a future that will be set true upon successful sync to durable storage, or else
   * set with an exception.
   * <p>
   * Note that over time, multiple calls to logEntries() may be issued before the prior call has signaled
   * full sync to the client.  This also implies that once this call returns, calls to the other methods
   * of this interface must now return data from these entries.  For example calling getLogTerm(long) should
   * return data from these entries even if they haven't been quite sync'ed to disk yet.
   *
   * @param entries new log entries
   * @return an future that indicates success.
   */
  ListenableFuture<Boolean> logEntries(List<LogEntry> entries);

  /**
   * Get a future which will return the entries in a specified range of indexes from start, inclusive, to end,
   * exclusive. If start and end are equal, the returned list will be empty.
   *
   * @param start the index of the low endpoint of the range (inclusive)
   * @param end   the index of the high endpoint of the range (exclusive)
   * @return a future which will yield the requested entries, or an exception if an error occurs.
   */
  ListenableFuture<List<LogEntry>> getLogEntries(long start, long end);

  /**
   * Get the term for a given log index. If the given index is not present in the log, then this
   * will return 0. A term value of 0 should be considered invalid. This is expected to be fast,
   * so it's a synchronous interface.
   *
   * @param index the index to look up the term for
   * @return the term value at 'index', or 0 if no such entry
   */
  long getLogTerm(long index);

  /**
   * Gets the term value for the last entry in the log. if the log is empty, then this will return
   * 0. A term value of 0 should never be valid.
   *
   * @return the last term or 0 if no such entry
   */
  long getLastTerm();

  /**
   * Gets the index of the most recent log entry.  An index is like a log sequence number, but there are
   * no holes.
   *
   * @return the index or 0 if the log is empty. This implies log entries start at 1.
   */
  long getLastIndex();

  /**
   * Delete all log entries after and including the specified index.
   * <p>
   * To persist the deletion, this might take a few so use a future.
   *
   * @param entryIndex the index entry to truncate log from.
   * @return A future set to true upon completion, or set with an exception upon failure.
   */
  ListenableFuture<Boolean> truncateLog(long entryIndex);

  /**
   * Get the latest, and thus current, configuration of the replicator's peers. (The configuration is
   * logged just like any other entry.)
   *
   * @return The latest configuration; or if there is none, an empty configuration.
   */
  QuorumConfiguration getLastConfiguration();

  /**
   * Get the index of the entry with the latest configuration of the replicator's peers.
   *
   * @return The greatest index of any configuration entry in the log, or zero if there is none.
   */
  long getLastConfigurationIndex();
} public class RpcRequest extends RpcMessage {

  public RpcRequest(long to, long from, String quorumId, Message message) {
    // Note that the RPC system should sub in a message id, that is an implementation detail
    // since not all transports (eg: in RAM only transport) need message IDs to keep request/replies in line.
    super(to, from, quorumId, message);
  }
} public interface CheckedConsumer<T, E extends Throwable> {

  void accept(T t) throws E;
} public class JUnitRuleFiberExceptions implements TestRule, Consumer<Throwable> {
  private final List<Throwable> throwables = Collections.synchronizedList(new ArrayList<>());

  public JUnitRuleFiberExceptions() {

  }

  @Override
  public void accept(Throwable throwable) {
    throwables.add(throwable);
  }

  @Override
  public Statement apply(final Statement base, Description description) {
    return new Statement() {
      @Override
      public void evaluate() throws Throwable {
        try {
          base.evaluate();
        } catch (Throwable t) {
          throwables.add(t);
        }
        MultipleFailureException.assertEmpty(throwables);
      }
    };
  }
} public static class ChannelHistoryMonitor<T> {
  private final List<T> messageLog = Collections.<T>synchronizedList(new ArrayList<>());
  private final Map<Matcher<? super T>, SettableFuture<T>> waitingToMatch = new HashMap<>();
  private final Fiber fiber;
  private static final int WAIT_TIMEOUT = 5; // seconds

  public ChannelHistoryMonitor(Subscriber<T> subscriber, Fiber fiber) {
    this.fiber = fiber;
    subscriber.subscribe(fiber, this::onMessage);
  }

  @FiberOnly
  private void onMessage(T message) {
    messageLog.add(message);
    Iterator<Matcher<? super T>> it = waitingToMatch.keySet().iterator();
    while (it.hasNext()) {
      Matcher<? super T> matcher = it.next();
      if (matcher.matches(message)) {
        waitingToMatch.get(matcher).set(message);
        it.remove();
      }
    }
  }

  public boolean hasAny(Matcher<? super T> matcher) {
    synchronized (messageLog) {
      for (T element : messageLog) {
        if (matcher.matches(element)) {
          return true;
        }
      }
    }
    return false;
  }

  public T getLatest(Matcher<? super T> matcher) {
    synchronized (messageLog) {
      for (T element : Lists.reverse(messageLog)) {
        if (matcher.matches(element)) {
          return element;
        }
      }
    }
    return null;
  }

  public T waitFor(Matcher<? super T> matcher) {
    ListenableFuture<T> finished = future(matcher);

    try {
      return finished.get(WAIT_TIMEOUT, TimeUnit.SECONDS);
    } catch (Exception e) {
      Description d = new StringDescription();
      matcher.describeTo(d);
      throw new AssertionError("Failed waiting for " + d.toString(), e);
    }
  }

  public void forgetHistory() {
    messageLog.clear();
  }

  private ListenableFuture<T> future(Matcher<? super T> matcher) {
    SettableFuture<T> finished = SettableFuture.create();

    fiber.execute(() -> {
      synchronized (messageLog) {
        for (T element : messageLog) {
          if (matcher.matches(element)) {
            finished.set(element);
            return;
          }
        }
      }
      waitingToMatch.put(matcher, finished);
    });

    return finished;
  }
} public static <T> Matcher<T> isIn(Collection<T> collection) {
  return new TypeSafeMatcher<T>() {
    @Override
    protected boolean matchesSafely(T item) {
      return collection.contains(item);
    }

    @Override
    public void describeTo(Description description) {
      description.appendText("an item contained within the collection ")
          .appendValue(collection);
    }
  };
} public static IndexCommitMatcher aCommitNotice() {
  return new IndexCommitMatcher().addCriterion(
      (item) -> true,
      (description) -> description
          .appendText("an IndexCommitNotice"));
} static Matcher<ReplicatorInstanceEvent> aQuorumChangeCommittedEvent(QuorumConfiguration configuration,
                                                                    Matcher<Long> fromMatcher) {
  return new TypeSafeMatcher<ReplicatorInstanceEvent>() {
    @Override
    protected boolean matchesSafely(ReplicatorInstanceEvent item) {
      return item.eventType == ReplicatorInstanceEvent.EventType.QUORUM_CONFIGURATION_COMMITTED
          && fromMatcher.matches(item.instance.getId())
          && item.configuration.equals(configuration);
    }

    @Override
    public void describeTo(Description description) {
      description.appendText("a ReplicatorInstanceEvent indicating quorum configuration ").appendValue(configuration)
          .appendText(" was committed from replicator with ID ").appendDescriptionOf(fromMatcher);
    }
  };
} static Matcher<ReplicatorInstanceEvent> aReplicatorEvent(ReplicatorInstanceEvent.EventType type) {
  return new TypeSafeMatcher<ReplicatorInstanceEvent>() {
    @Override
    protected boolean matchesSafely(ReplicatorInstanceEvent item) {
      return item.eventType == type;
    }

    @Override
    public void describeTo(Description description) {
      description.appendText("a ReplicatorInstanceEvent of type ").appendValue(type);
    }
  };
} static Matcher<InRamTest.PeerController> hasCommittedEntriesUpTo(long index) {
  return new TypeSafeMatcher<InRamTest.PeerController>() {
    @Override
    protected boolean matchesSafely(InRamTest.PeerController peer) {
      return peer.hasCommittedEntriesUpTo(index);
    }

    @Override
    public void describeTo(Description description) {
      description.appendText("Peer has committed log entries up to index" + index);
    }
  };
} static Matcher<ReplicatorInstanceEvent> leaderElectedEvent(Matcher<Long> leaderMatcher,
                                                           Matcher<Long> termMatcher) {
  return new TypeSafeMatcher<ReplicatorInstanceEvent>() {
    @Override
    protected boolean matchesSafely(ReplicatorInstanceEvent item) {
      return item.eventType == ReplicatorInstanceEvent.EventType.LEADER_ELECTED
          && leaderMatcher.matches(item.newLeader)
          && termMatcher.matches(item.leaderElectedTerm);
    }

    @Override
    public void describeTo(Description description) {
      description.appendText("a ReplicatorInstanceEvent indicating a leader was elected")
          .appendText(" with id ").appendDescriptionOf(leaderMatcher)
          .appendText(" with term ").appendDescriptionOf(termMatcher);

    }
  };
} static Matcher<InRamTest.PeerController> theLeader() {
  return new TypeSafeMatcher<InRamTest.PeerController>() {
    @Override
    protected boolean matchesSafely(InRamTest.PeerController peer) {
      return peer.isCurrentLeader();
    }

    @Override
    public void describeTo(Description description) {
      description.appendText("The peer who is the leader of the current term");
    }
  };
} static Matcher<InRamTest.PeerController> willCommitConfiguration(QuorumConfiguration configuration) {
  return new TypeSafeMatcher<InRamTest.PeerController>() {
    Throwable matchException;

    @Override
    protected boolean matchesSafely(InRamTest.PeerController peer) {
      try {
        peer.waitForQuorumCommit(configuration);
      } catch (Exception e) {
        matchException = e;
      }
      return true;
    }

    @Override
    public void describeTo(Description description) {
      description.appendText("Peer will commit the quorum configuration ").appendValue(configuration);
    }

    @Override
    public void describeMismatchSafely(InRamTest.PeerController peer, Description description) {
      if (matchException != null) {
        description.appendValue(matchException.toString());
      }
    }
  };
} static Matcher<InRamTest.PeerController> willCommitEntriesUpTo(long index) {
  return new TypeSafeMatcher<InRamTest.PeerController>() {
    Throwable matchException;

    @Override
    protected boolean matchesSafely(InRamTest.PeerController peer) {
      try {
        peer.waitForCommit(index);
        assert peer.log.getLastIndex() >= index;
      } catch (Exception e) {
        matchException = e;
      }
      return true;
    }

    @Override
    public void describeTo(Description description) {
      description.appendText("Peer will commit log entries up to index " + index);
    }

    @Override
    public void describeMismatchSafely(InRamTest.PeerController peer, Description description) {
      if (matchException != null) {
        description.appendValue(matchException.toString());
      }
    }
  };
} static Matcher<InRamTest.PeerController> willRespondToAnAppendRequest(long minimumTerm) {
  return new TypeSafeMatcher<InRamTest.PeerController>() {
    Throwable matchException;

    @Override
    protected boolean matchesSafely(InRamTest.PeerController peer) {
      try {
        peer.waitForAppendReply(greaterThanOrEqualTo(minimumTerm));
      } catch (Exception e) {
        matchException = e;
      }
      return true;
    }

    @Override
    public void describeTo(Description description) {
      description.appendText("Peer will respond to an AppendEntries request");
    }

    @Override
    public void describeMismatchSafely(InRamTest.PeerController peer, Description description) {
      if (matchException != null) {
        description.appendValue(matchException.toString());
      }
    }
  };
} static Matcher<InRamTest.PeerController> willSend(RpcMatchers.RequestMatcher requestMatcher) {
  return new TypeSafeMatcher<InRamTest.PeerController>() {
    Throwable matchException;

    @Override
    protected boolean matchesSafely(InRamTest.PeerController peer) {
      try {
        peer.waitForRequest(requestMatcher);
      } catch (Exception e) {
        matchException = e;
      }
      return true;
    }

    @Override
    public void describeTo(Description description) {
      description.appendText("Peer will send an AppendEntries request ").appendDescriptionOf(requestMatcher);
    }

    @Override
    public void describeMismatchSafely(InRamTest.PeerController peer, Description description) {
      if (matchException != null) {
        description.appendValue(matchException.toString());
      }
    }
  };
} static Matcher<InRamTest.PeerController> wonAnElectionWithTerm(Matcher<Long> termMatcher) {
  return new TypeSafeMatcher<InRamTest.PeerController>() {
    @Override
    protected boolean matchesSafely(InRamTest.PeerController peer) {
      return peer.hasWonAnElection(termMatcher);
    }

    @Override
    public void describeTo(Description description) {
      description.appendText("a peer who won an election with term ").appendDescriptionOf(termMatcher);
    }
  };
}
import c5db.interfaces.replication.IndexCommitNotice;
import c5db.interfaces.replication.QuorumConfiguration;
import c5db.interfaces.replication.Replicator;
import c5db.interfaces.replication.ReplicatorInstanceEvent;
import c5db.interfaces.replication.ReplicatorLog;
import c5db.interfaces.replication.ReplicatorReceipt;
import c5db.replication.rpc.RpcMessage;
import c5db.replication.rpc.RpcRequest;
import c5db.replication.rpc.RpcWireReply;
import c5db.util.CheckedConsumer;
import c5db.util.ExceptionHandlingBatchExecutor;
import c5db.util.JUnitRuleFiberExceptions;
import com.google.common.collect.Lists;
import com.google.common.collect.Sets;
import com.google.common.util.concurrent.ListenableFuture;
import io.netty.util.CharsetUtil;
import org.hamcrest.Matcher;
import org.jetlang.channels.MemoryChannel;
import org.jetlang.channels.Request;
import org.jetlang.core.BatchExecutor;
import org.jetlang.core.RunnableExecutor;
import org.jetlang.core.RunnableExecutorImpl;
import org.jetlang.fibers.Fiber;
import org.jetlang.fibers.ThreadFiber;
import org.junit.After;
import org.junit.Before;
import org.junit.Rule;
import org.junit.Test;
import java.nio.ByteBuffer;
import java.util.Collection;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.function.Predicate;
import java.util.stream.Collectors;
import static c5db.AsyncChannelAsserts.ChannelHistoryMonitor;
import static c5db.CollectionMatchers.isIn;
import static c5db.FutureMatchers.resultsIn;
import static c5db.IndexCommitMatcher.aCommitNotice;
import static c5db.RpcMatchers.ReplyMatcher.aPreElectionReply;
import static c5db.RpcMatchers.ReplyMatcher.anAppendReply;
import static c5db.RpcMatchers.RequestMatcher;
import static c5db.RpcMatchers.RequestMatcher.aPreElectionPoll;
import static c5db.RpcMatchers.RequestMatcher.anAppendRequest;
import static c5db.RpcMatchers.containsQuorumConfiguration;
import static c5db.interfaces.replication.Replicator.State.FOLLOWER;
import static c5db.interfaces.replication.ReplicatorInstanceEvent.EventType.ELECTION_TIMEOUT;
import static c5db.replication.ReplicationMatchers.aQuorumChangeCommittedEvent;
import static c5db.replication.ReplicationMatchers.aReplicatorEvent;
import static c5db.replication.ReplicationMatchers.hasCommittedEntriesUpTo;
import static c5db.replication.ReplicationMatchers.leaderElectedEvent;
import static c5db.replication.ReplicationMatchers.theLeader;
import static c5db.replication.ReplicationMatchers.willCommitConfiguration;
import static c5db.replication.ReplicationMatchers.willCommitEntriesUpTo;
import static c5db.replication.ReplicationMatchers.willRespondToAnAppendRequest;
import static c5db.replication.ReplicationMatchers.willSend;
import static c5db.replication.ReplicationMatchers.wonAnElectionWithTerm;
import static org.hamcrest.CoreMatchers.anyOf;
import static org.hamcrest.CoreMatchers.equalTo;
import static org.hamcrest.CoreMatchers.hasItem;
import static org.hamcrest.CoreMatchers.is;
import static org.hamcrest.CoreMatchers.nullValue;
import static org.hamcrest.MatcherAssert.assertThat;
import static org.hamcrest.Matchers.any;
import static org.hamcrest.Matchers.greaterThan;
import static org.hamcrest.Matchers.greaterThanOrEqualTo;
import static org.hamcrest.core.IsNot.not;
                .to(peer.id))));

    peers(newPeerIds).forEach((peer) ->
        assertThat(peer, willCommitConfiguration(transitionalConfig)));

    peersBeingRemoved(transitionalConfig).forEach(PeerController::allowToTimeout);
    waitForAnElectionTimeout();

    peersBeingRemoved(transitionalConfig).forEach((peer) -> {
      if (peer.id != leaderId) {
        assertThat(peer, willSend(aPreElectionPoll()));
      }
    });

    waitForALeaderWithId(isIn(newPeerIds));
    leader().log(someData());

    peers(newPeerIds).forEach((peer) ->
        assertThat(peer, willCommitConfiguration(finalConfig)));

    peersBeingRemoved(transitionalConfig).forEach((peer) ->
        assertThat(peer, not(wonAnElectionWithTerm(greaterThan(firstLeaderTerm)))));
  }

  @Test
  public void aQuorumCanElectANewLeaderEvenWhileReceivingMessagesFromRemovedPeersWhoHaveTimedOut() throws Exception {
    final Set<Long> newPeerIds = smallerPeerSetWithNoneInCommonWithInitialSet();
    final QuorumConfiguration transitionalConfig =
        QuorumConfiguration.of(initialPeerSet()).getTransitionalConfiguration(newPeerIds);
    final QuorumConfiguration finalConfig = transitionalConfig.getCompletedConfiguration();

    havingElectedALeaderAtOrAfter(term(1));
    long firstLeaderTerm = currentTerm();

    leader().changeQuorum(newPeerIds);
    sim.createAndStartReplicators(newPeerIds);

    peers(newPeerIds).forEach((peer) ->
        assertThat(peer, willCommitConfiguration(finalConfig)));

    peersBeingRemoved(transitionalConfig).forEach(PeerController::allowToTimeout);
    waitForAnElectionTimeout();

    waitForALeader(term(firstLeaderTerm + 1));

    leader().die();

    waitForANewLeader();
  }

  @Test
  public void aLateBootstrapCallWillBeDisregarded() throws Exception {
    havingElectedALeaderAtOrAfter(term(1));

    leader().logDataUpToIndex(2);
    allPeers((peer) -> assertThat(peer, willCommitEntriesUpTo(lastIndexLogged())));

    // Bootstrap calls to both leader and a non-leader -- both will be no-ops
    pickNonLeader().instance.bootstrapQuorum(smallerPeerSetWithNoneInCommonWithInitialSet());
    leader().instance.bootstrapQuorum(smallerPeerSetWithNoneInCommonWithInitialSet());

    // Verify that quorum is still in a working state
    assertThat(sim.getLog(leader().id).getLastIndex(), is(equalTo(2L)));

    leader().logDataUpToIndex(3);
    allPeers((peer) -> assertThat(peer, willCommitEntriesUpTo(lastIndexLogged())));
  }

  /**
   * Private methods
   */

  // Blocks until a leader is elected during some term >= minimumTerm.
  // Throws TimeoutException if that does not occur within the time limit.
  private void waitForALeader(long minimumTerm) {
    waitForALeaderElectedEventMatching(anyLeader(), greaterThanOrEqualTo(minimumTerm));
  }

  private void waitForALeaderWithId(Matcher<Long> leaderIdMatcher) {
    waitForALeaderElectedEventMatching(leaderIdMatcher, anyTerm());
  }

  private void waitForALeaderElectedEventMatching(Matcher<Long> leaderIdMatcher, Matcher<Long> termMatcher) {
    sim.startAllTimeouts();
    eventMonitor.waitFor(leaderElectedEvent(leaderIdMatcher, termMatcher));
    sim.stopAllTimeouts();

    // Wait for at least one other node to recognize the new leader. This is necessary because
    // some tests want to be able to identify a follower right away.
    pickNonLeader().waitForAppendReply(termMatcher);

    final long leaderId = currentLeader();
    assertThat(leaderCount(), is(equalTo(1)));

    sim.startTimeout(leaderId);
  }

  private void havingElectedALeaderAtOrAfter(long minimumTerm) {
    waitForALeader(minimumTerm);
  }

  private void waitForANewLeader() {
    waitForALeader(currentTerm() + 1);
  }

  // Counts leaders in the current term. Used to verify a sensible state.
  // If the simulation is running correctly, this should only ever return 0 or 1.
  private int leaderCount() {
    final long currentTerm = currentTerm();
    int leaderCount = 0;

    for (long id : sim.getOnlinePeers()) {
      if (peer(id).hasWonAnElection(greaterThanOrEqualTo(currentTerm))) {
        leaderCount++;
      }
    }
    return leaderCount;
  }

  private void waitForAnElectionTimeout() {
    eventMonitor.waitFor(aReplicatorEvent(ELECTION_TIMEOUT));