RunningExecutionsUpdater { @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."); } @Inject  RunningExecutionsUpdater(final ExecutorManagerUpdaterStage updaterStage,
      final AlerterHolder alerterHolder, final CommonMetrics commonMetrics,
      final ExecutorApiGateway apiGateway, final RunningExecutions runningExecutions,
      final ExecutionFinalizer executionFinalizer, final ExecutorLoader executorLoader); @SuppressWarnings("unchecked") void updateExecutions();  }
@Test public void updateExecutionsStillRunning() throws Exception { mockFlowStillRunning(); this.updater.updateExecutions(); verifyCallUpdateApi(); verifyZeroInteractions(this.executionFinalizer); }
@Test public void updateExecutionsSucceeded() throws Exception { mockFlowSucceeded(); this.updater.updateExecutions(); verifyCallUpdateApi(); verifyFinalizeFlow(); }
@Test public void updateExecutionsExecutorDoesNotExist() throws Exception { mockExecutorDoesNotExist(); this.updater.updateExecutions(); verifyFinalizeFlow(); }
@Test public void updateExecutionsFlowDoesNotExist() throws Exception { mockFlowDoesNotExist(); this.updater.updateExecutions(); verifyCallUpdateApi(); verifyFinalizeFlow(); }
@Test public void updateExecutionsUpdateCallFails() throws Exception { mockUpdateCallFails(); when(this.executorLoader.fetchExecutor(anyInt())).thenReturn(this.activeExecutor); DateTimeUtils.setCurrentMillisFixed(System.currentTimeMillis()); for (int i = 0; i < this.updater.numErrorsBeforeUnresponsiveEmail; i++) { this.updater.updateExecutions(); DateTimeUtils.setCurrentMillisFixed( DateTimeUtils.currentTimeMillis() + this.updater.errorThreshold + 1L); } verify(this.mailAlerter).alertOnFailedUpdate( this.activeExecutor, Collections.singletonList(this.execution), API_CALL_EXCEPTION); verifyZeroInteractions(this.executionFinalizer); }
@Test public void updateExecutionsUpdateCallFailsExecutorDoesntExist() throws Exception { mockUpdateCallFails(); when(this.executorLoader.fetchExecutor(anyInt())).thenReturn(null); DateTimeUtils.setCurrentMillisFixed(System.currentTimeMillis()); this.updater.updateExecutions(); verify(this.executionFinalizer).finalizeFlow( this.execution, "Not running on the assigned executor (any more)", null); }
@Test public void updateExecutionsUpdateCallFailsExecutorCheckThrows() throws Exception { mockUpdateCallFails(); when(this.executorLoader.fetchExecutor(anyInt())) .thenThrow(new ExecutorManagerException("Mocked fetchExecutor failure")); DateTimeUtils.setCurrentMillisFixed(System.currentTimeMillis()); this.updater.updateExecutions(); verifyZeroInteractions(this.executionFinalizer); }