@ Override public synchronized Set < String > recoverEvaluators ( ) { final Set < String > expectedContainers = new HashSet <> ( ) ; try { if ( this . fileSystem == null || this . changeLogLocation == null ) { LOG . log ( Level . WARNING , "Unable to recover evaluators due to failure to instantiate FileSystem. Returning an" + " empty set." ) ; return expectedContainers ; } try ( final CloseableIterable < String > evaluatorLogIterable = readerWriter . readFromEvaluatorLog ( ) ) { for ( final String line : evaluatorLogIterable ) { if ( line . startsWith ( ADD_FLAG ) ) { final String containerId = line . substring ( ADD_FLAG . length ( ) ) ; if ( expectedContainers . contains ( containerId ) ) { LOG . log ( Level . WARNING , "Duplicated add container record found in the change log for container " + containerId ) ; } else { expectedContainers . add ( containerId ) ; } } else if ( line . startsWith ( REMOVE_FLAG ) ) { final String containerId = line . substring ( REMOVE_FLAG . length ( ) ) ; if ( ! expectedContainers . contains ( containerId ) ) { LOG . log ( Level . WARNING , "Change log includes record that try to remove non-exist or duplicate " + "remove record for container + " + containerId ) ; } expectedContainers . remove ( containerId ) ; } } } } catch ( final Exception e ) { final String errMsg = "Cannot read from log file with Exception " + e + ", evaluators will not be recovered." ; final String fatalMsg = "Cannot read from evaluator log." ; this . handleException ( e , errMsg , fatalMsg ) ; } return expectedContainers ; }
Recovers the set of evaluators that are alive .