Sending empty parameter hashmap in case of retried execution

app-conf/SchedulerConf.xml

@@ -29,7 +29,8 @@
         <name>azkaban</name>
         <classname>com.linkedin.drelephant.schedulers.AzkabanScheduler</classname>
         <params>
-          <!--<exception_enabled>true</exception_enabled>-->
+          <exception_enabled>true</exception_enabled>
+          <exception_log_length_limit>5120</exception_log_length_limit>
           <!--<workflow_client>com.linkedin.drelephant.exceptions.azkaban.AzkabanWorkflowClient</workflow_client>-->
           <!--<private_key>/home/key/private_key</private_key>-->
           <!--<userame>elephant</userame>-->

app/com/linkedin/drelephant/clients/WorkflowClient.java

@@ -16,13 +16,13 @@
 
 package com.linkedin.drelephant.clients;
 
+import com.linkedin.drelephant.exceptions.JobState;
+import com.linkedin.drelephant.exceptions.LoggingEvent;
+import com.linkedin.drelephant.exceptions.azkaban.JobLogException;
 import java.io.File;
 import java.util.Map;
 import java.util.Set;
 
-import com.linkedin.drelephant.exceptions.JobState;
-import com.linkedin.drelephant.exceptions.LoggingEvent;
-
 
 /**
  * The interface WorkflowClient should be implemented by all the workflow client. The client should not
@@ -46,17 +46,24 @@ public interface WorkflowClient {
   public void login(String username, File privateKey);
 
   /**
-   * Return all the jobs in the workflow. It returns a Map<String,String> where the key \n
+   * Return all the jobs in the workflow. It returns a Map<String,String> where the key
    * is the execution id of the job and the value is the status of the job.
    * @return Return all the jobs in the workflow
    */
   public Map<String, String> getJobsFromFlow();
 
+  /**
+   * Return all the jobs with there types in the workflow. It returns a Map<String,String> where the key
+   * is the job name and the value is the status of the job.
+   * @return Return all the jobs with their jobTypes in the workflow
+   */
+  public Map<String, String> getJobTypeFromFlow();
+
   /**
    * Given a job id, this method analyzes the job
    * @param jobId The execution id of the job
    */
-  public void analyzeJob(String jobId);
+  public void analyzeJob(String jobId) throws JobLogException;
 
   /**
    * This method extracts out all the yarn applications from the job and returns the set of them.

app/com/linkedin/drelephant/clients/azkaban/AzkabanWorkflowClient.java

@@ -20,7 +20,7 @@
 import com.linkedin.drelephant.exceptions.JobState;
 import com.linkedin.drelephant.exceptions.LoggingEvent;
 import com.linkedin.drelephant.exceptions.azkaban.AzkabanJobLogAnalyzer;
-
+import com.linkedin.drelephant.exceptions.azkaban.JobLogException;
 import java.io.BufferedReader;
 import java.io.File;
 import java.io.FileInputStream;
@@ -48,12 +48,10 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
-
 import javax.crypto.BadPaddingException;
 import javax.crypto.Cipher;
 import javax.crypto.IllegalBlockSizeException;
 import javax.crypto.NoSuchPaddingException;
-
 import org.apache.commons.io.IOUtils;
 import org.apache.http.HttpResponse;
 import org.apache.http.HttpStatus;
@@ -92,6 +90,7 @@ public class AzkabanWorkflowClient implements WorkflowClient {
   private String _username;
   private String _password;
   private long _sessionUpdatedTime = 0;
+  private JSONArray jobInfo = null;
 
   private String AZKABAN_LOG_OFFSET = "0";
   private String AZKABAN_LOG_LENGTH_LIMIT = "9999999"; // limit the log limit to 10 mb
@@ -384,24 +383,57 @@ private String decodeHeadlessChallenge(String encodedPassword, File _privateKey)
    * Returns the jobs from the flow
    * @return The jobs from the flow
    */
+  private JSONArray getJobsInfoFromFlow() throws JSONException {
+    if (jobInfo == null) {
+      List<NameValuePair> urlParameters = new ArrayList<NameValuePair>();
+      urlParameters.add(new BasicNameValuePair("session.id", _sessionId));
+      urlParameters.add(new BasicNameValuePair("ajax", "fetchexecflow"));
+      urlParameters.add(new BasicNameValuePair("execid", _executionId));
+      JSONObject jsonObject = fetchJson(urlParameters, _workflowExecutionUrl);
+      jobInfo = jsonObject.getJSONArray("nodes");
+      return jobInfo;
+    } else {
+      return jobInfo;
+    }
+  }
+
   public Map<String, String> getJobsFromFlow() {
-    List<NameValuePair> urlParameters = new ArrayList<NameValuePair>();
-    urlParameters.add(new BasicNameValuePair("session.id", _sessionId));
-    urlParameters.add(new BasicNameValuePair("ajax", "fetchexecflow"));
-    urlParameters.add(new BasicNameValuePair("execid", _executionId));
+    JSONArray jobsInfoNode;
+    try {
+      jobsInfoNode = (jobInfo == null) ? getJobsInfoFromFlow() : jobInfo;
+    } catch (JSONException jsonEx) {
+      logger.error("Exception while fetching Job execution info from Azkaban", jsonEx);
+      return null;
+    }
     try {
-      JSONObject jsonObject = fetchJson(urlParameters, _workflowExecutionUrl);
-      JSONArray jobs = jsonObject.getJSONArray("nodes");
       Map<String, String> jobMap = new HashMap<String, String>();
-      addJobStatusForFlow(jobMap, jobs);
+      addJobStatusForFlow(jobMap, jobsInfoNode);
       return jobMap;
-    } catch (JSONException e) {
-      logger.error("Error in parsing azkaban output ", e);
+    } catch (JSONException ex) {
+      logger.error("Error while extracting Job status information from flow execution details", ex);
     }
     return null;
   }
 
-  public void addJobStatusForFlow(Map<String, String> jobMap, JSONArray jobs) throws JSONException {
+  public Map<String, String> getJobTypeFromFlow() {
+    JSONArray jobsInfoNode;
+    try {
+      jobsInfoNode = (jobInfo == null) ? getJobsInfoFromFlow() : jobInfo;
+    } catch (JSONException jsonEx) {
+      logger.error("Exception while fetching Job execution info from Azkaban", jsonEx);
+      return null;
+    }
+    try {
+      Map<String, String> jobTypeMap = new HashMap<String, String>();
+      addJobTypeForFlow(jobTypeMap, jobsInfoNode);
+      return jobTypeMap;
+    } catch (JSONException ex) {
+      logger.error("Error while extracting Job types information from flow execution details");
+    }
+    return null;
+  }
+
+  private void addJobStatusForFlow(Map<String, String> jobMap, JSONArray jobs) throws JSONException {
     if (jobs != null) {
       for (int i = 0; i < jobs.length(); i++) {
         JSONObject job = jobs.getJSONObject(i);
@@ -414,6 +446,19 @@ public void addJobStatusForFlow(Map<String, String> jobMap, JSONArray jobs) thro
     }
   }
 
+  private void addJobTypeForFlow(Map<String, String> jobMap, JSONArray jobs) throws JSONException {
+    if (jobs != null) {
+      for (int i = 0; i < jobs.length(); i++) {
+        JSONObject job = jobs.getJSONObject(i);
+        jobMap.put(job.get("nestedId").toString(), job.get("type").toString());
+        if (!job.isNull("nodes")) {
+          JSONArray internalJobs = job.getJSONArray("nodes");
+          addJobStatusForFlow(jobMap, internalJobs);
+        }
+      }
+    }
+  }
+
   /**
    * Returns the azkaban flow log
    * @param offset The offset from which logs should be found
@@ -441,8 +486,11 @@ public String getAzkabanFlowLog(String offset, String maximumlLogLengthLimit) {
   }
 
   @Override
-  public void analyzeJob(String jobId) {
+  public void analyzeJob(String jobId) throws JobLogException {
     String rawAzkabanJobLog = getAzkabanJobLog(jobId, AZKABAN_LOG_OFFSET, AZKABAN_LOG_LENGTH_LIMIT);
+    if (rawAzkabanJobLog == null || rawAzkabanJobLog.isEmpty()) {
+      throw new JobLogException(String.format("Logs for job %s from Azkaban is empty", jobId));
+    }
     AzkabanJobLogAnalyzer analyzedLog = new AzkabanJobLogAnalyzer(rawAzkabanJobLog);
     jobIdToLog.put(jobId, analyzedLog);
   }
@@ -479,7 +527,7 @@ public LoggingEvent getJobException(String jobId) {
    * @param length Maximum limit on length of log
    * @return Azkaban job log in the form of string
    */
-  public String getAzkabanJobLog(String jobId, String offset, String length) {
+   private String getAzkabanJobLog(String jobId, String offset, String length) {
     List<NameValuePair> urlParameters = new ArrayList<NameValuePair>();
     urlParameters.add(new BasicNameValuePair("session.id", _sessionId));
     urlParameters.add(new BasicNameValuePair("ajax", "fetchExecJobLogs"));