private CompletableFuture<FetchResponseData> handleFetchRequest(
        RaftRequest.Inbound requestMetadata,
        long currentTimeMs
    ) {
        FetchRequestData request = (FetchRequestData) requestMetadata.data();

        if (!hasValidClusterId(request.clusterId())) {
            return completedFuture(new FetchResponseData().setErrorCode(Errors.INCONSISTENT_CLUSTER_ID.code()));
        }

        if (!hasValidTopicPartition(request, log.topicPartition(), log.topicId())) {
            // Until we support multi-raft, we treat topic partition mismatches as invalid requests
            return completedFuture(new FetchResponseData().setErrorCode(Errors.INVALID_REQUEST.code()));
        }
        // If the ID is valid, we can set the topic name.
        request.topics().get(0).setTopic(log.topicPartition().topic());

        FetchRequestData.FetchPartition fetchPartition = request.topics().get(0).partitions().get(0);
        if (request.maxWaitMs() < 0
            || fetchPartition.fetchOffset() < 0
            || fetchPartition.lastFetchedEpoch() < 0
            || fetchPartition.lastFetchedEpoch() > fetchPartition.currentLeaderEpoch()) {
            return completedFuture(
                buildEmptyFetchResponse(
                    requestMetadata.listenerName(),
                    requestMetadata.apiVersion(),
                    Errors.INVALID_REQUEST,
                    Optional.empty()
                )
            );
        }

        ReplicaKey replicaKey = ReplicaKey.of(
            FetchRequest.replicaId(request),
            fetchPartition.replicaDirectoryId()
        );
        FetchResponseData response = tryCompleteFetchRequest(
            requestMetadata.listenerName(),
            requestMetadata.apiVersion(),
            replicaKey,
            fetchPartition,
            currentTimeMs
        );
        FetchResponseData.PartitionData partitionResponse =
            response.responses().get(0).partitions().get(0);

        if (partitionResponse.errorCode() != Errors.NONE.code()
            || FetchResponse.recordsSize(partitionResponse) > 0
            || request.maxWaitMs() == 0
            || isPartitionDiverged(partitionResponse)
            || isPartitionSnapshotted(partitionResponse)) {
            // Reply immediately if any of the following is true
            // 1. The response contains an error
            // 2. There are records in the response
            // 3. The fetching replica doesn't want to wait for the partition to contain new data
            // 4. The fetching replica needs to truncate because the log diverged
            // 5. The fetching replica needs to fetch a snapshot
            return completedFuture(response);
        }

        CompletableFuture<Long> future = fetchPurgatory.await(
            fetchPartition.fetchOffset(),
            request.maxWaitMs());

        return future.handle((completionTimeMs, exception) -> {
            if (exception != null) {
                Throwable cause = exception instanceof ExecutionException ?
                    exception.getCause() : exception;

                Errors error = Errors.forException(cause);
                if (error == Errors.REQUEST_TIMED_OUT) {
                    // Note that for this case the calling thread is the expiration service thread and not the
                    // polling thread.
                    //
                    // If the fetch request timed out in purgatory, it means no new data is available,
                    // just return the original fetch response.
                    return response;
                } else {
                    // If there was any error other than REQUEST_TIMED_OUT, return it.
                    logger.info(
                        "Failed to handle fetch from {} at {} due to {}",
                        replicaKey,
                        fetchPartition.fetchOffset(),
                        error
                    );
                    return buildEmptyFetchResponse(
                        requestMetadata.listenerName(),
                        requestMetadata.apiVersion(),
                        error,
                        Optional.empty()
                    );
                }
            }

            // FIXME: `completionTimeMs`, which can be null
            logger.trace(
                "Completing delayed fetch from {} starting at offset {} at {}",
                replicaKey,
                fetchPartition.fetchOffset(),
                completionTimeMs
            );

            // It is safe to call tryCompleteFetchRequest because only the polling thread completes this
            // future successfully. This is true because only the polling thread appends record batches to
            // the log from maybeAppendBatches.
            return tryCompleteFetchRequest(
                requestMetadata.listenerName(),
                requestMetadata.apiVersion(),
                replicaKey,
                fetchPartition,
                time.milliseconds()
            );
        });
    }
Handle a Fetch request. The fetch offset and last fetched epoch are always
validated against the current log. In the case that they do not match, the response will
indicate the diverging offset/epoch. A follower is expected to truncate its log in this
case and resend the fetch.

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
raft/src/main/java/org/apache/kafka/raft/KafkaRaftClient.java