flink-instrumentation/workspace/flink-1.4.1-instrumented/flink-1.4.1/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/savepoint/SavepointStore.java
/**
	 * Stores the savepoint metadata file to a state handle.
	 *
	 * @param directory Target directory to store savepoint in
	 * @param savepoint Savepoint to be stored
	 *
	 * @return State handle to the checkpoint metadata
	 * @throws IOException Failures during store are forwarded
	 */
Stores the savepoint metadata file to a state handle.
@param directory Target directory to store savepoint in
@param savepoint Savepoint to be stored

@return State handle to the checkpoint metadata
@throws IOException Failures during store are forwarded
Stores the savepoint metadata file to a state handle . @param directory Target directory to store savepoint in @param savepoint Savepoint to be stored @return State handle to the checkpoint metadata @throws IOException Failures during store are forwarded
static <T extends Savepoint> FileStateHandle storeSavepointToHandle(
			String directory,
			String filename,
			T savepoint) throws IOException {
		checkNotNull(directory, "Target directory");
		checkNotNull(savepoint, "Savepoint");
		final Path basePath = new Path(directory);
		final Path metadataFilePath = new Path(basePath, filename);
		final FileSystem fs = FileSystem.get(basePath.toUri());
		boolean success = false;
		try (FSDataOutputStream fdos = fs.create(metadataFilePath, WriteMode.NO_OVERWRITE);
				DataOutputStream dos = new DataOutputStream(fdos))
		{
			dos.writeInt(MAGIC_NUMBER);
			dos.writeInt(savepoint.getVersion());
			SavepointSerializer<T> serializer = SavepointSerializers.getSerializer(savepoint);
			serializer.serialize(savepoint, dos);
			FileStateHandle handle = new FileStateHandle(metadataFilePath, dos.size());
			success = true;
			return handle;
		}
		finally {
			if (!success && fs.exists(metadataFilePath)) {
				if (!fs.delete(metadataFilePath, true)) {
					LOG.warn("Failed to delete file {} after failed metadata write.", metadataFilePath);
				}
			}
		}
	}
static < T extends Savepoint > FileStateHandle storeSavepointToHandle ( String directory , String filename , T savepoint ) throws IOException { checkNotNull ( directory , "Target directory" ) ; checkNotNull ( savepoint , "Savepoint" ) ; final Path basePath = new Path ( directory ) ; final Path metadataFilePath = new Path ( basePath , filename ) ; final FileSystem fs = FileSystem . get ( basePath . toUri ( ) ) ; boolean success = false ; try ( FSDataOutputStream fdos = fs . create ( metadataFilePath , WriteMode . NO_OVERWRITE ) ; DataOutputStream dos = new DataOutputStream ( fdos ) ) { dos . writeInt ( MAGIC_NUMBER ) ; dos . writeInt ( savepoint . getVersion ( ) ) ; SavepointSerializer < T > serializer = SavepointSerializers . getSerializer ( savepoint ) ; serializer . serialize ( savepoint , dos ) ; FileStateHandle handle = new FileStateHandle ( metadataFilePath , dos . size ( ) ) ; success = true ; return handle ; } finally { if ( ! success && fs . exists ( metadataFilePath ) ) { if ( ! fs . delete ( metadataFilePath , true ) ) { LOG . warn ( "Failed to delete file {} after failed metadata write." , metadataFilePath ) ; } } } }
Stores the savepoint metadata file to a state handle.
Stores the savepoint metadata file to a state handle .
// Write header // Write savepoint // construct result handle // all good!