if (!unitOfWorkBoundaryReached) {
    String flowName = context.flowParameters['_flowName'];
    String envr = context.flowParameters['_agentName'];
    String host = context.flowParameters['_host'];
    subject = host + " - agent: " + envr + " - SymmetricDS Outgoing Batch in Error";
    email_to = context.flowParameters['email_to'];
    String batchErrorSystem = "";

    if (!(inputMessage instanceof ControlMessage)) {
        ArrayList<EntityData> outputRecs = inputMessage.getPayload();
        for (EntityData outputRec : outputRecs) {
            Row row = context.getFlowStep().getComponent().toRow(outputRec, false, true);
            String system_name = row.getString("system_name");
            batchErrorSystem += "Batch in Error on:<t> " + system_name + "<br/>";
        }
    }

body = '''\
  <br/>
  SymmetricDS currently has an outgoing batch in Error!
  <br/>
  Identified by Metl Flow: $(_flowName)
  <br/>
  Host: $(_host)
  <br/>
  Agent: $(_agentName)
  <br/><br/>
  <u>System(s) with Error:</u>
  <br/> ''' +  batchErrorSystem + '''\
  <br/>
''';
    body = FormatUtils.replaceTokens(body, context.getFlowParameters(), true);

    info("Email to: " + email_to);
    info("Email subject: " + subject);
    info("Email body: " + body);

}

params = [email_subject:subject, email_body:body, email_to:email_to];
forwardMessageWithParameters(params);