private void buildProducerSnapshotFile(UnifiedLog unifiedLog,
                                           long nextOffset,
                                           RemoteLogSegmentMetadata remoteLogSegmentMetadata,
                                           RemoteLogManager rlm) throws IOException, RemoteStorageException {
        // Restore producer snapshot
        File snapshotFile = LogFileUtils.producerSnapshotFile(unifiedLog.dir(), nextOffset);
        Path tmpSnapshotFile = Paths.get(snapshotFile.getAbsolutePath() + ".tmp");
        // Copy it to snapshot file in atomic manner.
        Files.copy(rlm.storageManager().fetchIndex(remoteLogSegmentMetadata, RemoteStorageManager.IndexType.PRODUCER_SNAPSHOT),
                tmpSnapshotFile, StandardCopyOption.REPLACE_EXISTING);
        Utils.atomicMoveWithFallback(tmpSnapshotFile, snapshotFile.toPath(), false);

        // Reload producer snapshots.
        unifiedLog.producerStateManager().truncateFullyAndReloadSnapshots();
        unifiedLog.loadProducerState(nextOffset);
    }
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