Emailer extends AbstractMailer implements Alerter { @Override public void alertOnFailedUpdate(final Executor executor, List<ExecutableFlow> flows, final ExecutorManagerException updateException) { flows = flows.stream() .filter(flow -> flow.getExecutionOptions() != null) .filter(flow -> CollectionUtils.isNotEmpty(flow.getExecutionOptions().getFailureEmails())) .collect(Collectors.toList()); final ImmutableListMultimap<String, ExecutableFlow> creatorsToFlows = Multimaps .index(flows, flow -> flow.getExecutionOptions().getMailCreator()); for (final String mailCreatorName : creatorsToFlows.keySet()) { final ImmutableList<ExecutableFlow> creatorFlows = creatorsToFlows.get(mailCreatorName); final MailCreator mailCreator = getMailCreator(mailCreatorName); final ImmutableListMultimap<List<String>, ExecutableFlow> emailsToFlows = Multimaps .index(creatorFlows, flow -> flow.getExecutionOptions().getFailureEmails()); for (final List<String> emailList : emailsToFlows.keySet()) { sendFailedUpdateEmail(executor, updateException, mailCreator, emailsToFlows.get(emailList)); } } } @Inject  Emailer(final Props props, final CommonMetrics commonMetrics,
      final EmailMessageCreator messageCreator, final ExecutorLoader executorLoader); String getAzkabanURL(); void sendEmail(final List<String> emailList, final String subject, final String body); @Override void alertOnSla(final SlaOption slaOption, final String slaMessage); @Override void alertOnFirstError(final ExecutableFlow flow); @Override void alertOnError(final ExecutableFlow flow, final String... extraReasons); @Override void alertOnSuccess(final ExecutableFlow flow); @Override void alertOnFailedUpdate(final Executor executor, List<ExecutableFlow> flows,
      final ExecutorManagerException updateException); @Override void alertOnFailedExecutorHealthCheck(final Executor executor,
      final List<ExecutableFlow> flows, final ExecutorManagerException failureException,
      final List<String> emailList); void sendEmail(final EmailMessage message, final boolean mailCreated,
      final String operation);  }
@Test public void alertOnFailedUpdate() throws Exception { final Flow flow = this.project.getFlow("jobe"); flow.addFailureEmails(this.receiveAddrList); Assert.assertNotNull(flow); final ExecutableFlow exFlow = new ExecutableFlow(this.project, flow); final CommonMetrics commonMetrics = new CommonMetrics(new MetricsManager(new MetricRegistry())); final Emailer emailer = new Emailer(this.props, commonMetrics, this.messageCreator, this.executorLoader); final Executor executor = new Executor(1, "executor1-host", 1234, true); final List<ExecutableFlow> executions = Arrays.asList(exFlow, exFlow); final ExecutorManagerException exception = DefaultMailCreatorTest.createTestStracktrace(); emailer.alertOnFailedUpdate(executor, executions, exception); verify(this.message).addAllToAddress(this.receiveAddrList); verify(this.message) .setSubject("Flow status could not be updated from executor1-host on azkaban"); assertThat(TestUtils.readResource("failedUpdateMessage2.html", this)) .isEqualToIgnoringWhitespace(this.message.getBody()); }