public SavepointOutputFormat(Path savepointPath) {
        this.savepointPath = savepointPath;
    }
@Test
    public void testSavepointOutputFormat() throws Exception {
        Path path = new Path(temporaryFolder.newFolder().getAbsolutePath());
        SavepointOutputFormat format = createSavepointOutputFormat(path);

        CheckpointMetadata metadata = createSavepoint();

        format.open(0, 1);
        format.writeRecord(metadata);
        format.close();

        CheckpointMetadata metadataOnDisk = SavepointLoader.loadSavepointMetadata(path.getPath());

        Assert.assertEquals(
                "Incorrect checkpoint id",
                metadata.getCheckpointId(),
                metadataOnDisk.getCheckpointId());

        Assert.assertEquals(
                "Incorrect number of operator states in savepoint",
                metadata.getOperatorStates().size(),
                metadataOnDisk.getOperatorStates().size());

        Assert.assertEquals(
                "Incorrect operator state in savepoint",
                metadata.getOperatorStates().iterator().next(),
                metadataOnDisk.getOperatorStates().iterator().next());
    }