flink-instrumentation/workspace/flink-1.4.1-instrumented/flink-1.4.1/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/ZooKeeperCompletedCheckpointStore.java
/**
	 * Gets the latest checkpoint from ZooKeeper and removes all others.
	 *
	 * <p><strong>Important</strong>: Even if there are more than one checkpoint in ZooKeeper,
	 * this will only recover the latest and discard the others. Otherwise, there is no guarantee
	 * that the history of checkpoints is consistent.
	 */
Gets the latest checkpoint from ZooKeeper and removes all others.
Important: Even if there are more than one checkpoint in ZooKeeper,
this will only recover the latest and discard the others. Otherwise, there is no guarantee
that the history of checkpoints is consistent.
Gets the latest checkpoint from ZooKeeper and removes all others . Important : Even if there are more than one checkpoint in ZooKeeper this will only recover the latest and discard the others . Otherwise there is no guarantee that the history of checkpoints is consistent .
@Override
	public void recover() throws Exception {
		LOG.info("Recovering checkpoints from ZooKeeper.");
		List<Tuple2<RetrievableStateHandle<CompletedCheckpoint>, String>> initialCheckpoints;
		while (true) {
			try {
				initialCheckpoints = checkpointsInZooKeeper.getAllSortedByNameAndLock();
				break;
			}
			catch (ConcurrentModificationException e) {
				LOG.warn("Concurrent modification while reading from ZooKeeper. Retrying.");
			}
		}
		int numberOfInitialCheckpoints = initialCheckpoints.size();
		LOG.info("Found {} checkpoints in ZooKeeper.", numberOfInitialCheckpoints);
		 These considerations are also important in the scope of incremental checkpoints, where
		 we use ref-counting for shared state handles and might accidentally delete shared state
		 of checkpoints that we don't read due to transient storage outages.
		List<CompletedCheckpoint> lastTryRetrievedCheckpoints = new ArrayList<>(numberOfInitialCheckpoints);
		List<CompletedCheckpoint> retrievedCheckpoints = new ArrayList<>(numberOfInitialCheckpoints);
		do {
			LOG.info("Trying to fetch {} checkpoints from storage.", numberOfInitialCheckpoints);
			lastTryRetrievedCheckpoints.clear();
			lastTryRetrievedCheckpoints.addAll(retrievedCheckpoints);
			retrievedCheckpoints.clear();
			for (Tuple2<RetrievableStateHandle<CompletedCheckpoint>, String> checkpointStateHandle : initialCheckpoints) {
				CompletedCheckpoint completedCheckpoint = null;
				try {
					completedCheckpoint = retrieveCompletedCheckpoint(checkpointStateHandle);
					if (completedCheckpoint != null) {
						retrievedCheckpoints.add(completedCheckpoint);
					}
				} catch (Exception e) {
					LOG.warn("Could not retrieve checkpoint, not adding to list of recovered checkpoints.", e);
				}
			}
		} while (retrievedCheckpoints.size() != numberOfInitialCheckpoints &&
			!lastTryRetrievedCheckpoints.equals(retrievedCheckpoints));
		 Clear local handles in order to prevent duplicates on
		 recovery. The local handles should reflect the state
		 of ZooKeeper.
		completedCheckpoints.clear();
		completedCheckpoints.addAll(retrievedCheckpoints);
		if (completedCheckpoints.isEmpty() && numberOfInitialCheckpoints > 0) {
			throw new FlinkException(
				"Could not read any of the " + numberOfInitialCheckpoints + " from storage.");
		} else if (completedCheckpoints.size() != numberOfInitialCheckpoints) {
			LOG.warn(
				"Could only fetch {} of {} checkpoints from storage.",
				completedCheckpoints.size(),
				numberOfInitialCheckpoints);
		}
	}
@ Override public void recover ( ) throws Exception { LOG . info ( "Recovering checkpoints from ZooKeeper." ) ; List < Tuple2 < RetrievableStateHandle < CompletedCheckpoint > , String > > initialCheckpoints ; while ( true ) { try { initialCheckpoints = checkpointsInZooKeeper . getAllSortedByNameAndLock ( ) ; break ; } catch ( ConcurrentModificationException e ) { LOG . warn ( "Concurrent modification while reading from ZooKeeper. Retrying." ) ; } } int numberOfInitialCheckpoints = initialCheckpoints . size ( ) ; LOG . info ( "Found {} checkpoints in ZooKeeper." , numberOfInitialCheckpoints ) ; List < CompletedCheckpoint > lastTryRetrievedCheckpoints = new ArrayList < > ( numberOfInitialCheckpoints ) ; List < CompletedCheckpoint > retrievedCheckpoints = new ArrayList < > ( numberOfInitialCheckpoints ) ; do { LOG . info ( "Trying to fetch {} checkpoints from storage." , numberOfInitialCheckpoints ) ; lastTryRetrievedCheckpoints . clear ( ) ; lastTryRetrievedCheckpoints . addAll ( retrievedCheckpoints ) ; retrievedCheckpoints . clear ( ) ; for ( Tuple2 < RetrievableStateHandle < CompletedCheckpoint > , String > checkpointStateHandle : initialCheckpoints ) { CompletedCheckpoint completedCheckpoint = null ; try { completedCheckpoint = retrieveCompletedCheckpoint ( checkpointStateHandle ) ; if ( completedCheckpoint != null ) { retrievedCheckpoints . add ( completedCheckpoint ) ; } } catch ( Exception e ) { LOG . warn ( "Could not retrieve checkpoint, not adding to list of recovered checkpoints." , e ) ; } } } while ( retrievedCheckpoints . size ( ) != numberOfInitialCheckpoints && ! lastTryRetrievedCheckpoints . equals ( retrievedCheckpoints ) ) ; completedCheckpoints . clear ( ) ; completedCheckpoints . addAll ( retrievedCheckpoints ) ; if ( completedCheckpoints . isEmpty ( ) && numberOfInitialCheckpoints > 0 ) { throw new FlinkException ( "Could not read any of the " + numberOfInitialCheckpoints + " from storage." ) ; } else if ( completedCheckpoints . size ( ) != numberOfInitialCheckpoints ) { LOG . warn ( "Could only fetch {} of {} checkpoints from storage." , completedCheckpoints . size ( ) , numberOfInitialCheckpoints ) ; } }
Gets the latest checkpoint from ZooKeeper and removes all others.
Gets the latest checkpoint from ZooKeeper and removes all others .
// Get all there is first // Try and read the state handles from storage. We try until we either successfully read // all of them or when we reach a stable state, i.e. when we successfully read the same set // of checkpoints in two tries. We do it like this to protect against transient outages // of the checkpoint store (for example a DFS): if the DFS comes online midway through // reading a set of checkpoints we would run the risk of reading only a partial set // of checkpoints while we could in fact read the other checkpoints as well if we retried. // Waiting until a stable state protects against this while also being resilient against // checkpoints being actually unreadable. // // These considerations are also important in the scope of incremental checkpoints, where // we use ref-counting for shared state handles and might accidentally delete shared state // of checkpoints that we don't read due to transient storage outages. // Clear local handles in order to prevent duplicates on // recovery. The local handles should reflect the state // of ZooKeeper.