@Override
    public FileMergingCheckpointStateOutputStream createCheckpointStateOutputStream(
            CheckpointedStateScope scope) throws IOException {
        return fileMergingSnapshotManager.createCheckpointStateOutputStream(
                subtaskKey, checkpointId, scope);
    }
@Test
    public void testWritingToClosedStream() {
        FileMergingSnapshotManager snapshotManager = createFileMergingSnapshotManager();
        FsMergingCheckpointStorageLocation storageLocation =
                createFsMergingCheckpointStorageLocation(1, snapshotManager);
        try (FileMergingCheckpointStateOutputStream stream =
                storageLocation.createCheckpointStateOutputStream(EXCLUSIVE)) {
            stream.flushToFile();
            stream.closeAndGetHandle();
            stream.flushToFile();
            fail("Expected IOException");
        } catch (IOException e) {
            assertThat(e.getMessage()).isEqualTo("Cannot call flushToFile() to a closed stream.");
        }
    }