@ SuppressWarnings ( "unchecked" ) public void updateExecutions ( ) { this . updaterStage . set ( "Starting update all flows." ) ; final Map < Optional < Executor > , List < ExecutableFlow > > exFlowMap = getFlowToExecutorMap ( ) ; final ArrayList < ExecutableFlow > finalizeFlows = new ArrayList <> ( ) ; for ( final Map . Entry < Optional < Executor > , List < ExecutableFlow > > entry : exFlowMap . entrySet ( ) ) { final Optional < Executor > executorOption = entry . getKey ( ) ; if ( ! executorOption . isPresent ( ) ) { for ( final ExecutableFlow flow : entry . getValue ( ) ) { logger . warn ( "Finalizing execution " + flow . getExecutionId ( ) + ". Executor id of this execution doesn't exist" ) ; finalizeFlows . add ( flow ) ; } continue ; } final Executor executor = executorOption . get ( ) ; this . updaterStage . set ( "Starting update flows on " + executor . getHost ( ) + ":" + executor . getPort ( ) ) ; Map < String , Object > results = null ; try { results = this . apiGateway . updateExecutions ( executor , entry . getValue ( ) ) ; } catch ( final ExecutorManagerException e ) { handleException ( entry , executor , e , finalizeFlows ) ; } if ( results != null ) { final List < Map < String , Object > > executionUpdates = ( List < Map < String , Object > > ) results . get ( ConnectorParams . RESPONSE_UPDATED_FLOWS ) ; for ( final Map < String , Object > updateMap : executionUpdates ) { try { final ExecutableFlow flow = updateExecution ( updateMap ) ; this . updaterStage . set ( "Updated flow " + flow . getExecutionId ( ) ) ; if ( ExecutionControllerUtils . isFinished ( flow ) ) { finalizeFlows . add ( flow ) ; } } catch ( final ExecutorManagerException e ) { final ExecutableFlow flow = e . getExecutableFlow ( ) ; logger . error ( e ) ; if ( flow != null ) { logger . warn ( "Finalizing execution " + flow . getExecutionId ( ) ) ; finalizeFlows . add ( flow ) ; } } } } } this . updaterStage . set ( "Finalizing " + finalizeFlows . size ( ) + " error flows." ) ; for ( final ExecutableFlow flow : finalizeFlows ) { this . executionFinalizer . finalizeFlow ( flow , "Not running on the assigned executor (any more)" , null ) ; } this . updaterStage . set ( "Updated all active flows. Waiting for next round." ) ; }
Updates running executions .