private CompletableFuture<RemoveRaftVoterResponseData> handleRemoveVoterRequest(
        RaftRequest.Inbound requestMetadata,
        long currentTimeMs
    ) {
        RemoveRaftVoterRequestData data = (RemoveRaftVoterRequestData) requestMetadata.data();

        if (!hasValidClusterId(data.clusterId())) {
            return completedFuture(
                new RemoveRaftVoterResponseData()
                    .setErrorCode(Errors.INCONSISTENT_CLUSTER_ID.code())
                    .setErrorMessage(
                        String.format(
                            "The given id \"%s\" doesn't match the cluster id \"%s\"",
                            data.clusterId(),
                            clusterId
                        )
                    )
            );
        }

        Optional<Errors> leaderValidation = validateLeaderOnlyRequest(quorum.epoch());
        if (leaderValidation.isPresent()) {
            return completedFuture(
                new RemoveRaftVoterResponseData().setErrorCode(leaderValidation.get().code())
            );
        }

        Optional<ReplicaKey> oldVoter = RaftUtil.removeVoterRequestVoterKey(data);
        if (!oldVoter.isPresent() || !oldVoter.get().directoryId().isPresent()) {
            return completedFuture(
                new RemoveRaftVoterResponseData()
                    .setErrorCode(Errors.INVALID_REQUEST.code())
                    .setErrorMessage("Remove voter request didn't include a valid voter")
            );
        }

        return removeVoterHandler.handleRemoveVoterRequest(
            quorum.leaderStateOrThrow(),
            oldVoter.get(),
            currentTimeMs
        );
    }
Handle a FetchSnapshot request, similar to the Fetch request but we use {@link UnalignedRecords}
in response because the records are not necessarily offset-aligned.

This API may return the following errors:

- {@link Errors#INCONSISTENT_CLUSTER_ID} if the cluster id is presented in request
    but different from this node
- {@link Errors#BROKER_NOT_AVAILABLE} if this node is currently shutting down
- {@link Errors#FENCED_LEADER_EPOCH} if the epoch is smaller than this node's epoch
- {@link Errors#INVALID_REQUEST} if the request epoch is larger than the leader's current epoch
    or if either the fetch offset or the last fetched epoch is invalid
- {@link Errors#SNAPSHOT_NOT_FOUND} if the request snapshot id does not exists
- {@link Errors#POSITION_OUT_OF_RANGE} if the request snapshot offset out of range
raft/src/main/java/org/apache/kafka/raft/KafkaRaftClient.java