@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.");
  }
@ 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.