PartitionFetchState start(TopicPartition topicPartition,
                              PartitionFetchState currentFetchState,
                              PartitionData fetchPartitionData) throws Exception {
        OffsetAndEpoch epochAndLeaderLocalStartOffset = leader.fetchEarliestLocalOffset(topicPartition, currentFetchState.currentLeaderEpoch());
        int epoch = epochAndLeaderLocalStartOffset.leaderEpoch();
        long leaderLocalStartOffset = epochAndLeaderLocalStartOffset.offset();

        long offsetToFetch;
        replicaMgr.brokerTopicStats().topicStats(topicPartition.topic()).buildRemoteLogAuxStateRequestRate().mark();
        replicaMgr.brokerTopicStats().allTopicsStats().buildRemoteLogAuxStateRequestRate().mark();

        UnifiedLog unifiedLog;
        if (useFutureLog) {
            unifiedLog = replicaMgr.futureLogOrException(topicPartition);
        } else {
            unifiedLog = replicaMgr.localLogOrException(topicPartition);
        }

        try {
            offsetToFetch = buildRemoteLogAuxState(topicPartition, currentFetchState.currentLeaderEpoch(), leaderLocalStartOffset, epoch, fetchPartitionData.logStartOffset(), unifiedLog);
        } catch (RemoteStorageException e) {
            replicaMgr.brokerTopicStats().topicStats(topicPartition.topic()).failedBuildRemoteLogAuxStateRate().mark();
            replicaMgr.brokerTopicStats().allTopicsStats().failedBuildRemoteLogAuxStateRate().mark();
            throw e;
        }

        OffsetAndEpoch fetchLatestOffsetResult = leader.fetchLatestOffset(topicPartition, currentFetchState.currentLeaderEpoch());
        long leaderEndOffset = fetchLatestOffsetResult.offset();

        long initialLag = leaderEndOffset - offsetToFetch;

        return PartitionFetchState.apply(currentFetchState.topicId(), offsetToFetch, Option.apply(initialLag), currentFetchState.currentLeaderEpoch(),
                Fetching$.MODULE$, unifiedLog.latestEpoch());

    }
Start the tier state machine for the provided topic partition.

@param topicPartition the topic partition
@param currentFetchState the current PartitionFetchState which will
                         be used to derive the return value
@param fetchPartitionData the data from the fetch response that returned the offset moved to tiered storage error

@return the new PartitionFetchState after the successful start of the
        tier state machine
core/src/main/java/kafka/server/TierStateMachine.java