@Override
    public FsCheckpointStateOutputStream createCheckpointStateOutputStream(
            CheckpointedStateScope scope) throws IOException {
        Path target = getTargetPath(scope);
        int bufferSize = Math.max(writeBufferSize, fileStateThreshold);

        // Whether the file system dynamically injects entropy into the file paths.
        final boolean entropyInjecting = EntropyInjector.isEntropyInjecting(filesystem, target);
        final boolean absolutePath = entropyInjecting || scope == CheckpointedStateScope.SHARED;
        return new FsCheckpointStateOutputStream(
                target, filesystem, bufferSize, fileStateThreshold, !absolutePath);
    }
@Test
    @SuppressWarnings("ConstantConditions")
    void testWriteFlushesIfAboveThreshold() throws IOException {
        int fileSizeThreshold = 100;
        final FsCheckpointStreamFactory factory =
                createFactory(
                        FileSystem.getLocalFileSystem(), fileSizeThreshold, fileSizeThreshold);
        final FsCheckpointStreamFactory.FsCheckpointStateOutputStream stream =
                factory.createCheckpointStateOutputStream(CheckpointedStateScope.EXCLUSIVE);
        stream.write(new byte[fileSizeThreshold]);
        File[] files = new File(exclusiveStateDir.toUri()).listFiles();
        assertThat(files).hasSize(1);
        File file = files[0];
        assertThat(file).hasSize(fileSizeThreshold);
        stream.write(new byte[fileSizeThreshold - 1]); // should buffer without flushing
        stream.write(127); // should buffer without flushing
        assertThat(file).hasSize(fileSizeThreshold);
    }