@Override
public CompletableFuture<ReconfigureResponse> onReconfigure(ReconfigureRequest request) {
  raft.checkThread();
  logRequest(request);

  if (raft.getLeader() == null) {
    return CompletableFuture.completedFuture(logResponse(ReconfigureResponse.builder()
        .withStatus(RaftResponse.Status.ERROR)
        .withError(RaftError.Type.NO_LEADER)
        .build()));
  } else {
    return forward(request, raft.getProtocol()::reconfigure)
        .exceptionally(error -> ReconfigureResponse.builder()
            .withStatus(RaftResponse.Status.ERROR)
            .withError(RaftError.Type.NO_LEADER)
            .build())
        .thenApply(this::logResponse);
  }
}
 

/**
 * Recursively reconfigures the cluster.
 */
private void configure(RaftMember.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().getRaftRole().onReconfigure(ReconfigureRequest.builder()
      .withIndex(cluster.getConfiguration().index())
      .withTerm(cluster.getConfiguration().term())
      .withMember(new DefaultRaftMember(id, type, updated))
      .build()).whenComplete((response, error) -> {
        if (error == null) {
          if (response.status() == RaftResponse.Status.OK) {
            cancelConfigureTimer();
            cluster.configure(new Configuration(response.index(), response.term(), response.timestamp(), response.members()));
            future.complete(null);
          } else if (response.error() == null
              || response.error().type() == RaftError.Type.UNAVAILABLE
              || response.error().type() == RaftError.Type.PROTOCOL_ERROR
              || response.error().type() == RaftError.Type.NO_LEADER) {
            cancelConfigureTimer();
            configureTimeout = cluster.getContext().getThreadContext().schedule(cluster.getContext().getElectionTimeout().multipliedBy(2), () -> configure(type, future));
          } else {
            cancelConfigureTimer();
            future.completeExceptionally(response.error().createException());
          }
        } else {
          future.completeExceptionally(error);
        }
      });
}
 

@Override
public CompletableFuture<ReconfigureResponse> onReconfigure(ReconfigureRequest request) {
  logRequest(request);
  return Futures.completedFuture(logResponse(ReconfigureResponse.builder()
      .withStatus(Status.ERROR)
      .withError(RaftError.Type.UNAVAILABLE)
      .build()));
}
 

@Override
public CompletableFuture<ReconfigureResponse> reconfigure(MemberId memberId,
                                                          ReconfigureRequest request) {
  return sendAndReceive(memberId, "reconfigure", request);
}
 

@Override
public void registerReconfigureHandler(Function<ReconfigureRequest,
    CompletableFuture<ReconfigureResponse>> handler) {
  registerHandler("reconfigure", handler);
}
 

@Override
public CompletableFuture<ReconfigureResponse> reconfigure(MemberId memberId,
                                                          ReconfigureRequest request) {
  return getServer(memberId).thenCompose(listener ->
      listener.reconfigure(encode(request))).thenApply(this::decode);
}
 

@Override
public void registerReconfigureHandler(Function<ReconfigureRequest,
    CompletableFuture<ReconfigureResponse>> handler) {
  this.reconfigureHandler = handler;
}
 

@Override
public CompletableFuture<ReconfigureResponse> reconfigure(MemberId memberId,
                                                          ReconfigureRequest request) {
  return sendAndReceive(memberId, "reconfigure", request);
}
 

@Override
public void registerReconfigureHandler(Function<ReconfigureRequest,
    CompletableFuture<ReconfigureResponse>> handler) {
  registerHandler("reconfigure", handler);
}
 

@Override
public CompletableFuture<ReconfigureResponse> reconfigure(MemberId memberId,
                                                          ReconfigureRequest request) {
  return getServer(memberId).thenCompose(listener ->
      listener.reconfigure(encode(request))).thenApply(this::decode);
}
 

@Override
public void registerReconfigureHandler(Function<ReconfigureRequest,
    CompletableFuture<ReconfigureResponse>> handler) {
  this.reconfigureHandler = handler;
}
 

@Override
public CompletableFuture<ReconfigureResponse> reconfigure(MemberId memberId, ReconfigureRequest request) {
  return sendAndReceive(context.reconfigureSubject, request, memberId);
}
 

@Override
public void registerReconfigureHandler(Function<ReconfigureRequest, CompletableFuture<ReconfigureResponse>> handler) {
  clusterCommunicator.subscribe(context.reconfigureSubject, serializer::decode, handler, serializer::encode);
}
 

@Override
public CompletableFuture<ReconfigureResponse> reconfigure(MemberId memberId, ReconfigureRequest request) {
  return sendAndReceive(memberId, "reconfigure", request);
}
 

@Override
public void registerReconfigureHandler(Function<ReconfigureRequest, CompletableFuture<ReconfigureResponse>> handler) {
  registerHandler("reconfigure", handler);
}
 

@Override
public CompletableFuture<ReconfigureResponse> reconfigure(MemberId memberId, ReconfigureRequest request) {
  return getServer(memberId).thenCompose(listener -> listener.reconfigure(encode(request))).thenApply(this::decode);
}
 

@Override
public void registerReconfigureHandler(Function<ReconfigureRequest, CompletableFuture<ReconfigureResponse>> handler) {
  this.reconfigureHandler = handler;
}
 

/**
 * Handles a configure request.
 *
 * @param request The request to handle.
 * @return A completable future to be completed with the request response.
 */
CompletableFuture<ReconfigureResponse> onReconfigure(ReconfigureRequest request);