private FetchResponseData buildFetchResponse(
        ListenerName listenerName,
        short apiVersion,
        Errors error,
        Records records,
        ValidOffsetAndEpoch validOffsetAndEpoch,
        Optional<LogOffsetMetadata> highWatermark
    ) {
        return RaftUtil.singletonFetchResponse(
            listenerName,
            apiVersion,
            log.topicPartition(),
            log.topicId(),
            Errors.NONE,
            quorum.leaderIdOrSentinel(),
            quorum.leaderEndpoints(),
            partitionData -> {
                partitionData
                    .setRecords(records)
                    .setErrorCode(error.code())
                    .setLogStartOffset(log.startOffset())
                    .setHighWatermark(
                        highWatermark.map(LogOffsetMetadata::offset).orElse(-1L)
                    );

                partitionData.currentLeader()
                    .setLeaderEpoch(quorum.epoch())
                    .setLeaderId(quorum.leaderIdOrSentinel());

                switch (validOffsetAndEpoch.kind()) {
                    case DIVERGING:
                        partitionData.divergingEpoch()
                            .setEpoch(validOffsetAndEpoch.offsetAndEpoch().epoch())
                            .setEndOffset(validOffsetAndEpoch.offsetAndEpoch().offset());
                        break;
                    case SNAPSHOT:
                        partitionData.snapshotId()
                            .setEpoch(validOffsetAndEpoch.offsetAndEpoch().epoch())
                            .setEndOffset(validOffsetAndEpoch.offsetAndEpoch().offset());
                        break;
                    default:
                }
            }
        );
    }
Handle an EndEpoch request. 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
raft/src/main/java/org/apache/kafka/raft/KafkaRaftClient.java