@Test
public void checkExecutorHealthConsecutiveFailures() throws Exception {

  // By default mocked methods will return an empty collection.
  // Therefore underlying call to apiGateway.callWithExecutionId returns an empty Map for all
  // invocations of executorHealthChecker.checkExecutorHealth() in this test.
  this.activeFlows.put(EXECUTION_ID_11, new Pair<>(
      new ExecutionReference(EXECUTION_ID_11, this.executor1), this.flow1));
  // Failed to ping executor. Failure count (=1) < MAX_FAILURE_COUNT (=2). Do not alert.
  this.executorHealthChecker.checkExecutorHealth();
  verify(this.apiGateway).callWithExecutionId(this.executor1.getHost(), this.executor1.getPort(),
      ConnectorParams.PING_ACTION, null, null);
  verifyZeroInteractions(this.alerterHolder);
  // Pinged executor successfully. Failure count (=0) < MAX_FAILURE_COUNT (=2). Do not alert.
  when(this.apiGateway.callWithExecutionId(this.executor1.getHost(), this.executor1.getPort(),
      ConnectorParams.PING_ACTION, null, null)).thenReturn(ImmutableMap.of(ConnectorParams
      .STATUS_PARAM, ConnectorParams.RESPONSE_ALIVE));
  this.executorHealthChecker.checkExecutorHealth();
  verifyZeroInteractions(this.alerterHolder);
  // Failed to ping executor. Failure count (=1) < MAX_FAILURE_COUNT (=2). Do not alert.
  when(this.apiGateway.callWithExecutionId(this.executor1.getHost(), this.executor1.getPort(),
      ConnectorParams.PING_ACTION, null, null)).thenReturn(null);
  this.executorHealthChecker.checkExecutorHealth();
  verifyZeroInteractions(this.alerterHolder);
  // Failed to ping executor again. Failure count (=2) = MAX_FAILURE_COUNT (=2). Alert AZ admin.
  when(this.loader.fetchExecutableFlow(flow1.getExecutionId())).thenReturn(flow1);
  this.executorHealthChecker.checkExecutorHealth();
  verify((this.alerterHolder).get("email"))
      .alertOnFailedExecutorHealthCheck(eq(this.executor1), eq(Arrays.asList(this.flow1)),
          any(ExecutorManagerException.class),
          eq(Arrays.asList(AZ_ADMIN_ALERT_EMAIL.split(","))));
  // Verify remediation tasks are performed for unreachable executors.
  // Flow should be finalized with alerts sent over email.
  assertThat(this.flow1.getStatus()).isEqualTo(Status.FAILED);
  String expectedReason = "Executor was unreachable, executor-id: 1, executor-host: localhost, "
      + "executor-port: 12345";
  verify((this.alerterHolder).get("email")).alertOnError(eq(flow1), eq(expectedReason));
}