@ Override public void recover ( ) throws Exception { LOG . info ( "Recovering checkpoints from ZooKeeper." ) ; // Get all there is first List < Tuple2 < RetrievableStateHandle < CompletedCheckpoint > , String > > initialCheckpoints ; while ( true ) { try { initialCheckpoints = checkpointsInZooKeeper . getAllAndLock ( ) ; break ; } catch ( ConcurrentModificationException e ) { LOG . warn ( "Concurrent modification while reading from ZooKeeper. Retrying." ) ; } } Collections . sort ( initialCheckpoints , STRING_COMPARATOR ) ; int numberOfInitialCheckpoints = initialCheckpoints . size ( ) ; LOG . info ( "Found {} checkpoints in ZooKeeper." , numberOfInitialCheckpoints ) ; // 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. 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 && ! CompletedCheckpoint . checkpointsMatch ( lastTryRetrievedCheckpoints , 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 + " checkpoints 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 .