Copycat - MemberShip

public interface Member {

  enum Type {

    INACTIVE,

    RESERVE,

    PASSIVE,

    ACTIVE,

  }

  CompletableFuture<Void> promote();
  
  CompletableFuture<Void> demote();

}

public final class ServerMember implements Member, CatalystSerializable, AutoCloseable {

  @Override
  public CompletableFuture<Void> promote() {
    return configure(Type.values()[type.ordinal() + 1]);
  }
  
  @Override
  public CompletableFuture<Void> demote() {
    return configure(Type.values()[type.ordinal() - 1]);
  }
        
  /**
   * Recursively reconfigures the cluster.
   */
  private void configure(Member.Type type, CompletableFuture<Void> future) {
    // Set a timer to retry the attempt to leave the cluster.
    configureTimeout = cluster.getContext().getThreadContext().schedule(cluster.getContext().getElectionTimeout(), () -> {
      configure(type, future);
    });

    // Attempt to leave the cluster by submitting a LeaveRequest directly to the server state.
    // Non-leader states should forward the request to the leader if there is one. Leader states
    // will log, replicate, and commit the reconfiguration.
    cluster.getContext().getServerState().reconfigure(ReconfigureRequest.builder() //往ServerState发送reconfigure请求
      .withIndex(cluster.getConfiguration().index())
      .withTerm(cluster.getConfiguration().term())
      .withMember(new ServerMember(type, serverAddress(), clientAddress(), updated))
      .build()).whenComplete((response, error) -> {
        //.......
    });
  }

public interface ServerState extends Managed<ServerState> {


  CopycatServer.State type();

  CompletableFuture<RegisterResponse> register(RegisterRequest request);

  CompletableFuture<ConnectResponse> connect(ConnectRequest request, Connection connection);

  CompletableFuture<KeepAliveResponse> keepAlive(KeepAliveRequest request);

  CompletableFuture<UnregisterResponse> unregister(UnregisterRequest request);

  CompletableFuture<PublishResponse> publish(PublishRequest request);

  CompletableFuture<ConfigureResponse> configure(ConfigureRequest request);

  CompletableFuture<InstallResponse> install(InstallRequest request);

  CompletableFuture<JoinResponse> join(JoinRequest request);

  CompletableFuture<ReconfigureResponse> reconfigure(ReconfigureRequest request);

  CompletableFuture<LeaveResponse> leave(LeaveRequest request);

  CompletableFuture<AppendResponse> append(AppendRequest request);

  CompletableFuture<PollResponse> poll(PollRequest request);

  CompletableFuture<VoteResponse> vote(VoteRequest request);

  CompletableFuture<CommandResponse> command(CommandRequest request);

  CompletableFuture<QueryResponse> query(QueryRequest request);

}

public abstract class AbstractState implements ServerState {

  /**
   * Forwards the given request to the leader if possible.
   */
  protected <T extends Request, U extends Response> CompletableFuture<U> forward(T request) {
    CompletableFuture<U> future = new CompletableFuture<>();
    context.getConnections().getConnection(context.getLeader().serverAddress()).whenComplete((connection, connectError) -> {
      if (connectError == null) {
        connection.<T, U>send(request).whenComplete((response, responseError) -> {
          if (responseError == null) {
            future.complete(response);
          } else {
            future.completeExceptionally(responseError);
          }
        });
      } else {
        future.completeExceptionally(connectError);
      }
    });
    return future;
  }

  /**
   * Updates the term and leader.
   */
  protected boolean updateTermAndLeader(long term, int leader) {
    // If the request indicates a term that is greater than the current term or no leader has been
    // set for the current term, update leader and term.
    if (term > context.getTerm() || (term == context.getTerm() && context.getLeader() == null && leader != 0)) {
      context.setTerm(term);
      context.setLeader(leader);

      // Reset the current cluster configuration to the last committed configuration when a leader change occurs.
      context.getClusterState().reset();
      return true;
    }
    return false;
  }

}

class InactiveState extends AbstractState {

  @Override
  public CompletableFuture<ConfigureResponse> configure(ConfigureRequest request) {

    updateTermAndLeader(request.term(), request.leader());

    Configuration configuration = new Configuration(request.index(), request.term(), request.timestamp(), request.members());

    // Configure the cluster membership. This will cause this server to transition to the
    // appropriate state if its type has changed.
    context.getClusterState().configure(configuration); // 更新clusterState中对应server的state

    // If the configuration is already committed, commit it to disk.
    // Check against the actual cluster Configuration rather than the received configuration in
    // case the received configuration was an older configuration that was not applied.
    if (context.getCommitIndex() >= context.getClusterState().getConfiguration().index()) {
      context.getClusterState().commit();
    }

    return CompletableFuture.completedFuture(logResponse(ConfigureResponse.builder()
      .withStatus(Response.Status.OK)
      .build()));
  }

if (context.getMetaStore().loadConfiguration().index() < configuration.index()) {
      context.getMetaStore().storeConfiguration(configuration);
}

class ReserveState extends InactiveState {
  @Override
  public CompletableFuture<AppendResponse> append(AppendRequest request) {
    context.checkThread();
    logRequest(request);
    updateTermAndLeader(request.term(), request.leader());

    // Update the local commitIndex and globalIndex.
    context.setCommitIndex(request.commitIndex());
    context.setGlobalIndex(request.globalIndex()); //Sets the maximum compaction index for major compaction

    return CompletableFuture.completedFuture(logResponse(AppendResponse.builder()
      .withStatus(Response.Status.OK)
      .withTerm(context.getTerm())
      .withSucceeded(true)
      .withLogIndex(0)
      .build()));
  }

protected boolean updateTermAndLeader(long term, int leader) {
    // If the request indicates a term that is greater than the current term or no leader has been
    // set for the current term, update leader and term.
    if (term > context.getTerm() || (term == context.getTerm() && context.getLeader() == null && leader != 0)) {
      context.setTerm(term); //如果term，leader是新的更新context信息
      context.setLeader(leader);

      // Reset the current cluster configuration to the last committed configuration when a leader change occurs.
      context.getClusterState().reset(); //需要reset到最新commited配置，避免老leader的脏数据
      return true;
    }
    return false;
  }

ServerContext setCommitIndex(long commitIndex) {

    long previousCommitIndex = this.commitIndex;
    if (commitIndex > previousCommitIndex) {
      this.commitIndex = commitIndex;
      log.commit(Math.min(commitIndex, log.lastIndex())); //log commit到这个index
      long configurationIndex = cluster.getConfiguration().index();
      if (configurationIndex > previousCommitIndex && configurationIndex <= commitIndex) {
        cluster.commit(); //commit cluster的configuration
      }
    }
    return this;
  }

@Override
  public CompletableFuture<ServerState> open() {
    return super.open()
      .thenRun(this::truncateUncommittedEntries)
      .thenApply(v -> this);
  }

private void truncateUncommittedEntries() {
    if (type() == CopycatServer.State.PASSIVE) {
      context.getLog().truncate(Math.min(context.getCommitIndex(), context.getLog().lastIndex()));
    }
  }

    // Close the old state.
    try {
      this.state.close().get();
    } catch (InterruptedException | ExecutionException e) {
      throw new IllegalStateException("failed to close Raft state", e);
    }

    // Force state transitions to occur synchronously in order to prevent race conditions.
    try {
      this.state = createState(state);
      this.state.open().get();
    } catch (InterruptedException | ExecutionException e) {
      throw new IllegalStateException("failed to initialize Raft state", e);
    }

context.getStateMachine().executor().context().sessions().registerConnection(request.client(), connection);

@Override
  public CompletableFuture<AppendResponse> append(final AppendRequest request) {
    context.checkThread();
    logRequest(request);
    updateTermAndLeader(request.term(), request.leader());

    return CompletableFuture.completedFuture(logResponse(handleAppend(request)));
  }

protected AppendResponse handleAppend(AppendRequest request) {
    // If the request term is less than the current term then immediately
    // reply false and return our current term. The leader will receive
    // the updated term and step down.
    if (request.term() < context.getTerm()) {
      LOGGER.debug("{} - Rejected {}: request term is less than the current term ({})", context.getCluster().member().address(), request, context.getTerm());
      return AppendResponse.builder()
        .withStatus(Response.Status.OK)
        .withTerm(context.getTerm())
        .withSucceeded(false)
        .withLogIndex(context.getLog().lastIndex())
        .build();
    } else {
      return checkGlobalIndex(request);
    }
  }

/**
   * Appends entries to the local log.
   */
  protected AppendResponse appendEntries(AppendRequest request) {
    // Get the last entry index or default to the request log index.
    long lastEntryIndex = request.logIndex();
    if (!request.entries().isEmpty()) {
      lastEntryIndex = request.entries().get(request.entries().size() - 1).getIndex();
    }

    // Ensure the commitIndex is not increased beyond the index of the last entry in the request.
    long commitIndex = Math.max(context.getCommitIndex(), Math.min(request.commitIndex(), lastEntryIndex));

    // Append entries to the log starting at the last log index.
    for (Entry entry : request.entries()) {
      // If the entry index is greater than the last index and less than the commit index, append the entry.
      // We perform no additional consistency checks here since passive members may only receive committed entries.
      if (context.getLog().lastIndex() < entry.getIndex() && entry.getIndex() <= commitIndex) {
        context.getLog().skip(entry.getIndex() - context.getLog().lastIndex() - 1).append(entry);
        LOGGER.debug("{} - Appended {} to log at index {}", context.getCluster().member().address(), entry, entry.getIndex());
      }
    }

    // Update the context commit and global indices.
    context.setCommitIndex(commitIndex);
    context.setGlobalIndex(request.globalIndex());

    // Apply commits to the state machine in batch.
    context.getStateMachine().applyAll(context.getCommitIndex());

    return AppendResponse.builder()
      .withStatus(Response.Status.OK)
      .withTerm(context.getTerm())
      .withSucceeded(true)
      .withLogIndex(context.getLog().lastIndex())
      .build();
  }

@Override
  public CompletableFuture<QueryResponse> query(QueryRequest request) {
    context.checkThread();
    logRequest(request);

    // If the query was submitted with RYW or monotonic read consistency, attempt to apply the query to the local state machine.
    if (request.query().consistency() == Query.ConsistencyLevel.SEQUENTIAL) {

      // If this server has not yet applied entries up to the client‘s session ID, forward the
      // query to the leader. This ensures that a follower does not tell the client its session
      // doesn‘t exist if the follower hasn‘t had a chance to see the session‘s registration entry.
      if (context.getStateMachine().getLastApplied() < request.session()) {
        LOGGER.debug("{} - State out of sync, forwarding query to leader");
        return queryForward(request);
      }

      // If the commit index is not in the log then we‘ve fallen too far behind the leader to perform a local query.
      // Forward the request to the leader.
      if (context.getLog().lastIndex() < context.getCommitIndex()) {
        LOGGER.debug("{} - State out of sync, forwarding query to leader");
        return queryForward(request);
      }

      QueryEntry entry = context.getLog().create(QueryEntry.class)
        .setIndex(request.index())
        .setTerm(context.getTerm())
        .setTimestamp(System.currentTimeMillis())
        .setSession(request.session())
        .setSequence(request.sequence())
        .setQuery(request.query());

      return queryLocal(entry); //查询local数据
    } else {
      return queryForward(request);
    }
  }

@Override
  public CompletableFuture<AppendResponse> append(final AppendRequest request) {
    context.checkThread();
    logRequest(request);

    // If the request indicates a term that is greater than the current term then
    // assign that term and leader to the current context and transition to follower.
    boolean transition = updateTermAndLeader(request.term(), request.leader()); //如果request的term比当前的term大，说明你肯定不是leader

    CompletableFuture<AppendResponse> future = CompletableFuture.completedFuture(logResponse(handleAppend(request)));

    // If a transition is required then transition back to the follower state.
    // If the node is already a follower then the transition will be ignored.
    if (transition) {
      context.transition(CopycatServer.State.FOLLOWER); //切换到Follower
    }
    return future;
  }

// Iterate through request entries and append them to the log.
    for (Entry entry : request.entries()) {
      // If the entry index is greater than the last log index, skip missing entries.
      if (context.getLog().lastIndex() < entry.getIndex()) {
        context.getLog().skip(entry.getIndex() - context.getLog().lastIndex() - 1).append(entry);
        LOGGER.debug("{} - Appended {} to log at index {}", context.getCluster().member().address(), entry, entry.getIndex());
      } else if (context.getCommitIndex() >= entry.getIndex()) {
        continue;
      } else {
        // Compare the term of the received entry with the matching entry in the log.
        long term = context.getLog().term(entry.getIndex());
        if (term != 0) {
          if (entry.getTerm() != term) {
            // We found an invalid entry in the log. Remove the invalid entry and append the new entry.
            // If appending to the log fails, apply commits and reply false to the append request.
            LOGGER.debug("{} - Appended entry term does not match local log, removing incorrect entries", context.getCluster().member().address());
            context.getLog().truncate(entry.getIndex() - 1).append(entry);
            LOGGER.debug("{} - Appended {} to log at index {}", context.getCluster().member().address(), entry, entry.getIndex());
          }
        } else {
          context.getLog().truncate(entry.getIndex() - 1).append(entry);
          LOGGER.debug("{} - Appended {} to log at index {}", context.getCluster().member().address(), entry, entry.getIndex());
        }
      }
    }

/**
   * Handles a poll request.
   */
  protected PollResponse handlePoll(PollRequest request) {
    // If the request term is not as great as the current context term then don‘t
    // vote for the candidate. We want to vote for candidates that are at least
    // as up to date as us.
    if (request.term() < context.getTerm()) {
      LOGGER.debug("{} - Rejected {}: candidate‘s term is less than the current term", context.getCluster().member().address(), request);
      return PollResponse.builder()
        .withStatus(Response.Status.OK)
        .withTerm(context.getTerm())
        .withAccepted(false) //拒绝
        .build();
    } else if (isLogUpToDate(request.logIndex(), request.logTerm(), request)) {
      return PollResponse.builder()
        .withStatus(Response.Status.OK)
        .withTerm(context.getTerm())
        .withAccepted(true) //接受
        .build();
    } else {
      return PollResponse.builder()
        .withStatus(Response.Status.OK)
        .withTerm(context.getTerm())
        .withAccepted(false)
        .build();
    }
  }

boolean isLogUpToDate(long lastIndex, long lastTerm, Request request) {
    // If the log is empty then vote for the candidate.
    if (context.getLog().isEmpty()) {
      LOGGER.debug("{} - Accepted {}: candidate‘s log is up-to-date", context.getCluster().member().address(), request);
      return true;
    }

    // Read the last entry index and term from the log.
    long localLastIndex = context.getLog().lastIndex();
    long localLastTerm = context.getLog().term(localLastIndex);

    // If the candidate‘s last log term is lower than the local log‘s last entry term, reject the request.
    if (lastTerm < localLastTerm) {
      LOGGER.debug("{} - Rejected {}: candidate‘s last log entry ({}) is at a lower term than the local log ({})", context.getCluster().member().address(), request, lastTerm, localLastTerm);
      return false;
    }

    // If the candidate‘s last term is equal to the local log‘s last entry term, reject the request if the
    // candidate‘s last index is less than the local log‘s last index. If the candidate‘s last log term is
    // greater than the local log‘s last term then it‘s considered up to date, and if both have the same term
    // then the candidate‘s last index must be greater than the local log‘s last index.
    if (lastTerm == localLastTerm && lastIndex < localLastIndex) {
      LOGGER.debug("{} - Rejected {}: candidate‘s last log entry ({}) is at a lower index than the local log ({})", context.getCluster().member().address(), request, lastIndex, localLastIndex);
      return false;
    }

    // If we made it this far, the candidate‘s last term is greater than or equal to the local log‘s last
    // term, and if equal to the local log‘s last term, the candidate‘s last index is equal to or greater
    // than the local log‘s last index.
    LOGGER.debug("{} - Accepted {}: candidate‘s log is up-to-date", context.getCluster().member().address(), request);
    return true;
  }

/**
   * Handles a vote request.
   */
  protected VoteResponse handleVote(VoteRequest request) {
    // If the request term is not as great as the current context term then don‘t
    // vote for the candidate. We want to vote for candidates that are at least
    // as up to date as us.
    if (request.term() < context.getTerm()) { //request的term旧
      LOGGER.debug("{} - Rejected {}: candidate‘s term is less than the current term", context.getCluster().member().address(), request);
      return VoteResponse.builder()
        .withStatus(Response.Status.OK)
        .withTerm(context.getTerm())
        .withVoted(false)
        .build();
    }
    // If a leader was already determined for this term then reject the request.
    else if (context.getLeader() != null) { //已经有leader
      LOGGER.debug("{} - Rejected {}: leader already exists", context.getCluster().member().address(), request);
      return VoteResponse.builder()
        .withStatus(Response.Status.OK)
        .withTerm(context.getTerm())
        .withVoted(false)
        .build();
    }
    // If the requesting candidate is not a known member of the cluster (to this
    // node) then don‘t vote for it. Only vote for candidates that we know about.
    else if (!context.getClusterState().getRemoteMemberStates().stream().<Integer>map(m -> m.getMember().id()).collect(Collectors.toSet()).contains(request.candidate())) {
      LOGGER.debug("{} - Rejected {}: candidate is not known to the local member", context.getCluster().member().address(), request);
      return VoteResponse.builder()
        .withStatus(Response.Status.OK)
        .withTerm(context.getTerm())
        .withVoted(false)
        .build();
    }
    // If no vote has been cast, check the log and cast a vote if necessary.
    else if (context.getLastVotedFor() == 0) { //如果还没有vote其他的candidator
      if (isLogUpToDate(request.logIndex(), request.logTerm(), request)) { //足够新
        context.setLastVotedFor(request.candidate());
        return VoteResponse.builder()
          .withStatus(Response.Status.OK)
          .withTerm(context.getTerm())
          .withVoted(true) //接受
          .build();
      } else {
        return VoteResponse.builder()
          .withStatus(Response.Status.OK)
          .withTerm(context.getTerm())
          .withVoted(false)
          .build();
      }
    }
    // If we already voted for the requesting server, respond successfully.
    else if (context.getLastVotedFor() == request.candidate()) { //之前选中也是该candidator
      LOGGER.debug("{} - Accepted {}: already voted for {}", context.getCluster().member().address(), request, context.getCluster().member(context.getLastVotedFor()).address());
      return VoteResponse.builder()
        .withStatus(Response.Status.OK)
        .withTerm(context.getTerm())
        .withVoted(true) //接受
        .build();
    }
    // In this case, we‘ve already voted for someone else.
    else {
      LOGGER.debug("{} - Rejected {}: already voted for {}", context.getCluster().member().address(), request, context.getCluster().member(context.getLastVotedFor()).address());
      return VoteResponse.builder()
        .withStatus(Response.Status.OK)
        .withTerm(context.getTerm())
        .withVoted(false)
        .build();
    }
  }

public synchronized CompletableFuture<ServerState> open() {
    return super.open().thenRun(this::startHeartbeatTimeout).thenApply(v -> this);
  }

private void resetHeartbeatTimeout() {

    // Set the election timeout in a semi-random fashion with the random range
    // being election timeout and 2 * election timeout.
    Duration delay = context.getElectionTimeout().plus(Duration.ofMillis(random.nextInt((int) context.getElectionTimeout().toMillis()))); //随机产生delay时间
    heartbeatTimer = context.getThreadContext().schedule(delay, () -> { //delay时间到后
      heartbeatTimer = null;
      if (isOpen()) {
        context.setLeader(0); //清空leader
        sendPollRequests(); //发送poll请求
      }
    });
  }

/**
   * Polls all members of the cluster to determine whether this member should transition to the CANDIDATE state.
   */
  private void sendPollRequests() {
    // Create a quorum that will track the number of nodes that have responded to the poll request.
    final AtomicBoolean complete = new AtomicBoolean();
    final Set<ServerMember> votingMembers = new HashSet<>(context.getClusterState().getActiveMemberStates().stream().map(MemberState::getMember).collect(Collectors.toList()));

    //找到所有active的members，并且生成Quorum
    final Quorum quorum = new Quorum(context.getClusterState().getQuorum(), (elected) -> {
      // If a majority of the cluster indicated they would vote for us then transition to candidate.
      complete.set(true);
      if (elected) { //如果elected成功
        context.transition(CopycatServer.State.CANDIDATE); //迁移到candidate
      } else {
        resetHeartbeatTimeout();
      }
    });

    // Once we got the last log term, iterate through each current member
    // of the cluster and vote each member for a vote.
    for (ServerMember member : votingMembers) {
      LOGGER.debug("{} - Polling {} for next term {}", context.getCluster().member().address(), member, context.getTerm() + 1);
      PollRequest request = PollRequest.builder()
        .withTerm(context.getTerm())
        .withCandidate(context.getCluster().member().id())
        .withLogIndex(lastIndex) //当前我的lastindex
        .withLogTerm(lastTerm) //当前我的lastTerm，别人需要根据index和term来决定是否poll我
        .build();
      context.getConnections().getConnection(member.serverAddress()).thenAccept(connection -> {
        connection.<PollRequest, PollResponse>send(request).whenCompleteAsync((response, error) -> { //异步发送request，并且加上callback
          context.checkThread();
          if (isOpen() && !complete.get()) {
            if (error != null) {
              LOGGER.warn("{} - {}", context.getCluster().member().address(), error.getMessage());
              quorum.fail();
            } else {
              if (response.term() > context.getTerm()) {
                context.setTerm(response.term());
              }

              if (!response.accepted()) {
                LOGGER.debug("{} - Received rejected poll from {}", context.getCluster().member().address(), member);
                quorum.fail();
              } else if (response.term() != context.getTerm()) {
                LOGGER.debug("{} - Received accepted poll for a different term from {}", context.getCluster().member().address(), member);
                quorum.fail();
              } else {
                LOGGER.debug("{} - Received accepted poll from {}", context.getCluster().member().address(), member);
                quorum.succeed(); //各种错误后，只有这个表示对方poll我了
              }
            }
          }
        }, context.getThreadContext().executor());
      });
    }
  }

public class Quorum {
  private final int quorum;
  private int succeeded = 1;
  private int failed;
  private Consumer<Boolean> callback;
  private boolean complete;

  public Quorum(int quorum, Consumer<Boolean> callback) {
    this.quorum = quorum;
    this.callback = callback;
  }

  private void checkComplete() {
    if (!complete && callback != null) {
      if (succeeded >= quorum) {
        complete = true;
        callback.accept(true);
      } else if (failed >= quorum) {
        complete = true;
        callback.accept(false);
      }
    }
  }

  /**
   * Indicates that a call in the quorum succeeded.
   */
  public Quorum succeed() {
    succeeded++;
    checkComplete();
    return this;
  }

@Override
  public CompletableFuture<AppendResponse> append(AppendRequest request) {
    CompletableFuture<AppendResponse> future = super.append(request);

    // Reset the heartbeat timeout.
    resetHeartbeatTimeout();

    // Send AppendEntries requests to passive members if necessary.
    appender.appendEntries();
    return future;
  }

final class FollowerAppender extends AbstractAppender {

  public FollowerAppender(ServerContext context) {
    super(context);
  }

  /**
   * Sends append requests to assigned passive members.
   */
  public void appendEntries() {
    if (open) {
      for (MemberState member : context.getClusterState().getAssignedPassiveMemberStates()) {
        appendEntries(member);
      }
    }
  }

public synchronized CompletableFuture<ServerState> open() {
    return super.open().thenRun(this::startElection).thenApply(v -> this);
  }

/**
   * Resets the election timer.
   */
  private void sendVoteRequests() {

    // When the election timer is reset, increment the current term and
    // restart the election.
    context.setTerm(context.getTerm() + 1).setLastVotedFor(context.getCluster().member().id());
    //重新选举，所以term+1；setLastVoteFor，设成self，先投自己一票

    final AtomicBoolean complete = new AtomicBoolean();
    final Set<ServerMember> votingMembers = new HashSet<>(context.getClusterState().getActiveMemberStates().stream().map(MemberState::getMember).collect(Collectors.toList()));

    // Send vote requests to all nodes. The vote request that is sent
    // to this node will be automatically successful.
    // First check if the quorum is null. If the quorum isn‘t null then that
    // indicates that another vote is already going on.
    final Quorum quorum = new Quorum(context.getClusterState().getQuorum(), (elected) -> {
      complete.set(true);
      if (elected) {
        context.transition(CopycatServer.State.LEADER); //vote成功，就是leader
      } else {
        context.transition(CopycatServer.State.FOLLOWER);
      }
    });

    // Once we got the last log term, iterate through each current member
    // of the cluster and vote each member for a vote.
    for (ServerMember member : votingMembers) {
      LOGGER.debug("{} - Requesting vote from {} for term {}", context.getCluster().member().address(), member, context.getTerm());
      VoteRequest request = VoteRequest.builder()
        .withTerm(context.getTerm())
        .withCandidate(context.getCluster().member().id())
        .withLogIndex(lastIndex)
        .withLogTerm(lastTerm)
        .build();

      context.getConnections().getConnection(member.serverAddress()).thenAccept(connection -> {
        connection.<VoteRequest, VoteResponse>send(request).whenCompleteAsync((response, error) -> {
          context.checkThread();
          if (isOpen() && !complete.get()) {
            if (error != null) {
              LOGGER.warn(error.getMessage());
              quorum.fail();
            } else {
              if (response.term() > context.getTerm()) {
                LOGGER.debug("{} - Received greater term from {}", context.getCluster().member().address(), member);
                context.setTerm(response.term());
                complete.set(true);
                context.transition(CopycatServer.State.FOLLOWER);
              } else if (!response.voted()) {
                LOGGER.debug("{} - Received rejected vote from {}", context.getCluster().member().address(), member);
                quorum.fail();
              } else if (response.term() != context.getTerm()) {
                LOGGER.debug("{} - Received successful vote for a different term from {}", context.getCluster().member().address(), member);
                quorum.fail();
              } else {
                LOGGER.debug("{} - Received successful vote from {}", context.getCluster().member().address(), member);
                quorum.succeed();
              }
            }
          }
        }, context.getThreadContext().executor());
      });
    }
  }

public CompletableFuture<AppendResponse> append(AppendRequest request) {
    context.checkThread();

    // If the request indicates a term that is greater than the current term then
    // assign that term and leader to the current context and step down as a candidate.
    if (request.term() >= context.getTerm()) { //如果term比我的大，说明已经有leader
      context.setTerm(request.term());
      context.transition(CopycatServer.State.FOLLOWER); //退化成follower
    }
    return super.append(request);
  }

@Override
  public CompletableFuture<VoteResponse> vote(VoteRequest request) {
    context.checkThread();
    logRequest(request);

    // If the request indicates a term that is greater than the current term then
    // assign that term and leader to the current context and step down as a candidate.
    if (updateTermAndLeader(request.term(), 0)) { //如果request term比我大，说明已经有leader
      CompletableFuture<VoteResponse> future = super.vote(request);
      context.transition(CopycatServer.State.FOLLOWER); //退化成follower
      return future;
    }

    // If the vote request is not for this candidate then reject the vote.
    if (request.candidate() == context.getCluster().member().id()) { //否则，只有request的candidate id是我，我才同意
      return CompletableFuture.completedFuture(logResponse(VoteResponse.builder()
        .withStatus(Response.Status.OK)
        .withTerm(context.getTerm())
        .withVoted(true)
        .build()));
    } else { //candidate不会同意其他的candidate
      return CompletableFuture.completedFuture(logResponse(VoteResponse.builder()
        .withStatus(Response.Status.OK)
        .withTerm(context.getTerm())
        .withVoted(false)
        .build()));
    }
  }