From bd0561163d21feedf6b62c58fb42f3f683ae7ece Mon Sep 17 00:00:00 2001
From: shubhamjainflipkart <shubham.j@flipkart.com>
Date: Mon, 29 Jan 2018 19:03:56 +0530
Subject: [PATCH 1/6] TEZ fixes

---
 app-conf/FetcherConf.xml                      |   8 +-
 app-conf/GeneralConf.xml                      |   2 +-
 .../drelephant/analysis/AnalyticJob.java      |  36 +++++
 .../analysis/AnalyticJobGeneratorHadoop2.java |  35 +++--
 .../drelephant/analysis/ElephantFetcher.java  |   3 +
 .../fetchers/MapReduceFSFetcherHadoop2.java   |   5 +
 .../fetchers/MapReduceFetcherHadoop2.java     |  17 ++-
 .../drelephant/spark/fetchers/FSFetcher.scala |   2 +
 .../spark/fetchers/SparkFetcher.scala         |   2 +
 .../tez/fetchers/TezDataFetcherHadoop2.java   | 111 +++++++---------
 .../drelephant/util/InfoExtractor.java        |  13 ++
 app/models/AppResult.java                     |   2 +-
 app/models/FailedAppResult.java               | 124 ++++++++++++++++++
 .../spark/deploy/history/SparkFSFetcher.scala |   2 +
 conf/evolutions/default/5.sql                 |  25 ++++
 15 files changed, 300 insertions(+), 87 deletions(-)
 create mode 100644 app/models/FailedAppResult.java

diff --git a/app-conf/FetcherConf.xml b/app-conf/FetcherConf.xml
index 6af995169..eae082c53 100644
--- a/app-conf/FetcherConf.xml
+++ b/app-conf/FetcherConf.xml
@@ -29,7 +29,7 @@
   </fetcher>
 -->
 <fetchers>
-  <!--
+
   <fetcher>
     <applicationtype>mapreduce</applicationtype>
     <classname>com.linkedin.drelephant.mapreduce.fetchers.MapReduceFetcherHadoop2</classname>
@@ -37,7 +37,7 @@
       <sampling_enabled>false</sampling_enabled>
     </params>
   </fetcher>
-  -->
+
   <!--
      This is a replacement for the MapReduceFetcherHadoop2 that attempts to burn
      through queues of jobs faster by pulling data directly from HDFS rather than going through
@@ -51,7 +51,7 @@
      To work properly, this fetcher should use the same timezone with the job history server.
      If not set, the local timezone will be used.
    -->
-
+<!--
   <fetcher>
     <applicationtype>mapreduce</applicationtype>
     <classname>com.linkedin.drelephant.mapreduce.fetchers.MapReduceFSFetcherHadoop2</classname>
@@ -62,7 +62,7 @@
     </params>
   </fetcher>
 
-
+-->
   <!--
     FSFetcher for Spark. Loads the eventlog from HDFS and replays to get the metrics and application properties
 
diff --git a/app-conf/GeneralConf.xml b/app-conf/GeneralConf.xml
index 61f3c3928..8dcbf465b 100644
--- a/app-conf/GeneralConf.xml
+++ b/app-conf/GeneralConf.xml
@@ -39,7 +39,7 @@
   </property>
   <property>
     <name>drelephant.executor.service.class.name</name>
-    <value>com.linkedin.drelephant.executors.QuartzExecutorService</value>
+    <value>com.linkedin.drelephant.executors.ThreadPoolExecutorService</value>
     <description>Executor service (can be one of ThreadPoolExecutorService or QuartzExecutorService)</description>
   </property>
   <!--
diff --git a/app/com/linkedin/drelephant/analysis/AnalyticJob.java b/app/com/linkedin/drelephant/analysis/AnalyticJob.java
index 3a07a74e3..c9c5951a3 100644
--- a/app/com/linkedin/drelephant/analysis/AnalyticJob.java
+++ b/app/com/linkedin/drelephant/analysis/AnalyticJob.java
@@ -27,6 +27,8 @@
 import models.AppHeuristicResult;
 import models.AppHeuristicResultDetails;
 import models.AppResult;
+import models.FailedAppResult;
+import org.apache.commons.lang.exception.ExceptionUtils;
 import org.apache.log4j.Logger;
 
 
@@ -249,6 +251,40 @@ public AnalyticJob setTrackingUrl(String trackingUrl) {
     return this;
   }
 
+
+  public FailedAppResult getFailedAppResult(Exception e) throws Exception {
+
+    ElephantFetcher fetcher = ElephantContext.instance().getFetcherForApplicationType(getAppType());
+    HadoopApplicationData data = fetcher.fetchConfData(this);
+
+    JobType jobType = ElephantContext.instance().matchJobType(data);
+    String jobTypeName = jobType == null ? UNKNOWN_JOB_TYPE : jobType.getName();
+
+    AppResult result = new AppResult();
+    InfoExtractor.loadInfo(result, data);
+
+    FailedAppResult failedApp = new FailedAppResult();
+    failedApp.appId = Utils.truncateField(getAppId(), AppResult.ID_LIMIT, getAppId());
+    failedApp.startTime = getStartTime();
+    failedApp.finishTime = getFinishTime();
+    failedApp.name = Utils.truncateField(getName(), AppResult.APP_NAME_LIMIT, getAppId());
+    failedApp.trackingUrl = Utils.truncateField(getTrackingUrl(), AppResult.TRACKING_URL_LIMIT, getAppId());
+    failedApp.jobType = Utils.truncateField(jobTypeName, AppResult.JOBTYPE_LIMIT, getAppId());
+    failedApp.scheduler = result.scheduler;
+    failedApp.jobName =result.jobName;
+    failedApp.jobDefId = result.jobDefId;
+    failedApp.jobExecId = result.jobExecId;
+    failedApp.flowDefId = result.flowDefId;
+    failedApp.flowExecId = result.flowExecId;
+    failedApp.jobDefUrl = result.jobDefUrl;
+    failedApp.jobExecUrl = result.jobExecUrl;
+    failedApp.flowDefId =result.flowDefId;
+    failedApp.flowExecUrl = result.jobExecUrl;
+    failedApp.error = ExceptionUtils.getStackTrace(e);
+
+    return failedApp;
+  }
+
   /**
    * Returns the analysed AppResult that could be directly serialized into DB.
    *
diff --git a/app/com/linkedin/drelephant/analysis/AnalyticJobGeneratorHadoop2.java b/app/com/linkedin/drelephant/analysis/AnalyticJobGeneratorHadoop2.java
index 4f6d18d19..bdf4cc3c1 100644
--- a/app/com/linkedin/drelephant/analysis/AnalyticJobGeneratorHadoop2.java
+++ b/app/com/linkedin/drelephant/analysis/AnalyticJobGeneratorHadoop2.java
@@ -28,6 +28,7 @@
 import java.util.*;
 import models.AppResult;
 import models.CheckPoint;
+import models.FailedAppResult;
 import org.apache.commons.lang.exception.ExceptionUtils;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.security.authentication.client.AuthenticatedURL;
@@ -230,24 +231,32 @@ public void analyseJob(AnalyticJob analyticJob) {
       logger.error(e.getMessage());
       logger.error(ExceptionUtils.getStackTrace(e));
 
-      if (analyticJob != null && analyticJob.isPrimaryPhaseRetry()) {
-        logger.error("Add analytic job id [" + analyticJob.getAppId() + "] into the retry list.");
-        MetricsController.triggerJobFailedEvent(analyticJob.getRetriesCount());
-        ElephantRunner.getInstance().getExecutorService().onPrimaryRetry(analyticJob);
-
-      } else if (analyticJob != null && analyticJob.isSecondPhaseRetry()) {
-        //Putting the job into a second retry queue which fetches jobs after some interval. Some spark jobs may need more time than usual to process, hence the queue.
-        logger.error("Add analytic job id [" + analyticJob.getAppId() + "] into the second retry list.");
-        MetricsController.triggerJobFailedEvent(analyticJob.getRetriesCount());
-        ElephantRunner.getInstance().getExecutorService().onSecondaryRetry(analyticJob);
-
-      } else {
+//      if (analyticJob != null && analyticJob.isPrimaryPhaseRetry()) {
+//        logger.error("Add analytic job id [" + analyticJob.getAppId() + "] into the retry list.");
+//        MetricsController.triggerJobFailedEvent(analyticJob.getRetriesCount());
+//        ElephantRunner.getInstance().getExecutorService().onPrimaryRetry(analyticJob);
+//
+//      } else if (analyticJob != null && analyticJob.isSecondPhaseRetry()) {
+//        //Putting the job into a second retry queue which fetches jobs after some interval. Some spark jobs may need more time than usual to process, hence the queue.
+//        logger.error("Add analytic job id [" + analyticJob.getAppId() + "] into the second retry list.");
+//        MetricsController.triggerJobFailedEvent(analyticJob.getRetriesCount());
+//        ElephantRunner.getInstance().getExecutorService().onSecondaryRetry(analyticJob);
+//
+//      } else {
         if (analyticJob != null) {
           MetricsController.triggerJobRetriesExhaustionEvent();
+          try {
+            FailedAppResult result = analyticJob.getFailedAppResult(e);
+            result.save();
+
+          } catch (Exception ex) {
+            logger.error("Failed to get info for failed app: " + analyticJob.getAppId());
+            logger.error(ExceptionUtils.getStackTrace(ex));
+          }
           logger.error("Drop the analytic job. Reason: reached the max retries for application id = ["
                   + analyticJob.getAppId() + "].");
         }
-      }
+     // }
     }
   }
 
diff --git a/app/com/linkedin/drelephant/analysis/ElephantFetcher.java b/app/com/linkedin/drelephant/analysis/ElephantFetcher.java
index d83ff90ad..a20d1b29c 100644
--- a/app/com/linkedin/drelephant/analysis/ElephantFetcher.java
+++ b/app/com/linkedin/drelephant/analysis/ElephantFetcher.java
@@ -32,4 +32,7 @@ public interface ElephantFetcher<T extends HadoopApplicationData> {
    */
   public T fetchData(AnalyticJob job)
       throws Exception;
+
+  public T fetchConfData(AnalyticJob job);
+
 }
\ No newline at end of file
diff --git a/app/com/linkedin/drelephant/mapreduce/fetchers/MapReduceFSFetcherHadoop2.java b/app/com/linkedin/drelephant/mapreduce/fetchers/MapReduceFSFetcherHadoop2.java
index ec025927d..91fb49830 100644
--- a/app/com/linkedin/drelephant/mapreduce/fetchers/MapReduceFSFetcherHadoop2.java
+++ b/app/com/linkedin/drelephant/mapreduce/fetchers/MapReduceFSFetcherHadoop2.java
@@ -202,6 +202,11 @@ private DataFiles getHistoryFiles(AnalyticJob job) throws IOException {
     return new DataFiles(jobConfPath, jobHistPath);
   }
 
+  @Override
+  public MapReduceApplicationData fetchConfData(AnalyticJob analyticJob) {
+    return new MapReduceApplicationData();
+  }
+
   @Override
   public MapReduceApplicationData fetchData(AnalyticJob job) throws IOException {
     DataFiles files = getHistoryFiles(job);
diff --git a/app/com/linkedin/drelephant/mapreduce/fetchers/MapReduceFetcherHadoop2.java b/app/com/linkedin/drelephant/mapreduce/fetchers/MapReduceFetcherHadoop2.java
index 4165971aa..c62928460 100644
--- a/app/com/linkedin/drelephant/mapreduce/fetchers/MapReduceFetcherHadoop2.java
+++ b/app/com/linkedin/drelephant/mapreduce/fetchers/MapReduceFetcherHadoop2.java
@@ -71,19 +71,32 @@ public MapReduceFetcherHadoop2(FetcherConfigurationData fetcherConfData) throws
   }
 
   @Override
-  public MapReduceApplicationData fetchData(AnalyticJob analyticJob) throws IOException, AuthenticationException {
+  public MapReduceApplicationData fetchConfData(AnalyticJob analyticJob) {
     String appId = analyticJob.getAppId();
     MapReduceApplicationData jobData = new MapReduceApplicationData();
     String jobId = Utils.getJobIdFromApplicationId(appId);
     jobData.setAppId(appId).setJobId(jobId);
     // Change job tracking url to job history page
     analyticJob.setTrackingUrl(_jhistoryWebAddr + jobId);
-    try {
 
       // Fetch job config
+    try {
       Properties jobConf = _jsonFactory.getProperties(_urlFactory.getJobConfigURL(jobId));
       jobData.setJobConf(jobConf);
+    } catch (Exception e) {
+      logger.error("Failed to fetch conf data: ", e);
+    }
+      return jobData;
+  }
+
+  @Override
+  public MapReduceApplicationData fetchData(AnalyticJob analyticJob) throws IOException, AuthenticationException {
+
+    String appId = analyticJob.getAppId();
+    String jobId = Utils.getJobIdFromApplicationId(appId);
+    MapReduceApplicationData jobData = fetchConfData(analyticJob);
 
+    try {
       URL jobURL = _urlFactory.getJobURL(jobId);
       String state = _jsonFactory.getState(jobURL);
 
diff --git a/app/com/linkedin/drelephant/spark/fetchers/FSFetcher.scala b/app/com/linkedin/drelephant/spark/fetchers/FSFetcher.scala
index e85196c2c..f6e95d182 100644
--- a/app/com/linkedin/drelephant/spark/fetchers/FSFetcher.scala
+++ b/app/com/linkedin/drelephant/spark/fetchers/FSFetcher.scala
@@ -30,6 +30,8 @@ class FSFetcher(fetcherConfigurationData: FetcherConfigurationData)
   extends ElephantFetcher[SparkApplicationData] {
   lazy val legacyFetcher = new SparkFSFetcher(fetcherConfigurationData)
 
+  override def fetchConfData(job: AnalyticJob): SparkApplicationData = ???
+
   override def fetchData(analyticJob: AnalyticJob): SparkApplicationData = {
     val legacyData = legacyFetcher.fetchData(analyticJob)
     LegacyDataConverters.convert(legacyData)
diff --git a/app/com/linkedin/drelephant/spark/fetchers/SparkFetcher.scala b/app/com/linkedin/drelephant/spark/fetchers/SparkFetcher.scala
index 4f2020237..2568e254c 100644
--- a/app/com/linkedin/drelephant/spark/fetchers/SparkFetcher.scala
+++ b/app/com/linkedin/drelephant/spark/fetchers/SparkFetcher.scala
@@ -76,6 +76,8 @@ class SparkFetcher(fetcherConfigurationData: FetcherConfigurationData)
     new SparkLogClient(hadoopConfiguration, sparkConf, eventLogUri)
   }
 
+  override def fetchConfData(job: AnalyticJob): SparkApplicationData = ???
+
   override def fetchData(analyticJob: AnalyticJob): SparkApplicationData = {
     doFetchData(analyticJob) match {
       case Success(data) => data
diff --git a/app/com/linkedin/drelephant/tez/fetchers/TezDataFetcherHadoop2.java b/app/com/linkedin/drelephant/tez/fetchers/TezDataFetcherHadoop2.java
index 9ff3ca8fa..f2be27ce3 100644
--- a/app/com/linkedin/drelephant/tez/fetchers/TezDataFetcherHadoop2.java
+++ b/app/com/linkedin/drelephant/tez/fetchers/TezDataFetcherHadoop2.java
@@ -62,7 +62,7 @@ public class TezDataFetcherHadoop2 extends TezDataFetcher {
   private String _timelinewebaddress;
   private String _jhistoryWebAddr;
   private String _resourcemanager;
-  private String _resourcemanagerWebAddress;
+  private String _applicationHistoryAddress;
   private String _dagId;
   private static final String SUCCEEDED="SUCCEEDED";
   /**
@@ -90,33 +90,41 @@ public TezDataFetcherHadoop2(FetcherConfigurationData fetcherConfData) throws IO
     _jhistoryWebAddr = "http://" + jhistoryAddr + "/jobhistory/job/";
 
     _resourcemanager = String.format("http://%s/cluster/app/", resourcemanager);
-    _resourcemanagerWebAddress = String.format("http://%s/ws/v1/cluster/apps/", resourcemanager);
-
+    _applicationHistoryAddress = String.format("http://%s/ws/v1/applicationhistory/apps/", timelineaddress);
   }
 
-  /**
-   * The fetcher gets the number of dags submitted per application using the diagnostic info from Resource Manager
-   * Then each dag is analyzed and vertex/task information is parsed using the data from timeline server
-   * The analysis is done at the application level and not DAG level.
-   */
   @Override
-  public TezDAGApplicationData fetchData(AnalyticJob analyticJob) throws IOException, AuthenticationException {
+  public TezDAGApplicationData fetchConfData(AnalyticJob analyticJob) {
     String appId = analyticJob.getAppId();
     TezDAGApplicationData jobData = new TezDAGApplicationData();
     String jobId = Utils.getJobIdFromApplicationId(appId);
     jobData.setAppId(appId).setJobId(jobId);
     // Change job tracking url to job history page
     analyticJob.setTrackingUrl(_resourcemanager + appId);
-    try {
 
-      // Fetch job config
+    // Fetch job config
+    try {
       Properties jobConf = _jsonFactory.getProperties(_urlFactory.getJobConfigURL(appId));
       jobData.setJobConf(jobConf);
+    } catch (Exception e) {
+      logger.error("Failed to fetch conf data: ", e);
+    }
+    return jobData;
+  }
 
-      //Fetch number of dags submitted
-      int dagCount = getDagSubmittedCount(_urlFactory.getTezDagSubmittedCountURL(appId),jobData);
-
+  /**
+   * The fetcher gets the number of dags submitted per application using the diagnostic info from Resource Manager
+   * Then each dag is analyzed and vertex/task information is parsed using the data from timeline server
+   * The analysis is done at the application level and not DAG level.
+   */
+  @Override
+  public TezDAGApplicationData fetchData(AnalyticJob analyticJob) throws IOException, AuthenticationException {
 
+    String appId = analyticJob.getAppId();
+    String jobId = Utils.getJobIdFromApplicationId(appId);
+    TezDAGApplicationData jobData = fetchConfData(analyticJob);
+    try {
+      int dagCount = getDagSubmittedCount(_urlFactory.getTezDagSubmittedCountURL(appId),jobData);
 
       //Right now it only supports DAGs which succeeded. It looks at DAG details to understand wether the application failed or succeded later in the code.
       String state = "SUCCEEDED";
@@ -178,10 +186,9 @@ private int getDagSubmittedCount(URL url,TezDAGApplicationData jobData) throws I
     List<AnalyticJob> appList = new ArrayList<AnalyticJob>();
 
     JsonNode rootNode = ThreadContextTez.readJsonNode(url);
-
-    String diagnosticsInfo = rootNode.path("app").path("diagnostics").getValueAsText();
-    Long startedTime = (rootNode.path("app").path("startedTime").getLongValue());
-    Long finishedTime = (rootNode.path("app").path("finishedTime").getLongValue());
+    String diagnosticsInfo = rootNode.path("diagnosticsInfo").getValueAsText();
+    Long startedTime = (rootNode.path("startedTime").getLongValue());
+    Long finishedTime = (rootNode.path("finishedTime").getLongValue());
     int dagCount = 0;
     if(diagnosticsInfo!=null){
       String dagTypes[] = diagnosticsInfo.replace("Session stats:", "").split(",");
@@ -191,8 +198,9 @@ private int getDagSubmittedCount(URL url,TezDAGApplicationData jobData) throws I
 
       for(String dagType:dagTypes){
 
+        if(dagType.contains("successfulDAGs") && Integer.parseInt(dagType.substring((dagType.indexOf("=")+1)).trim())==0)
+          throw new RuntimeException("No successfull DAG found");
         dagCount += Integer.parseInt(dagType.substring((dagType.indexOf("=")+1)).trim());
-
       }
     }
     return dagCount;
@@ -260,10 +268,9 @@ private URL getTaskAttemptURL(String jobId, String taskId, String attemptId) thr
       return new URL(_restRoot + "/" + jobId + "/tasks/" + taskId + "/attempts/" + attemptId);
     }
     private URL getTezDAGURL(String appId,int id) throws MalformedURLException {
+
       String dagId = appId.replace("application","dag");
-      //	System.out.println(dagId+"dagId");
-      dagId =_tezRoot+"TEZ_DAG_ID/"+dagId+"_"+id;
-      //	System.out.println(dagId+"dagId");
+      dagId =_tezRoot+"TEZ_VERTEX_ID?limit=9007199254740991&primaryFilter=TEZ_DAG_ID:"+dagId+"_"+id+"&secondaryFilter=status:SUCCEEDED";
       return new URL(dagId);
     }
     private String getTezDAGId(String appId,int id) throws MalformedURLException {
@@ -274,7 +281,7 @@ private String getTezDAGId(String appId,int id) throws MalformedURLException {
       return (dagId);
     }
     private URL getTezDagSubmittedCountURL(String appId) throws MalformedURLException{
-      return new URL(_resourcemanagerWebAddress+appId);
+      return new URL(_applicationHistoryAddress+appId);
     }
     private URL getTezDAGURL(String dagId) throws MalformedURLException {
       return new URL(_tezRoot+"TEZ_DAG_ID/"+dagId);
@@ -408,27 +415,21 @@ private void getTaskDataAll(List<TezVertexData> tezVertexList, List<TezVertexTas
         return;
       }
 
-      JsonNode vertexNode = rootNode.path("otherinfo").path("vertexNameIdMapping");
-      Iterator<JsonNode> tasks = vertexNode.iterator();
-      Iterator<String> vertexNames = vertexNode.getFieldNames();
+      Iterator<JsonNode> vertexNode = rootNode.path("entities").getElements();
 
-      while (tasks.hasNext() && vertexNames.hasNext()) {
+      while (vertexNode.hasNext()) {
 
-        JsonNode vertex = tasks.next();
-        String vertexId = vertex.getValueAsText();
+        JsonNode vertex = vertexNode.next();
+        String vertexId = vertex.get("entity").getValueAsText();
         TezVertexData tezVertexData = new TezVertexData(vertexId);
-        JsonNode vertexRootNode = ThreadContextTez.readJsonNode(_urlFactory.getTezVertexURL(vertexId));
+        JsonNode vertexNameNode = vertex.path("otherinfo").get("vertexName");
+        String vertexName = (vertexNameNode==null)?"Reducer":vertexNameNode.getValueAsText();
 
-        if(!(vertexRootNode.path("primaryfilters").get("status").getElementValue(0).getTextValue()).contains((SUCCEEDED))){
-          break;
-        }
-
-        String vertexName = vertexNames.next();
         tezVertexData.setVertexName(vertexName);
         long startTime = 0l;
         long finishTime = 0l;
         long initialTime = 0l;
-        for(JsonNode event:vertexRootNode.path("events") ){
+        for(JsonNode event:vertex.path("events") ){
           //	System.out.println("vertexevents"+event);
           if("VERTEX_STARTED".equals(event.get("eventtype").getValueAsText())){
 
@@ -448,7 +449,7 @@ private void getTaskDataAll(List<TezVertexData> tezVertexList, List<TezVertexTas
         }
         long time [] =  { finishTime - startTime, 0, 0, startTime, finishTime };
         tezVertexData.setTime(time);
-        JsonNode groupsTez = vertexRootNode.path("otherinfo").path("counters").path("counterGroups");
+        JsonNode groupsTez = vertex.path("otherinfo").path("counters").path("counterGroups");
         TezCounterData holder = new TezCounterData();
 
         for (JsonNode group : groupsTez) {
@@ -457,36 +458,17 @@ private void getTaskDataAll(List<TezVertexData> tezVertexList, List<TezVertexTas
             Long counterValue = counter.get("counterValue").getLongValue();
             String groupName = group.get("counterGroupName").getValueAsText();
             holder.set(groupName, counterName, counterValue);
-            //  System.out.println("counter"+counterName+counterValue+groupName);
-
           }
         }
         tezVertexData.setCounter(holder);
-        JsonNode tasksNode = vertexRootNode.path("relatedentities").get("TEZ_TASK_ID");
-        if(tasksNode == null) {
-          logger.warn("Task ids are not available for vertex: " + vertexId);
-        }
-
-        //   System.out.println("taskNode"+tasksNode);
-        Iterator<JsonNode> iterator = (tasksNode == null)?null:tasksNode.getElements();
-        //   System.out.println("iterator"+iterator);
-        while(iterator != null && iterator.hasNext()){
-          JsonNode taskNode = iterator.next();
-          String taskId=taskNode.getValueAsText();
-          //     System.out.println("taskId is "+taskId);
 
-          JsonNode tezTaskRootNode = ThreadContextTez.readJsonNode(_urlFactory.getTezTaskIdURL(taskId));
-          //    System.out.println("task root node  "+tezTaskRootNode);
-
-          if(!(tezTaskRootNode.path("primaryfilters").get("status").getElementValue(0).getTextValue()).contains(SUCCEEDED)){
-            break;
-          }
-
-          //    System.out.println("task succeeded");
+        Iterator<JsonNode> taskRootNode = ThreadContextTez.readJsonNode(new URL("http://prod-fdphadoop-bheema-hs-0001:8188/ws/v1/timeline/TEZ_TASK_ID?limit=9007199254740991&primaryFilter=TEZ_VERTEX_ID:"+vertexId+"&secondaryFilter=status:SUCCEEDED")).path("entities").getElements();
+        while(taskRootNode.hasNext()){
+          JsonNode taskNode = taskRootNode.next();
+          String taskId=taskNode.get("entity").getValueAsText();
 
           TezVertexTaskData mapReduceTaskData = new TezVertexTaskData(taskId,
-                  tezTaskRootNode.path("relatedentities").get("TEZ_TASK_ATTEMPT_ID").getValueAsText());
-          //    System.out.println("mapreduce data"+mapReduceTaskData);
+                  taskNode.path("otherinfo").get("successfulAttemptId").getValueAsText());
 
           if(vertexName.contains("Map")){
             mapperList.add(mapReduceTaskData);
@@ -496,7 +478,7 @@ private void getTaskDataAll(List<TezVertexData> tezVertexList, List<TezVertexTas
           }else{
             reducerList.add(mapReduceTaskData);
           }
-          for(JsonNode event:tezTaskRootNode.path("events") ){
+          for(JsonNode event:taskNode.path("events") ){
             if("TASK_STARTED".equals(event.get("eventtype").getValueAsText())){
               startTime=(event.get("timestamp").getValueAsLong());
             }
@@ -507,7 +489,7 @@ private void getTaskDataAll(List<TezVertexData> tezVertexList, List<TezVertexTas
           }
           long taskTime [] =  { finishTime - startTime, 0, 0, startTime, finishTime };
           mapReduceTaskData.setTime(taskTime);
-          JsonNode groupsTezTask = tezTaskRootNode.path("otherinfo").path("counters").path("counterGroups");
+          JsonNode groupsTezTask = taskNode.path("otherinfo").path("counters").path("counterGroups");
           TezCounterData holderTask = new TezCounterData();
           for (JsonNode group : groupsTezTask) {
             for (JsonNode counter : group.path("counters")) {
@@ -515,9 +497,6 @@ private void getTaskDataAll(List<TezVertexData> tezVertexList, List<TezVertexTas
               Long counterValue = counter.get("counterValue").getLongValue();
               String groupName = group.get("counterGroupName").getValueAsText();
               holderTask.set(groupName, counterName, counterValue);
-              //     System.out.println("counter task"+counterName+counterValue);
-
-
             }
           }
           mapReduceTaskData.setCounter(holderTask);
diff --git a/app/com/linkedin/drelephant/util/InfoExtractor.java b/app/com/linkedin/drelephant/util/InfoExtractor.java
index a81de19d9..c22789f81 100644
--- a/app/com/linkedin/drelephant/util/InfoExtractor.java
+++ b/app/com/linkedin/drelephant/util/InfoExtractor.java
@@ -31,6 +31,7 @@
 import java.util.Set;
 import java.util.Map;
 
+import com.linkedin.drelephant.tez.data.TezDAGApplicationData;
 import org.apache.commons.lang.StringUtils;
 import org.apache.log4j.Logger;
 import org.w3c.dom.Document;
@@ -115,7 +116,10 @@ public static void loadInfo(AppResult result, HadoopApplicationData data) {
       properties = retrieveMapreduceProperties((MapReduceApplicationData) data);
     } else if ( data instanceof SparkApplicationData) {
       properties = retrieveSparkProperties((SparkApplicationData) data);
+    } else if ( data instanceof TezDAGApplicationData) {
+      properties = retrieveTezProperties((TezDAGApplicationData) data);
     }
+
     Scheduler scheduler = getSchedulerInstance(data.getAppId(), properties);
 
     if (scheduler == null) {
@@ -166,6 +170,15 @@ public static Properties retrieveMapreduceProperties(MapReduceApplicationData ap
     return appData.getConf();
   }
 
+  /**
+   * Retrieve the tez application properties
+   * @param appData the Tez Application Data
+   * @return the retrieve tez properties
+   */
+  public static Properties retrieveTezProperties(TezDAGApplicationData appData) {
+    return appData.getConf();
+  }
+
   /**
    * Populates the given app result with the info from the given application data and scheduler.
    *
diff --git a/app/models/AppResult.java b/app/models/AppResult.java
index e7c9c7f6c..b1864336b 100644
--- a/app/models/AppResult.java
+++ b/app/models/AppResult.java
@@ -57,7 +57,7 @@ public static class TABLE {
     public static final String ID = "id";
     public static final String NAME = "name";
     public static final String USERNAME = "username";
-      public static final String QUEUE_NAME = "queueName";
+    public static final String QUEUE_NAME = "queueName";
     public static final String START_TIME = "startTime";
     public static final String FINISH_TIME = "finishTime";
     public static final String TRACKING_URL = "trackingUrl";
diff --git a/app/models/FailedAppResult.java b/app/models/FailedAppResult.java
new file mode 100644
index 000000000..cdfbbbba2
--- /dev/null
+++ b/app/models/FailedAppResult.java
@@ -0,0 +1,124 @@
+/*
+ * Copyright 2016 LinkedIn Corp.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License"); you may not
+ * use this file except in compliance with the License. You may obtain a copy of
+ * the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+ * License for the specific language governing permissions and limitations under
+ * the License.
+ */
+
+package models;
+
+        import org.apache.commons.httpclient.util.ExceptionUtil;
+        import play.db.ebean.Model;
+        import javax.persistence.Column;
+        import javax.persistence.Entity;
+        import javax.persistence.Id;
+        import javax.persistence.Table;
+
+
+@Entity
+@Table(name = "failed_app_result")
+public class FailedAppResult extends Model {
+
+    private static final long serialVersionUID = 1L;
+
+    public static final int ID_LIMIT = 50;
+    public static final int USERNAME_LIMIT = 50;
+    public static final int QUEUE_NAME_LIMIT = 50;
+    public static final int APP_NAME_LIMIT = 100;
+    public static final int JOB_NAME_LIMIT = 255;
+    public static final int TRACKING_URL_LIMIT = 255;
+    public static final int JOBTYPE_LIMIT = 20;
+    public static final int SCHEDULER_LIMIT = 20;
+    public static final int URL_LEN_LIMIT = 800;
+    public static final int FLOW_EXEC_ID_LIMIT = 255;
+
+    // Note that the Table column constants are actually the java variable names defined in this model.
+    // This is because ebean operations require the model variable names to be passed as strings.
+    public static class TABLE {
+        public static final String TABLE_NAME = "failed_app_result";
+        public static final String ID = "id";
+        public static final String APP_ID = "appId";
+        public static final String NAME = "name";
+        public static final String START_TIME = "startTime";
+        public static final String FINISH_TIME = "finishTime";
+        public static final String TRACKING_URL = "trackingUrl";
+        public static final String JOB_TYPE = "jobType";
+        public static final String SCHEDULER = "scheduler";
+        public static final String JOB_NAME = "jobName";
+        public static final String JOB_EXEC_ID = "jobExecId";
+        public static final String FLOW_EXEC_ID = "flowExecId";
+        public static final String JOB_DEF_ID = "jobDefId";
+        public static final String FLOW_DEF_ID = "flowDefId";
+        public static final String JOB_EXEC_URL = "jobExecUrl";
+        public static final String FLOW_EXEC_URL = "flowExecUrl";
+        public static final String JOB_DEF_URL = "jobDefUrl";
+        public static final String FLOW_DEF_URL = "flowDefUrl";
+        public static final String ERROR = "error";
+    }
+
+    @Id
+    @Column(unique = true, nullable = false)
+    public String id;
+
+    @Column(length = ID_LIMIT, unique = true, nullable = false)
+    public String appId;
+
+    @Column(length = APP_NAME_LIMIT, nullable = false)
+    public String name;
+
+    @Column(nullable = false)
+    public long startTime;
+
+    @Column(nullable = false)
+    public long finishTime;
+
+    @Column(length = TRACKING_URL_LIMIT, nullable = false)
+    public String trackingUrl;
+
+    @Column(length = JOBTYPE_LIMIT, nullable = false)
+    public String jobType;
+
+    @Column(length = SCHEDULER_LIMIT, nullable = true)
+    public String scheduler;
+
+    @Column(length = JOB_NAME_LIMIT, nullable = false)
+    public String jobName;
+
+    @Column(length = URL_LEN_LIMIT, nullable = false)
+    public String jobExecId;
+
+    @Column(length = FLOW_EXEC_ID_LIMIT, nullable = false)
+    public String flowExecId;
+
+    @Column(length = URL_LEN_LIMIT, nullable = false)
+    public String jobDefId;
+
+    @Column(length = URL_LEN_LIMIT, nullable = false)
+    public String flowDefId;
+
+    @Column(length = URL_LEN_LIMIT, nullable = false)
+    public String jobExecUrl;
+
+    @Column(length = URL_LEN_LIMIT, nullable = false)
+    public String flowExecUrl;
+
+    @Column(length = URL_LEN_LIMIT, nullable = false)
+    public String jobDefUrl;
+
+    @Column(length = URL_LEN_LIMIT, nullable = false)
+    public String flowDefUrl;
+
+    @Column(nullable = false)
+    public String error;
+
+    public static Finder<String, AppResult> find = new Finder<String, AppResult>(String.class, AppResult.class);
+}
diff --git a/app/org/apache/spark/deploy/history/SparkFSFetcher.scala b/app/org/apache/spark/deploy/history/SparkFSFetcher.scala
index ecac7f608..6f75e2089 100644
--- a/app/org/apache/spark/deploy/history/SparkFSFetcher.scala
+++ b/app/org/apache/spark/deploy/history/SparkFSFetcher.scala
@@ -69,6 +69,8 @@ class SparkFSFetcher(fetcherConfData: FetcherConfigurationData) extends Elephant
     sparkConf
   }
 
+  override def fetchConfData(job: AnalyticJob): SparkApplicationData = ???
+
   def fetchData(analyticJob: AnalyticJob): SparkApplicationData = {
     val appId = analyticJob.getAppId()
     doAsPrivilegedAction { () => doFetchData(appId) }
diff --git a/conf/evolutions/default/5.sql b/conf/evolutions/default/5.sql
index c47dbb3a2..54a357abb 100644
--- a/conf/evolutions/default/5.sql
+++ b/conf/evolutions/default/5.sql
@@ -16,6 +16,29 @@
 
 # --- !Ups
 
+CREATE TABLE failed_app_result (
+  id               INT(11)       NOT NULL AUTO_INCREMENT,
+  app_id           VARCHAR(50)   NOT NULL                   COMMENT 'The application id, e.g., application_1236543456321_1234567',
+  name             VARCHAR(100)  NOT NULL                   COMMENT 'The application name',
+  start_time       BIGINT        UNSIGNED NOT NULL          COMMENT 'The time in which application started',
+  finish_time      BIGINT        UNSIGNED NOT NULL          COMMENT 'The time in which application finished',
+  tracking_url     VARCHAR(255)  NOT NULL                   COMMENT 'The web URL that can be used to track the application',
+  job_type         VARCHAR(20)   NOT NULL                   COMMENT 'The Job Type e.g, Pig, Hive, Spark, HadoopJava',
+  scheduler        VARCHAR(20)                              COMMENT 'The scheduler which triggered the application',
+  job_name         VARCHAR(255)                             COMMENT 'The name of the job in the flow to which this app belongs',
+  job_exec_id      VARCHAR(800)                             COMMENT 'A unique reference to a specific execution of the job/action(job in the workflow). This should filter all applications (mapreduce/spark) triggered by the job for a particular execution.',
+  flow_exec_id     VARCHAR(255)                             COMMENT 'A unique reference to a specific flow execution. This should filter all applications fired by a particular flow execution. Note that if the scheduler supports sub-workflows, then this ID should be the super parent flow execution id that triggered the the applications and sub-workflows.',
+  job_def_id       VARCHAR(800)                             COMMENT 'A unique reference to the job in the entire flow independent of the execution. This should filter all the applications(mapreduce/spark) triggered by the job for all the historic executions of that job.',
+  flow_def_id      VARCHAR(800)                             COMMENT 'A unique reference to the entire flow independent of any execution. This should filter all the historic mr jobs belonging to the flow. Note that if your scheduler supports sub-workflows, then this ID should reference the super parent flow that triggered the all the jobs and sub-workflows.',
+  job_exec_url     VARCHAR(800)                             COMMENT 'A url to the job execution on the scheduler',
+  flow_exec_url    VARCHAR(800)                             COMMENT 'A url to the flow execution on the scheduler',
+  job_def_url      VARCHAR(800)                             COMMENT 'A url to the job definition on the scheduler',
+  flow_def_url     VARCHAR(800)                             COMMENT 'A url to the flow definition on the scheduler',
+  error            TEXT                             COMMENT 'error stack trace',
+
+  PRIMARY KEY (id)
+);
+
 
 CREATE TABLE `check_point` (
   `id`              BIGINT          NOT NULL AUTO_INCREMENT,
@@ -211,6 +234,8 @@ DROP TABLE QRTZ_FIRED_TRIGGERS;
 
 DROP TABLE check_point;
 
+DROP TABLE failed_app_result;
+
 SET FOREIGN_KEY_CHECKS=1;
 
 # --- Created by Ebean DDL

From 3f937431ad7b2fdb63e3a278f2579cbc8b40b0c9 Mon Sep 17 00:00:00 2001
From: shubhamjainflipkart <shubham.j@flipkart.com>
Date: Mon, 29 Jan 2018 19:14:17 +0530
Subject: [PATCH 2/6] uncommenting retries logic

---
 .../analysis/AnalyticJobGeneratorHadoop2.java | 26 +++++++++----------
 1 file changed, 13 insertions(+), 13 deletions(-)

diff --git a/app/com/linkedin/drelephant/analysis/AnalyticJobGeneratorHadoop2.java b/app/com/linkedin/drelephant/analysis/AnalyticJobGeneratorHadoop2.java
index bdf4cc3c1..9b6901705 100644
--- a/app/com/linkedin/drelephant/analysis/AnalyticJobGeneratorHadoop2.java
+++ b/app/com/linkedin/drelephant/analysis/AnalyticJobGeneratorHadoop2.java
@@ -231,18 +231,18 @@ public void analyseJob(AnalyticJob analyticJob) {
       logger.error(e.getMessage());
       logger.error(ExceptionUtils.getStackTrace(e));
 
-//      if (analyticJob != null && analyticJob.isPrimaryPhaseRetry()) {
-//        logger.error("Add analytic job id [" + analyticJob.getAppId() + "] into the retry list.");
-//        MetricsController.triggerJobFailedEvent(analyticJob.getRetriesCount());
-//        ElephantRunner.getInstance().getExecutorService().onPrimaryRetry(analyticJob);
-//
-//      } else if (analyticJob != null && analyticJob.isSecondPhaseRetry()) {
-//        //Putting the job into a second retry queue which fetches jobs after some interval. Some spark jobs may need more time than usual to process, hence the queue.
-//        logger.error("Add analytic job id [" + analyticJob.getAppId() + "] into the second retry list.");
-//        MetricsController.triggerJobFailedEvent(analyticJob.getRetriesCount());
-//        ElephantRunner.getInstance().getExecutorService().onSecondaryRetry(analyticJob);
-//
-//      } else {
+      if (analyticJob != null && analyticJob.isPrimaryPhaseRetry()) {
+        logger.error("Add analytic job id [" + analyticJob.getAppId() + "] into the retry list.");
+        MetricsController.triggerJobFailedEvent(analyticJob.getRetriesCount());
+        ElephantRunner.getInstance().getExecutorService().onPrimaryRetry(analyticJob);
+
+      } else if (analyticJob != null && analyticJob.isSecondPhaseRetry()) {
+        //Putting the job into a second retry queue which fetches jobs after some interval. Some spark jobs may need more time than usual to process, hence the queue.
+        logger.error("Add analytic job id [" + analyticJob.getAppId() + "] into the second retry list.");
+        MetricsController.triggerJobFailedEvent(analyticJob.getRetriesCount());
+        ElephantRunner.getInstance().getExecutorService().onSecondaryRetry(analyticJob);
+
+      } else {
         if (analyticJob != null) {
           MetricsController.triggerJobRetriesExhaustionEvent();
           try {
@@ -256,7 +256,7 @@ public void analyseJob(AnalyticJob analyticJob) {
           logger.error("Drop the analytic job. Reason: reached the max retries for application id = ["
                   + analyticJob.getAppId() + "].");
         }
-     // }
+      }
     }
   }
 

From 82174f4535fa32f560f215ab32424074d362e7ca Mon Sep 17 00:00:00 2001
From: shubhamjainflipkart <shubham.j@flipkart.com>
Date: Mon, 29 Jan 2018 20:38:33 +0530
Subject: [PATCH 3/6] version bumped to 2.2.1

---
 build.sbt | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/build.sbt b/build.sbt
index bf55803ac..9158ebb38 100644
--- a/build.sbt
+++ b/build.sbt
@@ -19,7 +19,7 @@ import Dependencies._
 
 name := "dr-elephant"
 
-version := "2.2.0"
+version := "2.2.1"
 
 organization := "com.linkedin.drelephant"
 

From fc483433eaff860e1790918268cea10ba3ebffd6 Mon Sep 17 00:00:00 2001
From: shubhamjainflipkart <shubham.j@flipkart.com>
Date: Tue, 30 Jan 2018 10:50:16 +0530
Subject: [PATCH 4/6] removing the test specific code

---
 .../tez/fetchers/TezDataFetcherHadoop2.java   | 23 +++----------------
 1 file changed, 3 insertions(+), 20 deletions(-)

diff --git a/app/com/linkedin/drelephant/tez/fetchers/TezDataFetcherHadoop2.java b/app/com/linkedin/drelephant/tez/fetchers/TezDataFetcherHadoop2.java
index f2be27ce3..49b784582 100644
--- a/app/com/linkedin/drelephant/tez/fetchers/TezDataFetcherHadoop2.java
+++ b/app/com/linkedin/drelephant/tez/fetchers/TezDataFetcherHadoop2.java
@@ -138,7 +138,6 @@ public TezDAGApplicationData fetchData(AnalyticJob analyticJob) throws IOExcepti
 
         TezVertexData[] tezVertexData=null;
         String dagId = null;
-        //TezDAGData[] tezDAGData=new TezDAGData[dagCount];
         List<TezDAGData> tezDAGDataList = new ArrayList<TezDAGData>();
         List<TezVertexTaskData> mapperList = new ArrayList<TezVertexTaskData>();
         List<TezVertexTaskData> reducerList = new ArrayList<TezVertexTaskData>();
@@ -242,9 +241,7 @@ private URL getJobURL(String jobId) throws MalformedURLException {
 
     private URL getJobConfigURL(String appId) throws MalformedURLException {
       appId = "tez_"+appId;
-      //System.out.println(appId+"appId");
       appId =_tezRoot+"TEZ_APPLICATION/"+appId;
-      //System.out.println(appId+"appId");
       return new URL(appId);
     }
 
@@ -275,25 +272,14 @@ private URL getTezDAGURL(String appId,int id) throws MalformedURLException {
     }
     private String getTezDAGId(String appId,int id) throws MalformedURLException {
       String dagId = appId.replace("application","dag");
-      //	System.out.println(dagId+"dagId");
       dagId =dagId+"_"+id;
-      //	System.out.println(dagId+"dagId");
       return (dagId);
     }
     private URL getTezDagSubmittedCountURL(String appId) throws MalformedURLException{
       return new URL(_applicationHistoryAddress+appId);
     }
-    private URL getTezDAGURL(String dagId) throws MalformedURLException {
-      return new URL(_tezRoot+"TEZ_DAG_ID/"+dagId);
-    }
-
-    private URL getTezVertexURL(String tezVertexId) throws MalformedURLException {
-
-      String tezVertexURL = _tezRoot+"TEZ_VERTEX_ID/"+tezVertexId;
-      return new URL(tezVertexURL);
-    }
-    private URL getTezTaskIdURL(String tezTaskId) throws MalformedURLException {
-      String tezTaskIdURL = _tezRoot+"TEZ_TASK_ID/"+tezTaskId;
+    private URL getTezTaskIdURL(String vertexId) throws MalformedURLException {
+      String tezTaskIdURL = _tezRoot + "TEZ_TASK_ID?limit=9007199254740991&primaryFilter=TEZ_VERTEX_ID:" + vertexId + "&secondaryFilter=status:SUCCEEDED";
       return new URL(tezTaskIdURL);
     }
 
@@ -355,8 +341,6 @@ private Properties getProperties(URL url) throws IOException, AuthenticationExce
     private TezCounterData getJobCounter(URL url) throws IOException, AuthenticationException {
       TezCounterData holder = new TezCounterData();
 
-      //  URL tezURL = _urlFactory.getTezDAGURL(_dagId);
-      //new URL("http://localhost:8188/ws/v1/timeline/TEZ_DAG_ID/dag_1475423301689_0009_1");
       JsonNode rootNodeTez = ThreadContextTez.readJsonNode(url);
       JsonNode groupsTez = rootNodeTez.path("otherinfo").path("counters").path("counterGroups");
       for (JsonNode group : groupsTez) {
@@ -461,8 +445,7 @@ private void getTaskDataAll(List<TezVertexData> tezVertexList, List<TezVertexTas
           }
         }
         tezVertexData.setCounter(holder);
-
-        Iterator<JsonNode> taskRootNode = ThreadContextTez.readJsonNode(new URL("http://prod-fdphadoop-bheema-hs-0001:8188/ws/v1/timeline/TEZ_TASK_ID?limit=9007199254740991&primaryFilter=TEZ_VERTEX_ID:"+vertexId+"&secondaryFilter=status:SUCCEEDED")).path("entities").getElements();
+        Iterator<JsonNode> taskRootNode = ThreadContextTez.readJsonNode( _urlFactory.getTezTaskIdURL(vertexId)).path("entities").getElements();
         while(taskRootNode.hasNext()){
           JsonNode taskNode = taskRootNode.next();
           String taskId=taskNode.get("entity").getValueAsText();

From 0591710f997c116a14e01d13237e1487231d44a0 Mon Sep 17 00:00:00 2001
From: shubhamjainflipkart <shubham.j@flipkart.com>
Date: Wed, 31 Jan 2018 00:30:40 +0530
Subject: [PATCH 5/6] correcting Quartzscheduling retries logic

---
 app/com/linkedin/drelephant/analysis/AnalyticJob.java      | 7 ++++++-
 .../drelephant/executors/QuartzExecutorService.java        | 2 ++
 conf/evolutions/default/5.sql                              | 2 +-
 3 files changed, 9 insertions(+), 2 deletions(-)

diff --git a/app/com/linkedin/drelephant/analysis/AnalyticJob.java b/app/com/linkedin/drelephant/analysis/AnalyticJob.java
index c9c5951a3..37c2005b3 100644
--- a/app/com/linkedin/drelephant/analysis/AnalyticJob.java
+++ b/app/com/linkedin/drelephant/analysis/AnalyticJob.java
@@ -396,6 +396,11 @@ public boolean isSecondPhaseRetry(){
    * @return true if should retry, else false
    */
   public boolean isPrimaryPhaseRetry() {
-    return (_retries++) < _RETRY_LIMIT;
+    if(_retries < _RETRY_LIMIT) {
+      _retries++;
+      return true;
+    }
+    // Not incrementing _retries if condition fails so as to get correct value of total retries in case of QuartzScheduling.
+    return false;
   }
 }
diff --git a/app/com/linkedin/drelephant/executors/QuartzExecutorService.java b/app/com/linkedin/drelephant/executors/QuartzExecutorService.java
index 139821c7d..284e5d05f 100644
--- a/app/com/linkedin/drelephant/executors/QuartzExecutorService.java
+++ b/app/com/linkedin/drelephant/executors/QuartzExecutorService.java
@@ -165,6 +165,8 @@ public void execute(AnalyticJob analyticJob) {
             logger.error("job already exist with app_id: "+ analyticJob.getAppId());
         } catch (SchedulerException e) {
             throw new RuntimeException("Error while setting up scheduler : ", e);
+        } finally {
+            interval = 0;
         }
     }
 
diff --git a/conf/evolutions/default/5.sql b/conf/evolutions/default/5.sql
index 54a357abb..93fa54ccf 100644
--- a/conf/evolutions/default/5.sql
+++ b/conf/evolutions/default/5.sql
@@ -34,7 +34,7 @@ CREATE TABLE failed_app_result (
   flow_exec_url    VARCHAR(800)                             COMMENT 'A url to the flow execution on the scheduler',
   job_def_url      VARCHAR(800)                             COMMENT 'A url to the job definition on the scheduler',
   flow_def_url     VARCHAR(800)                             COMMENT 'A url to the flow definition on the scheduler',
-  error            TEXT                             COMMENT 'error stack trace',
+  error            TEXT                                     COMMENT 'error stack trace',
 
   PRIMARY KEY (id)
 );

From 13aed04cefca6516a161b3a23a2655269b54216e Mon Sep 17 00:00:00 2001
From: shubhamjainflipkart <shubham.j@flipkart.com>
Date: Wed, 31 Jan 2018 00:31:34 +0530
Subject: [PATCH 6/6] version bumped to 2.2.2

---
 build.sbt | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/build.sbt b/build.sbt
index 9158ebb38..b3540229c 100644
--- a/build.sbt
+++ b/build.sbt
@@ -19,7 +19,7 @@ import Dependencies._
 
 name := "dr-elephant"
 
-version := "2.2.1"
+version := "2.2.2"
 
 organization := "com.linkedin.drelephant"