Test email commit 2

Added Hashmap for success and failure email address for taskId mapping

Author: Balamurugan Kannadasan

api/src/main/java/com/homeaway/datapullclient/process/DataPullRequestProcessor.java

@@ -85,9 +85,9 @@ public class DataPullRequestProcessor implements DataPullClientService {
 
     private final ThreadPoolTaskScheduler scheduler;
 
-    private String userEmail;
-    private String failureEmail;
+    HashMap<String,String> successEmails = new HashMap<>();
 
+    HashMap<String,String> failureEmails = new HashMap<>();
     public DataPullRequestProcessor(){
         scheduler = new ThreadPoolTaskScheduler();
         scheduler.setPoolSize(POOL_SIZE);
@@ -127,6 +127,10 @@ public void runSimpleDataPull(String awsenv, String pipelinename) {
     }
 
     private void runDataPull(String json, boolean isStart, boolean validateJson) throws ProcessingException {
+
+        String userEmail;
+        String failureEmail;
+
         String originalInputJson = json;
         json = extractUserJsonFromS3IfProvided(json, isStart);
 
@@ -148,6 +152,9 @@ private void runDataPull(String json, boolean isStart, boolean validateJson) thr
             JsonNode jsonNode = objectMapper.readTree(json);
             userEmail = null != jsonNode.get("useremailaddress") ?  jsonNode.get("useremailaddress").asText(): "";
 
+            String taskId = jsonNode.get("cluster").get("awsenv").asText().concat("-emr-").concat(jsonNode.get("cluster").get("pipelinename").asText()).concat("-pipeline");
+            successEmails.put(taskId,userEmail);
+
             JsonNode failureEmailNode = jsonNode.get("failureemailaddress");
             if(StringUtils.isNotEmpty(userEmail)){
                 failureEmail = (failureEmailNode != null) ? userEmail.concat(",").concat(failureEmailNode.asText()): userEmail;
@@ -156,7 +163,7 @@ private void runDataPull(String json, boolean isStart, boolean validateJson) thr
                 failureEmail = (failureEmailNode != null) ? (failureEmailNode.asText()): "";
 
             }
-
+            failureEmails.put(taskId,failureEmail);
 
 
             List<Map.Entry<String, JsonNode>> result = new LinkedList<Map.Entry<String, JsonNode>>();
@@ -218,13 +225,13 @@ private void runDataPull(String json, boolean isStart, boolean validateJson) thr
             log.debug("runDataPull <- return");
     }
 
-    public String successMailAddress() throws ProcessingException {
-        return userEmail;
+    public HashMap<String, String> successMailAddress() throws ProcessingException {
+        return successEmails;
     }
 
 
-    public String failureMailAddress() throws ProcessingException {
-        return failureEmail;
+    public HashMap<String, String> failureMailAddress() throws ProcessingException {
+        return failureEmails;
     }
 
 

api/src/main/java/com/homeaway/datapullclient/process/DataPullTask.java

@@ -76,7 +76,6 @@ public class DataPullTask implements Runnable {
 
     String jobFlowId = null;
 
-    String emailAddress = null;
     String ClusterId = null;
 
     enum emailStatusCode {
@@ -453,14 +452,15 @@ private RunJobFlowResult runTaskInNewCluster(final AmazonElasticMapReduce emr, f
 
     private void sendEmailNotification() throws ProcessingException {
         String emailStatusCodeVal = null;
+        String emailAddress = null;
         try {
 
             if (jobFlowId != null) {
                 emailStatusCodeVal = emailStatusCode.CLUSTER_CREATION_SUCCESS.name();
-                emailAddress = dataPullRequestProcessor.successMailAddress();
+                emailAddress = dataPullRequestProcessor.successMailAddress().get(taskId);
             } else if (ClusterId != null) {
                 emailStatusCodeVal = emailStatusCode.SPARK_EXEC_ON_EXISTING_CLUSTER.name();
-                emailAddress = dataPullRequestProcessor.successMailAddress();
+                emailAddress = dataPullRequestProcessor.successMailAddress().get(taskId);
             } else {
                 if (jobFlowId == null) {
                     emailStatusCodeVal = emailStatusCode.CLUSTER_CREATION_FAILED.name();
@@ -471,7 +471,7 @@ private void sendEmailNotification() throws ProcessingException {
                 } else if (runTaskOnExistingClusterFail == true) {
                     emailStatusCodeVal = emailStatusCode.RUN_ON_EXISTING_CLUSTER_FAILED.name();
                 }
-                emailAddress = dataPullRequestProcessor.failureMailAddress();
+                emailAddress = dataPullRequestProcessor.failureMailAddress().get(taskId);
             }
 
             String[] emailAddressArray = null;