private static DescribeQuorumResponseData.ReplicaState toReplicaState(
        short apiVersion,
        int leaderId,
        LeaderState.ReplicaState replicaState,
        long currentTimeMs
    ) {
        final long lastCaughtUpTimestamp;
        final long lastFetchTimestamp;
        if (replicaState.replicaKey().id() == leaderId) {
            lastCaughtUpTimestamp = currentTimeMs;
            lastFetchTimestamp = currentTimeMs;
        } else {
            lastCaughtUpTimestamp = replicaState.lastCaughtUpTimestamp();
            lastFetchTimestamp = replicaState.lastFetchTimestamp();
        }
        DescribeQuorumResponseData.ReplicaState replicaStateData = new DescribeQuorumResponseData.ReplicaState()
            .setReplicaId(replicaState.replicaKey().id())
            .setLogEndOffset(replicaState.endOffset().map(LogOffsetMetadata::offset).orElse(-1L))
            .setLastCaughtUpTimestamp(lastCaughtUpTimestamp)
            .setLastFetchTimestamp(lastFetchTimestamp);

        if (apiVersion >= 2) {
            replicaStateData.setReplicaDirectoryId(replicaState.replicaKey().directoryId().orElse(ReplicaKey.NO_DIRECTORY_ID));
        }
        return replicaStateData;
    }
Creates a FetchSnapshotResponseData with a single PartitionSnapshot for the topic partition.

The partition index will already be populated when calling operator.

@param listenerName the listener used to accept the request
@param apiVersion the api version of the request
@param topicPartition the topic partition to include
@param leaderId the id of the leader
@param endpoints the endpoints of the leader
@param operator unary operator responsible for populating all of the appropriate fields
@return the created fetch snapshot response data
raft/src/main/java/org/apache/kafka/raft/RaftUtil.java