You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kylin.apache.org by nj...@apache.org on 2016/06/03 12:31:26 UTC
[2/3] kylin git commit: KYLIN-1189: Automatically resume running jobs
when job engine failover
KYLIN-1189: Automatically resume running jobs when job engine failover
Project: http://git-wip-us.apache.org/repos/asf/kylin/repo
Commit: http://git-wip-us.apache.org/repos/asf/kylin/commit/b2208860
Tree: http://git-wip-us.apache.org/repos/asf/kylin/tree/b2208860
Diff: http://git-wip-us.apache.org/repos/asf/kylin/diff/b2208860
Branch: refs/heads/1.4-rc
Commit: b2208860fc4855018931fb686233dd7ca4388a54
Parents: d02a5f5
Author: Zhong <ya...@lm-shc-16501214.corp.ebay.com>
Authored: Mon May 30 11:27:44 2016 +0800
Committer: Zhong <ya...@lm-shc-16501214.corp.ebay.com>
Committed: Mon May 30 11:27:44 2016 +0800
----------------------------------------------------------------------
.../kylin/job/impl/threadpool/DefaultScheduler.java | 2 +-
.../apache/kylin/job/manager/ExecutableManager.java | 15 +++++++++++++++
2 files changed, 16 insertions(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/kylin/blob/b2208860/core-job/src/main/java/org/apache/kylin/job/impl/threadpool/DefaultScheduler.java
----------------------------------------------------------------------
diff --git a/core-job/src/main/java/org/apache/kylin/job/impl/threadpool/DefaultScheduler.java b/core-job/src/main/java/org/apache/kylin/job/impl/threadpool/DefaultScheduler.java
index 417e279..4abaa23 100644
--- a/core-job/src/main/java/org/apache/kylin/job/impl/threadpool/DefaultScheduler.java
+++ b/core-job/src/main/java/org/apache/kylin/job/impl/threadpool/DefaultScheduler.java
@@ -175,7 +175,7 @@ public class DefaultScheduler implements Scheduler<AbstractExecutable>, Connecti
executableManager.updateJobOutput(executable.getId(), ExecutableState.ERROR, null, "scheduler initializing work to reset job to ERROR status");
}
}
- executableManager.updateAllRunningJobsToError();
+ executableManager.resumeAllRunningJobs();
Runtime.getRuntime().addShutdownHook(new Thread() {
public void run() {
http://git-wip-us.apache.org/repos/asf/kylin/blob/b2208860/core-job/src/main/java/org/apache/kylin/job/manager/ExecutableManager.java
----------------------------------------------------------------------
diff --git a/core-job/src/main/java/org/apache/kylin/job/manager/ExecutableManager.java b/core-job/src/main/java/org/apache/kylin/job/manager/ExecutableManager.java
index ff92a50..244c295 100644
--- a/core-job/src/main/java/org/apache/kylin/job/manager/ExecutableManager.java
+++ b/core-job/src/main/java/org/apache/kylin/job/manager/ExecutableManager.java
@@ -190,6 +190,21 @@ public class ExecutableManager {
}
}
+ public void resumeAllRunningJobs() {
+ try {
+ final List<ExecutableOutputPO> jobOutputs = executableDao.getJobOutputs();
+ for (ExecutableOutputPO executableOutputPO : jobOutputs) {
+ if (executableOutputPO.getStatus().equalsIgnoreCase(ExecutableState.RUNNING.toString())) {
+ executableOutputPO.setStatus(ExecutableState.READY.toString());
+ executableDao.updateJobOutput(executableOutputPO);
+ }
+ }
+ } catch (PersistentException e) {
+ logger.error("error reset job status from RUNNING to READY", e);
+ throw new RuntimeException(e);
+ }
+ }
+
public void resumeJob(String jobId) {
AbstractExecutable job = getJob(jobId);
if (job == null) {