### Question:
SavepointOutputFormat extends RichOutputFormat<Savepoint> { public SavepointOutputFormat(Path savepointPath) { this.savepointPath = savepointPath; } SavepointOutputFormat(Path savepointPath); @Override void configure(Configuration parameters); @Override void open(int taskNumber, int numTasks); @Override void writeRecord(Savepoint savepoint); @Override void close();  }### Answer:
@Test public void testSavepointOutputFormat() throws Exception { Path path = new Path(temporaryFolder.newFolder().getAbsolutePath()); SavepointOutputFormat format = createSavepointOutputFormat(path); Savepoint savepoint = createSavepoint(); format.open(0, 1); format.writeRecord(savepoint); format.close(); Savepoint savepointOnDisk = SavepointLoader.loadSavepoint(path.getPath()); Assert.assertEquals( "Incorrect checkpoint id", savepoint.getCheckpointId(), savepointOnDisk.getCheckpointId()); Assert.assertEquals( "Incorrect number of operator states in savepoint", savepoint.getOperatorStates().size(), savepointOnDisk.getOperatorStates().size()); Assert.assertEquals( "Incorrect operator state in savepoint", savepoint.getOperatorStates().iterator().next(), savepointOnDisk.getOperatorStates().iterator().next()); }