@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
    void testExclusiveStateHasRelativePathHandles() throws IOException {
        final FsCheckpointStreamFactory factory = createFactory(FileSystem.getLocalFileSystem(), 0);

        final FsCheckpointStreamFactory.FsCheckpointStateOutputStream stream =
                factory.createCheckpointStateOutputStream(CheckpointedStateScope.EXCLUSIVE);
        stream.write(1657);
        final StreamStateHandle handle = stream.closeAndGetHandle();

        assertThat(handle).isInstanceOf(RelativeFileStateHandle.class);
        assertPathsEqual(
                exclusiveStateDir, ((RelativeFileStateHandle) handle).getFilePath().getParent());
    }