From 71b27cc2e2d49a95524b6181805d317977ea2bf3 Mon Sep 17 00:00:00 2001
From: 10069681 <peng.jianhua@zte.com.cn>
Date: Thu, 1 Jun 2017 20:11:01 +0800
Subject: [PATCH 1/1] KYLIN-2655 There are some minor problems with the
 duration of the job when resuming the error job or stopped job.

---
 .../main/java/org/apache/kylin/job/JobInstance.java | 16 ++++++++++++++++
 .../kylin/job/execution/AbstractExecutable.java     | 21 +++++++++++++++++----
 .../job/execution/DefaultChainedExecutable.java     | 12 ++++++++++--
 .../kylin/engine/mr/common/JobInfoConverter.java    |  3 ++-
 .../org/apache/kylin/tool/JobInstanceExtractor.java |  3 ++-
 5 files changed, 47 insertions(+), 8 deletions(-)

diff --git a/core-job/src/main/java/org/apache/kylin/job/JobInstance.java b/core-job/src/main/java/org/apache/kylin/job/JobInstance.java
index 8dcdff676..377883419 100644
--- a/core-job/src/main/java/org/apache/kylin/job/JobInstance.java
+++ b/core-job/src/main/java/org/apache/kylin/job/JobInstance.java
@@ -58,6 +58,8 @@ public class JobInstance extends RootPersistentEntity implements Comparable<JobI
     private long execStartTime;
     @JsonProperty("exec_end_time")
     private long execEndTime;
+    @JsonProperty("exec_interrupt_time")
+    private long execInterruptTime;
     @JsonProperty("mr_waiting")
     private long mrWaiting = 0;
     @JsonManagedReference
@@ -203,6 +205,20 @@ public class JobInstance extends RootPersistentEntity implements Comparable<JobI
     }
 
     /**
+     * @return the execInterruptTime
+     */
+    public long getExecInterruptTime() {
+        return execInterruptTime;
+    }
+
+    /**
+     * @param execInterruptTime the execInterruptTime to set
+     */
+    public void setExecInterruptTime(long execInterruptTime) {
+        this.execInterruptTime = execInterruptTime;
+    }
+
+    /**
      * @param execEndTime the execEndTime to set
      */
     public void setExecEndTime(long execEndTime) {
diff --git a/core-job/src/main/java/org/apache/kylin/job/execution/AbstractExecutable.java b/core-job/src/main/java/org/apache/kylin/job/execution/AbstractExecutable.java
index 1f1be4126..d36f598af 100644
--- a/core-job/src/main/java/org/apache/kylin/job/execution/AbstractExecutable.java
+++ b/core-job/src/main/java/org/apache/kylin/job/execution/AbstractExecutable.java
@@ -48,6 +48,7 @@ public abstract class AbstractExecutable implements Executable, Idempotent {
     protected static final String NOTIFY_LIST = "notify_list";
     protected static final String START_TIME = "startTime";
     protected static final String END_TIME = "endTime";
+    protected static final String INTERRUPT_TIME = "interruptTime";
 
     protected static final Logger logger = LoggerFactory.getLogger(AbstractExecutable.class);
     protected int retry = 0;
@@ -322,15 +323,19 @@ public abstract class AbstractExecutable implements Executable, Idempotent {
     public static long getEndTime(Output output) {
         return getExtraInfoAsLong(output, END_TIME, 0L);
     }
+    
+    public static long getInterruptTime(Output output) {
+        return getExtraInfoAsLong(output, INTERRUPT_TIME, 0L);
+    }
 
-    public static long getDuration(long startTime, long endTime) {
+    public static long getDuration(long startTime, long endTime, long interruptTime) {
         if (startTime == 0) {
             return 0;
         }
         if (endTime == 0) {
-            return System.currentTimeMillis() - startTime;
+            return System.currentTimeMillis() - startTime - interruptTime;
         } else {
-            return endTime - startTime;
+            return endTime - startTime - interruptTime;
         }
     }
 
@@ -359,6 +364,10 @@ public abstract class AbstractExecutable implements Executable, Idempotent {
         addExtraInfo(END_TIME, time + "");
     }
 
+    public final void setInterruptTime(long time) {
+        addExtraInfo(INTERRUPT_TIME, time + "");
+    }
+
     public final long getStartTime() {
         return getExtraInfoAsLong(START_TIME, 0L);
     }
@@ -367,8 +376,12 @@ public abstract class AbstractExecutable implements Executable, Idempotent {
         return getExtraInfoAsLong(END_TIME, 0L);
     }
 
+    public final long getInterruptTime() {
+        return getExtraInfoAsLong(INTERRUPT_TIME, 0L);
+    }
+
     public final long getDuration() {
-        return getDuration(getStartTime(), getEndTime());
+        return getDuration(getStartTime(), getEndTime(), getInterruptTime());
     }
 
     /*
diff --git a/core-job/src/main/java/org/apache/kylin/job/execution/DefaultChainedExecutable.java b/core-job/src/main/java/org/apache/kylin/job/execution/DefaultChainedExecutable.java
index 8bcaaad1f..ae129abdf 100644
--- a/core-job/src/main/java/org/apache/kylin/job/execution/DefaultChainedExecutable.java
+++ b/core-job/src/main/java/org/apache/kylin/job/execution/DefaultChainedExecutable.java
@@ -70,11 +70,19 @@ public class DefaultChainedExecutable extends AbstractExecutable implements Chai
     @Override
     protected void onExecuteStart(ExecutableContext executableContext) {
         Map<String, String> info = Maps.newHashMap();
-        info.put(START_TIME, Long.toString(System.currentTimeMillis()));
         final long startTime = getStartTime();
         if (startTime > 0) {
-            getManager().updateJobOutput(getId(), ExecutableState.RUNNING, null, null);
+            final long endTime = getEndTime();
+            if (endTime > 0) {
+                long interruptTime = System.currentTimeMillis() - endTime + getInterruptTime();
+                info.put(START_TIME, Long.toString(startTime));
+                info.put(INTERRUPT_TIME, Long.toString(interruptTime));
+                getManager().updateJobOutput(getId(), ExecutableState.RUNNING, info, null);
+            } else {
+                getManager().updateJobOutput(getId(), ExecutableState.RUNNING, null, null);
+            }
         } else {
+            info.put(START_TIME, Long.toString(System.currentTimeMillis()));
             getManager().updateJobOutput(getId(), ExecutableState.RUNNING, info, null);
         }
     }
diff --git a/engine-mr/src/main/java/org/apache/kylin/engine/mr/common/JobInfoConverter.java b/engine-mr/src/main/java/org/apache/kylin/engine/mr/common/JobInfoConverter.java
index 189e0194e..c465e3f78 100644
--- a/engine-mr/src/main/java/org/apache/kylin/engine/mr/common/JobInfoConverter.java
+++ b/engine-mr/src/main/java/org/apache/kylin/engine/mr/common/JobInfoConverter.java
@@ -53,7 +53,8 @@ public class JobInfoConverter {
         result.setMrWaiting(AbstractExecutable.getExtraInfoAsLong(output, CubingJob.MAP_REDUCE_WAIT_TIME, 0L) / 1000);
         result.setExecStartTime(AbstractExecutable.getStartTime(output));
         result.setExecEndTime(AbstractExecutable.getEndTime(output));
-        result.setDuration(AbstractExecutable.getDuration(result.getExecStartTime(), result.getExecEndTime()) / 1000);
+        result.setExecInterruptTime(AbstractExecutable.getInterruptTime(output));
+        result.setDuration(AbstractExecutable.getDuration(result.getExecStartTime(), result.getExecEndTime(), result.getExecInterruptTime()) / 1000);
         for (int i = 0; i < cubeJob.getTasks().size(); ++i) {
             AbstractExecutable task = cubeJob.getTasks().get(i);
             result.addStep(parseToJobStep(task, i, outputs.get(task.getId())));
diff --git a/tool/src/main/java/org/apache/kylin/tool/JobInstanceExtractor.java b/tool/src/main/java/org/apache/kylin/tool/JobInstanceExtractor.java
index 068dbda4f..52fd0a0f0 100644
--- a/tool/src/main/java/org/apache/kylin/tool/JobInstanceExtractor.java
+++ b/tool/src/main/java/org/apache/kylin/tool/JobInstanceExtractor.java
@@ -135,7 +135,8 @@ public class JobInstanceExtractor extends AbstractInfoExtractor {
         result.setMrWaiting(AbstractExecutable.getExtraInfoAsLong(output, CubingJob.MAP_REDUCE_WAIT_TIME, 0L) / 1000);
         result.setExecStartTime(AbstractExecutable.getStartTime(output));
         result.setExecEndTime(AbstractExecutable.getEndTime(output));
-        result.setDuration(AbstractExecutable.getDuration(AbstractExecutable.getStartTime(output), AbstractExecutable.getEndTime(output)) / 1000);
+        result.setExecInterruptTime(AbstractExecutable.getInterruptTime(output));
+        result.setDuration(AbstractExecutable.getDuration(AbstractExecutable.getStartTime(output), AbstractExecutable.getEndTime(output), AbstractExecutable.getInterruptTime(output)) / 1000);
         for (int i = 0; i < cubeJob.getTasks().size(); ++i) {
             AbstractExecutable task = cubeJob.getTasks().get(i);
             result.addStep(parseToJobStep(task, i, outputs.get(task.getId())));
-- 
2.11.0.windows.1