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
/**
	 * Loads the savepoint at the specified path. This methods returns the savepoint, as well as the
	 * handle to the metadata.
	 *
	 * @param savepointFileOrDirectory Path to the parent savepoint directory or the meta data file.
	 * @param classLoader The class loader used to resolve serialized classes from legacy savepoint formats.
	 * @return The loaded savepoint
	 *
	 * @throws IOException Failures during load are forwarded
	 */
Loads the savepoint at the specified path. This methods returns the savepoint, as well as the
handle to the metadata.

@param savepointFileOrDirectory Path to the parent savepoint directory or the meta data file.
@param classLoader The class loader used to resolve serialized classes from legacy savepoint formats.
@return The loaded savepoint

@throws IOException Failures during load are forwarded
Loads the savepoint at the specified path . This methods returns the savepoint as well as the handle to the metadata . @param savepointFileOrDirectory Path to the parent savepoint directory or the meta data file . @param classLoader The class loader used to resolve serialized classes from legacy savepoint formats . @return The loaded savepoint @throws IOException Failures during load are forwarded
public static Tuple2<Savepoint, StreamStateHandle> loadSavepointWithHandle(
			String savepointFileOrDirectory,
			ClassLoader classLoader) throws IOException {
		checkNotNull(savepointFileOrDirectory, "savepointFileOrDirectory");
		checkNotNull(classLoader, "classLoader");
		Path path = new Path(savepointFileOrDirectory);
		LOG.info("Loading savepoint from {}", path);
		FileSystem fs = FileSystem.get(path.toUri());
		FileStatus status = fs.getFileStatus(path);
		if (status.isDir()) {
			Path candidatePath = new Path(path, SAVEPOINT_METADATA_FILE);
			if (fs.exists(candidatePath)) {
				path = candidatePath;
				LOG.info("Using savepoint file in {}", path);
			} else {
				throw new IOException("Cannot find meta data file in directory " + path
						+ ". Please try to load the savepoint directly from the meta data file "
						+ "instead of the directory.");
			}
		}
		final Savepoint savepoint;
		try (DataInputStream dis = new DataInputViewStreamWrapper(fs.open(path))) {
			int magicNumber = dis.readInt();
			if (magicNumber == MAGIC_NUMBER) {
				int version = dis.readInt();
				SavepointSerializer<?> serializer = SavepointSerializers.getSerializer(version);
				savepoint = serializer.deserialize(dis, classLoader);
			} else {
				throw new RuntimeException("Unexpected magic number. This can have multiple reasons: " +
						"(1) You are trying to load a Flink 1.0 savepoint, which is not supported by this " +
						"version of Flink. (2) The file you were pointing to is not a savepoint at all. " +
						"(3) The savepoint file has been corrupted.");
			}
		}
		long size = 0;
		try {
			size = fs.getFileStatus(path).getLen();
		}
		catch (Exception ignored) {
		}
		StreamStateHandle metadataHandle = new FileStateHandle(path, size);
		return new Tuple2<>(savepoint, metadataHandle);
	}
public static Tuple2 < Savepoint , StreamStateHandle > loadSavepointWithHandle ( String savepointFileOrDirectory , ClassLoader classLoader ) throws IOException { checkNotNull ( savepointFileOrDirectory , "savepointFileOrDirectory" ) ; checkNotNull ( classLoader , "classLoader" ) ; Path path = new Path ( savepointFileOrDirectory ) ; LOG . info ( "Loading savepoint from {}" , path ) ; FileSystem fs = FileSystem . get ( path . toUri ( ) ) ; FileStatus status = fs . getFileStatus ( path ) ; if ( status . isDir ( ) ) { Path candidatePath = new Path ( path , SAVEPOINT_METADATA_FILE ) ; if ( fs . exists ( candidatePath ) ) { path = candidatePath ; LOG . info ( "Using savepoint file in {}" , path ) ; } else { throw new IOException ( "Cannot find meta data file in directory " + path + ". Please try to load the savepoint directly from the meta data file " + "instead of the directory." ) ; } } final Savepoint savepoint ; try ( DataInputStream dis = new DataInputViewStreamWrapper ( fs . open ( path ) ) ) { int magicNumber = dis . readInt ( ) ; if ( magicNumber == MAGIC_NUMBER ) { int version = dis . readInt ( ) ; SavepointSerializer < ? > serializer = SavepointSerializers . getSerializer ( version ) ; savepoint = serializer . deserialize ( dis , classLoader ) ; } else { throw new RuntimeException ( "Unexpected magic number. This can have multiple reasons: " + "(1) You are trying to load a Flink 1.0 savepoint, which is not supported by this " + "version of Flink. (2) The file you were pointing to is not a savepoint at all. " + "(3) The savepoint file has been corrupted." ) ; } } long size = 0 ; try { size = fs . getFileStatus ( path ) . getLen ( ) ; } catch ( Exception ignored ) { } StreamStateHandle metadataHandle = new FileStateHandle ( path , size ) ; return new Tuple2 < > ( savepoint , metadataHandle ) ; }
Loads the savepoint at the specified path.
Loads the savepoint at the specified path .
// If this is a directory, we need to find the meta data file // load the savepoint // construct the stream handle to the metadata file // we get the size best-effort // we don't know the size, but we don't want to fail the savepoint loading for that