@Override
    public FileMergingCheckpointStateOutputStream createCheckpointStateOutputStream(
            CheckpointedStateScope scope) throws IOException {
        return fileMergingSnapshotManager.createCheckpointStateOutputStream(
                subtaskKey, checkpointId, scope);
    }
@Test
    public void testCheckpointStreamClosedExceptionally() throws Exception {
        try (FileMergingSnapshotManager snapshotManager = createFileMergingSnapshotManager()) {
            Path filePath1 = null;
            try (FileMergingCheckpointStateOutputStream stream1 =
                    snapshotManager.createCheckpointStateOutputStream(SUBTASK_KEY, 1, EXCLUSIVE)) {
                stream1.flushToFile();
                filePath1 = stream1.getFilePath();
                assertPathNotNullAndCheckExistence(filePath1, true);
                throw new IOException();
            } catch (IOException ignored) {
            }
            assertPathNotNullAndCheckExistence(filePath1, false);
        }
    }