public static void writeSegmentFinishFile(
            String basePath,
            TieredStoragePartitionId partitionId,
            int subpartitionId,
            int segmentId)
            throws IOException {
        Path segmentFinishDir = getSegmentFinishDirPath(basePath, partitionId, subpartitionId);
        FileSystem fs = segmentFinishDir.getFileSystem();
        Path segmentFinishFile = new Path(segmentFinishDir, String.valueOf(segmentId));
        if (!fs.exists(segmentFinishDir)) {
            fs.mkdirs(segmentFinishDir);
            OutputStream outputStream =
                    fs.create(segmentFinishFile, FileSystem.WriteMode.OVERWRITE);
            outputStream.close();
            return;
        }

        FileStatus[] files = fs.listStatus(segmentFinishDir);
        if (files.length == 0) {
            OutputStream outputStream =
                    fs.create(segmentFinishFile, FileSystem.WriteMode.OVERWRITE);
            outputStream.close();
        } else {
            // To minimize the number of files, each subpartition keeps only a single segment-finish
            // file. For instance, if segment-finish file 5 exists, it indicates that segments 1 to
            // 5 have all been finished.
            // Note that this check requires the file system to ensure that only one file is in the
            // directory can be accessed when renaming a file.
            checkState(files.length == 1, "Wong number of segment-finish files.");
            fs.rename(files[0].getPath(), segmentFinishFile);
        }
    }
@Test
    void testWriteSegmentFinishFile() throws IOException {
        TieredStoragePartitionId partitionId =
                TieredStorageIdMappingUtils.convertId(new ResultPartitionID());
        int subpartitionId = 0;
        int segmentId = 1;
        int newSegmentId = 5;

        writeAndCheckSegmentFinishFile(
                tempFolder.getPath(), partitionId, subpartitionId, segmentId);
        writeAndCheckSegmentFinishFile(
                tempFolder.getPath(), partitionId, subpartitionId, newSegmentId);
    }