You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kylin.apache.org by li...@apache.org on 2015/01/23 09:32:33 UTC
[46/50] [abbrv] incubator-kylin git commit: renaming
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/ccfecdb3/job/src/main/java/com/kylinolap/job/manager/ExecutableManager.java
----------------------------------------------------------------------
diff --git a/job/src/main/java/com/kylinolap/job/manager/ExecutableManager.java b/job/src/main/java/com/kylinolap/job/manager/ExecutableManager.java
new file mode 100644
index 0000000..5c18952
--- /dev/null
+++ b/job/src/main/java/com/kylinolap/job/manager/ExecutableManager.java
@@ -0,0 +1,298 @@
+package com.kylinolap.job.manager;
+
+import com.google.common.base.Function;
+import com.google.common.base.Preconditions;
+import com.google.common.collect.Lists;
+import com.google.common.collect.Maps;
+import com.kylinolap.common.KylinConfig;
+import com.kylinolap.job.dao.ExecutableDao;
+import com.kylinolap.job.dao.ExecutableOutputPO;
+import com.kylinolap.job.dao.ExecutablePO;
+import com.kylinolap.job.exception.IllegalStateTranferException;
+import com.kylinolap.job.exception.PersistentException;
+import com.kylinolap.job.execution.DefaultOutput;
+import com.kylinolap.job.execution.ExecutableState;
+import com.kylinolap.job.execution.Output;
+import com.kylinolap.job.execution.AbstractExecutable;
+import com.kylinolap.job.execution.DefaultChainedExecutable;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.annotation.Nullable;
+import java.lang.reflect.Constructor;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+
+/**
+ * Created by qianzhou on 12/16/14.
+ */
+public class ExecutableManager {
+
+ private static final Logger logger = LoggerFactory.getLogger(ExecutableManager.class);
+ private static final ConcurrentHashMap<KylinConfig, ExecutableManager> CACHE = new ConcurrentHashMap<KylinConfig, ExecutableManager>();
+ private final KylinConfig config;
+
+ private ExecutableDao executableDao;
+
+ public static ExecutableManager getInstance(KylinConfig config) {
+ ExecutableManager r = CACHE.get(config);
+ if (r == null) {
+ r = new ExecutableManager(config);
+ CACHE.put(config, r);
+ if (CACHE.size() > 1) {
+ logger.warn("More than one singleton exist");
+ }
+
+ }
+ return r;
+ }
+
+ private ExecutableManager(KylinConfig config) {
+ logger.info("Using metadata url: " + config);
+ this.config = config;
+ this.executableDao = ExecutableDao.getInstance(config);
+ }
+
+ public void addJob(AbstractExecutable executable) {
+ try {
+ executableDao.addJob(getJobPO(executable));
+ addJobOutput(executable);
+ } catch (PersistentException e) {
+ logger.error("fail to submit job:" + executable.getId(), e);
+ throw new RuntimeException(e);
+ }
+ }
+
+ private void addJobOutput(AbstractExecutable executable) throws PersistentException {
+ ExecutableOutputPO executableOutputPO = new ExecutableOutputPO();
+ executableOutputPO.setUuid(executable.getId());
+ executableDao.addJobOutput(executableOutputPO);
+ if (executable instanceof DefaultChainedExecutable) {
+ for (AbstractExecutable subTask: ((DefaultChainedExecutable) executable).getTasks()) {
+ addJobOutput(subTask);
+ }
+ }
+ }
+
+ //for ut
+ public void deleteJob(String jobId) {
+ try {
+ executableDao.deleteJob(jobId);
+ } catch (PersistentException e) {
+ logger.error("fail to delete job:" + jobId, e);
+ throw new RuntimeException(e);
+ }
+ }
+
+ public AbstractExecutable getJob(String uuid) {
+ try {
+ return parseTo(executableDao.getJob(uuid));
+ } catch (PersistentException e) {
+ logger.error("fail to get job:" + uuid, e);
+ throw new RuntimeException(e);
+ }
+ }
+
+ public Output getOutput(String uuid) {
+ try {
+ final ExecutableOutputPO jobOutput = executableDao.getJobOutput(uuid);
+ Preconditions.checkArgument(jobOutput != null, "there is no related output for job id:" + uuid);
+ final DefaultOutput result = new DefaultOutput();
+ result.setExtra(jobOutput.getInfo());
+ result.setState(ExecutableState.valueOf(jobOutput.getStatus()));
+ result.setVerboseMsg(jobOutput.getContent());
+ result.setLastModified(jobOutput.getLastModified());
+ return result;
+ } catch (PersistentException e) {
+ logger.error("fail to get job output:" + uuid, e);
+ throw new RuntimeException(e);
+ }
+ }
+
+ public List<AbstractExecutable> getAllExecutables() {
+ try {
+ return Lists.transform(executableDao.getJobs(), new Function<ExecutablePO, AbstractExecutable>() {
+ @Nullable
+ @Override
+ public AbstractExecutable apply(ExecutablePO input) {
+ return parseTo(input);
+ }
+ });
+ } catch (PersistentException e) {
+ logger.error("error get All Jobs", e);
+ throw new RuntimeException(e);
+ }
+ }
+
+ public List<String> getAllJobIds() {
+ try {
+ return executableDao.getJobIds();
+ } catch (PersistentException e) {
+ logger.error("error get All Job Ids", e);
+ throw new RuntimeException(e);
+ }
+ }
+
+ public void updateAllRunningJobsToError() {
+ try {
+ final List<ExecutableOutputPO> jobOutputs = executableDao.getJobOutputs();
+ for (ExecutableOutputPO executableOutputPO : jobOutputs) {
+ if (executableOutputPO.getStatus().equalsIgnoreCase(ExecutableState.RUNNING.toString())) {
+ executableOutputPO.setStatus(ExecutableState.ERROR.toString());
+ executableDao.updateJobOutput(executableOutputPO);
+ }
+ }
+ } catch (PersistentException e) {
+ logger.error("error reset job status from RUNNING to ERROR", e);
+ throw new RuntimeException(e);
+ }
+ }
+
+ public void resumeJob(String jobId) {
+ AbstractExecutable job = getJob(jobId);
+ if (job == null) {
+ return;
+ }
+ updateJobOutput(jobId, ExecutableState.READY, null, null);
+ if (job instanceof DefaultChainedExecutable) {
+ List<AbstractExecutable> tasks = ((DefaultChainedExecutable) job).getTasks();
+ for (AbstractExecutable task : tasks) {
+ if (task.getStatus() == ExecutableState.ERROR) {
+ updateJobOutput(task.getId(), ExecutableState.READY, null, null);
+ break;
+ }
+ }
+ }
+ }
+
+ public void discardJob(String jobId) {
+ AbstractExecutable job = getJob(jobId);
+ if (job instanceof DefaultChainedExecutable) {
+ List<AbstractExecutable> tasks = ((DefaultChainedExecutable) job).getTasks();
+ for (AbstractExecutable task : tasks) {
+ if (!task.getStatus().isFinalState()) {
+ updateJobOutput(task.getId(), ExecutableState.DISCARDED, null, null);
+ }
+ }
+ }
+ updateJobOutput(jobId, ExecutableState.DISCARDED, null, null);
+ }
+
+ public void updateJobOutput(String jobId, ExecutableState newStatus, Map<String, String> info, String output) {
+ try {
+ final ExecutableOutputPO jobOutput = executableDao.getJobOutput(jobId);
+ Preconditions.checkArgument(jobOutput != null, "there is no related output for job id:" + jobId);
+ ExecutableState oldStatus = ExecutableState.valueOf(jobOutput.getStatus());
+ if (newStatus != null && oldStatus != newStatus) {
+ if (!ExecutableState.isValidStateTransfer(oldStatus, newStatus)) {
+ throw new IllegalStateTranferException("there is no valid state transfer from:" + oldStatus + " to:" + newStatus);
+ }
+ jobOutput.setStatus(newStatus.toString());
+ }
+ if (info != null) {
+ jobOutput.setInfo(info);
+ }
+ if (output != null) {
+ jobOutput.setContent(output);
+ }
+ executableDao.updateJobOutput(jobOutput);
+ logger.info("job id:" + jobId + " from " + oldStatus + " to " + newStatus);
+ } catch (PersistentException e) {
+ logger.error("error change job:" + jobId + " to " + newStatus.toString());
+ throw new RuntimeException(e);
+ }
+ }
+
+ //for migration only
+ //TODO delete when migration finished
+ public void resetJobOutput(String jobId, ExecutableState state, String output) {
+ try {
+ final ExecutableOutputPO jobOutput = executableDao.getJobOutput(jobId);
+ jobOutput.setStatus(state.toString());
+ if (output != null) {
+ jobOutput.setContent(output);
+ }
+ executableDao.updateJobOutput(jobOutput);
+ } catch (PersistentException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ public void addJobInfo(String id, Map<String, String> info) {
+ if (info == null) {
+ return;
+ }
+ try {
+ ExecutableOutputPO output = executableDao.getJobOutput(id);
+ Preconditions.checkArgument(output != null, "there is no related output for job id:" + id);
+ output.getInfo().putAll(info);
+ executableDao.updateJobOutput(output);
+ } catch (PersistentException e) {
+ logger.error("error update job info, id:" + id + " info:" + info.toString());
+ throw new RuntimeException(e);
+ }
+ }
+
+ public void addJobInfo(String id, String key, String value) {
+ Map<String, String> info = Maps.newHashMap();
+ info.put(key, value);
+ addJobInfo(id, info);
+ }
+
+ private void stopJob(AbstractExecutable job) {
+ final ExecutableState status = job.getStatus();
+ if (status == ExecutableState.RUNNING) {
+ updateJobOutput(job.getId(), ExecutableState.STOPPED, null, null);
+ if (job instanceof DefaultChainedExecutable) {
+ final List<AbstractExecutable> tasks = ((DefaultChainedExecutable) job).getTasks();
+ for (AbstractExecutable task: tasks) {
+ if (task.getStatus() == ExecutableState.RUNNING) {
+ stopJob(task);
+ break;
+ }
+ }
+ }
+ } else {
+ updateJobOutput(job.getId(), ExecutableState.STOPPED, null, null);
+ }
+ }
+
+
+ private ExecutablePO getJobPO(AbstractExecutable executable) {
+ final ExecutablePO result = executable.getJobPO();
+ if (executable instanceof DefaultChainedExecutable) {
+ for (AbstractExecutable task: ((DefaultChainedExecutable) executable).getTasks()) {
+ result.getTasks().add(getJobPO(task));
+ }
+ }
+ return result;
+ }
+
+ public String getAdminDls() {
+ return config.getAdminDls();
+ }
+
+ private AbstractExecutable parseTo(ExecutablePO executablePO) {
+ if (executablePO == null) {
+ return null;
+ }
+ String type = executablePO.getType();
+ try {
+ Class<? extends AbstractExecutable> clazz = (Class<? extends AbstractExecutable>) Class.forName(type);
+ Constructor<? extends AbstractExecutable> constructor = clazz.getConstructor(ExecutablePO.class);
+ AbstractExecutable result = constructor.newInstance(executablePO);
+ List<ExecutablePO> tasks = executablePO.getTasks();
+ if (tasks != null && !tasks.isEmpty()) {
+ Preconditions.checkArgument(result instanceof DefaultChainedExecutable);
+ for (ExecutablePO subTask: tasks) {
+ ((DefaultChainedExecutable) result).addTask(parseTo(subTask));
+ }
+ }
+ return result;
+ } catch (ReflectiveOperationException e) {
+ throw new IllegalArgumentException("cannot parse this job:" + executablePO.getId(), e);
+ }
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/ccfecdb3/job/src/main/java/com/kylinolap/job/service/ExecutableManager.java
----------------------------------------------------------------------
diff --git a/job/src/main/java/com/kylinolap/job/service/ExecutableManager.java b/job/src/main/java/com/kylinolap/job/service/ExecutableManager.java
deleted file mode 100644
index 4e51056..0000000
--- a/job/src/main/java/com/kylinolap/job/service/ExecutableManager.java
+++ /dev/null
@@ -1,299 +0,0 @@
-package com.kylinolap.job.service;
-
-import com.google.common.base.Function;
-import com.google.common.base.Preconditions;
-import com.google.common.collect.Lists;
-import com.google.common.collect.Maps;
-import com.kylinolap.common.KylinConfig;
-import com.kylinolap.job.dao.JobDao;
-import com.kylinolap.job.dao.JobOutputPO;
-import com.kylinolap.job.dao.JobPO;
-import com.kylinolap.job.exception.IllegalStateTranferException;
-import com.kylinolap.job.exception.PersistentException;
-import com.kylinolap.job.execution.DefaultOutput;
-import com.kylinolap.job.execution.ExecutableState;
-import com.kylinolap.job.execution.Output;
-import com.kylinolap.job.impl.threadpool.AbstractExecutable;
-import com.kylinolap.job.impl.threadpool.DefaultChainedExecutable;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import javax.annotation.Nullable;
-import java.lang.reflect.Constructor;
-import java.util.Collections;
-import java.util.List;
-import java.util.Map;
-import java.util.concurrent.ConcurrentHashMap;
-
-/**
- * Created by qianzhou on 12/16/14.
- */
-public class ExecutableManager {
-
- private static final Logger logger = LoggerFactory.getLogger(ExecutableManager.class);
- private static final ConcurrentHashMap<KylinConfig, ExecutableManager> CACHE = new ConcurrentHashMap<KylinConfig, ExecutableManager>();
- private final KylinConfig config;
-
- private JobDao jobDao;
-
- public static ExecutableManager getInstance(KylinConfig config) {
- ExecutableManager r = CACHE.get(config);
- if (r == null) {
- r = new ExecutableManager(config);
- CACHE.put(config, r);
- if (CACHE.size() > 1) {
- logger.warn("More than one singleton exist");
- }
-
- }
- return r;
- }
-
- private ExecutableManager(KylinConfig config) {
- logger.info("Using metadata url: " + config);
- this.config = config;
- this.jobDao = JobDao.getInstance(config);
- }
-
- public void addJob(AbstractExecutable executable) {
- try {
- jobDao.addJob(getJobPO(executable));
- addJobOutput(executable);
- } catch (PersistentException e) {
- logger.error("fail to submit job:" + executable.getId(), e);
- throw new RuntimeException(e);
- }
- }
-
- private void addJobOutput(AbstractExecutable executable) throws PersistentException {
- JobOutputPO jobOutputPO = new JobOutputPO();
- jobOutputPO.setUuid(executable.getId());
- jobDao.addJobOutput(jobOutputPO);
- if (executable instanceof DefaultChainedExecutable) {
- for (AbstractExecutable subTask: ((DefaultChainedExecutable) executable).getTasks()) {
- addJobOutput(subTask);
- }
- }
- }
-
- //for ut
- public void deleteJob(String jobId) {
- try {
- jobDao.deleteJob(jobId);
- } catch (PersistentException e) {
- logger.error("fail to delete job:" + jobId, e);
- throw new RuntimeException(e);
- }
- }
-
- public AbstractExecutable getJob(String uuid) {
- try {
- return parseTo(jobDao.getJob(uuid));
- } catch (PersistentException e) {
- logger.error("fail to get job:" + uuid, e);
- throw new RuntimeException(e);
- }
- }
-
- public Output getOutput(String uuid) {
- try {
- final JobOutputPO jobOutput = jobDao.getJobOutput(uuid);
- Preconditions.checkArgument(jobOutput != null, "there is no related output for job id:" + uuid);
- final DefaultOutput result = new DefaultOutput();
- result.setExtra(jobOutput.getInfo());
- result.setState(ExecutableState.valueOf(jobOutput.getStatus()));
- result.setVerboseMsg(jobOutput.getContent());
- result.setLastModified(jobOutput.getLastModified());
- return result;
- } catch (PersistentException e) {
- logger.error("fail to get job output:" + uuid, e);
- throw new RuntimeException(e);
- }
- }
-
- public List<AbstractExecutable> getAllExecutables() {
- try {
- return Lists.transform(jobDao.getJobs(), new Function<JobPO, AbstractExecutable>() {
- @Nullable
- @Override
- public AbstractExecutable apply(JobPO input) {
- return parseTo(input);
- }
- });
- } catch (PersistentException e) {
- logger.error("error get All Jobs", e);
- throw new RuntimeException(e);
- }
- }
-
- public List<String> getAllJobIds() {
- try {
- return jobDao.getJobIds();
- } catch (PersistentException e) {
- logger.error("error get All Job Ids", e);
- throw new RuntimeException(e);
- }
- }
-
- public void updateAllRunningJobsToError() {
- try {
- final List<JobOutputPO> jobOutputs = jobDao.getJobOutputs();
- for (JobOutputPO jobOutputPO: jobOutputs) {
- if (jobOutputPO.getStatus().equalsIgnoreCase(ExecutableState.RUNNING.toString())) {
- jobOutputPO.setStatus(ExecutableState.ERROR.toString());
- jobDao.updateJobOutput(jobOutputPO);
- }
- }
- } catch (PersistentException e) {
- logger.error("error reset job status from RUNNING to ERROR", e);
- throw new RuntimeException(e);
- }
- }
-
- public void resumeJob(String jobId) {
- AbstractExecutable job = getJob(jobId);
- if (job == null) {
- return;
- }
- updateJobOutput(jobId, ExecutableState.READY, null, null);
- if (job instanceof DefaultChainedExecutable) {
- List<AbstractExecutable> tasks = ((DefaultChainedExecutable) job).getTasks();
- for (AbstractExecutable task : tasks) {
- if (task.getStatus() == ExecutableState.ERROR) {
- updateJobOutput(task.getId(), ExecutableState.READY, null, null);
- break;
- }
- }
- }
- }
-
- public void discardJob(String jobId) {
- AbstractExecutable job = getJob(jobId);
- if (job instanceof DefaultChainedExecutable) {
- List<AbstractExecutable> tasks = ((DefaultChainedExecutable) job).getTasks();
- for (AbstractExecutable task : tasks) {
- if (!task.getStatus().isFinalState()) {
- updateJobOutput(task.getId(), ExecutableState.DISCARDED, null, null);
- }
- }
- }
- updateJobOutput(jobId, ExecutableState.DISCARDED, null, null);
- }
-
- public void updateJobOutput(String jobId, ExecutableState newStatus, Map<String, String> info, String output) {
- try {
- final JobOutputPO jobOutput = jobDao.getJobOutput(jobId);
- Preconditions.checkArgument(jobOutput != null, "there is no related output for job id:" + jobId);
- ExecutableState oldStatus = ExecutableState.valueOf(jobOutput.getStatus());
- if (newStatus != null && oldStatus != newStatus) {
- if (!ExecutableState.isValidStateTransfer(oldStatus, newStatus)) {
- throw new IllegalStateTranferException("there is no valid state transfer from:" + oldStatus + " to:" + newStatus);
- }
- jobOutput.setStatus(newStatus.toString());
- }
- if (info != null) {
- jobOutput.setInfo(info);
- }
- if (output != null) {
- jobOutput.setContent(output);
- }
- jobDao.updateJobOutput(jobOutput);
- logger.info("job id:" + jobId + " from " + oldStatus + " to " + newStatus);
- } catch (PersistentException e) {
- logger.error("error change job:" + jobId + " to " + newStatus.toString());
- throw new RuntimeException(e);
- }
- }
-
- //for migration only
- //TODO delete when migration finished
- public void resetJobOutput(String jobId, ExecutableState state, String output) {
- try {
- final JobOutputPO jobOutput = jobDao.getJobOutput(jobId);
- jobOutput.setStatus(state.toString());
- if (output != null) {
- jobOutput.setContent(output);
- }
- jobDao.updateJobOutput(jobOutput);
- } catch (PersistentException e) {
- throw new RuntimeException(e);
- }
- }
-
- public void addJobInfo(String id, Map<String, String> info) {
- if (info == null) {
- return;
- }
- try {
- JobOutputPO output = jobDao.getJobOutput(id);
- Preconditions.checkArgument(output != null, "there is no related output for job id:" + id);
- output.getInfo().putAll(info);
- jobDao.updateJobOutput(output);
- } catch (PersistentException e) {
- logger.error("error update job info, id:" + id + " info:" + info.toString());
- throw new RuntimeException(e);
- }
- }
-
- public void addJobInfo(String id, String key, String value) {
- Map<String, String> info = Maps.newHashMap();
- info.put(key, value);
- addJobInfo(id, info);
- }
-
- private void stopJob(AbstractExecutable job) {
- final ExecutableState status = job.getStatus();
- if (status == ExecutableState.RUNNING) {
- updateJobOutput(job.getId(), ExecutableState.STOPPED, null, null);
- if (job instanceof DefaultChainedExecutable) {
- final List<AbstractExecutable> tasks = ((DefaultChainedExecutable) job).getTasks();
- for (AbstractExecutable task: tasks) {
- if (task.getStatus() == ExecutableState.RUNNING) {
- stopJob(task);
- break;
- }
- }
- }
- } else {
- updateJobOutput(job.getId(), ExecutableState.STOPPED, null, null);
- }
- }
-
-
- private JobPO getJobPO(AbstractExecutable executable) {
- final JobPO result = executable.getJobPO();
- if (executable instanceof DefaultChainedExecutable) {
- for (AbstractExecutable task: ((DefaultChainedExecutable) executable).getTasks()) {
- result.getTasks().add(getJobPO(task));
- }
- }
- return result;
- }
-
- public String getAdminDls() {
- return config.getAdminDls();
- }
-
- private AbstractExecutable parseTo(JobPO jobPO) {
- if (jobPO == null) {
- return null;
- }
- String type = jobPO.getType();
- try {
- Class<? extends AbstractExecutable> clazz = (Class<? extends AbstractExecutable>) Class.forName(type);
- Constructor<? extends AbstractExecutable> constructor = clazz.getConstructor(JobPO.class);
- AbstractExecutable result = constructor.newInstance(jobPO);
- List<JobPO> tasks = jobPO.getTasks();
- if (tasks != null && !tasks.isEmpty()) {
- Preconditions.checkArgument(result instanceof DefaultChainedExecutable);
- for (JobPO subTask: tasks) {
- ((DefaultChainedExecutable) result).addTask(parseTo(subTask));
- }
- }
- return result;
- } catch (ReflectiveOperationException e) {
- throw new IllegalArgumentException("cannot parse this job:" + jobPO.getId(), e);
- }
- }
-
-}
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/ccfecdb3/job/src/test/java/com/kylinolap/job/BaseTestExecutable.java
----------------------------------------------------------------------
diff --git a/job/src/test/java/com/kylinolap/job/BaseTestExecutable.java b/job/src/test/java/com/kylinolap/job/BaseTestExecutable.java
index ac58a97..0f94ba5 100644
--- a/job/src/test/java/com/kylinolap/job/BaseTestExecutable.java
+++ b/job/src/test/java/com/kylinolap/job/BaseTestExecutable.java
@@ -1,7 +1,7 @@
package com.kylinolap.job;
-import com.kylinolap.job.dao.JobPO;
-import com.kylinolap.job.impl.threadpool.AbstractExecutable;
+import com.kylinolap.job.dao.ExecutablePO;
+import com.kylinolap.job.execution.AbstractExecutable;
/**
* Created by qianzhou on 12/16/14.
@@ -12,7 +12,7 @@ public abstract class BaseTestExecutable extends AbstractExecutable {
public BaseTestExecutable() {
}
- public BaseTestExecutable(JobPO job) {
+ public BaseTestExecutable(ExecutablePO job) {
super(job);
}
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/ccfecdb3/job/src/test/java/com/kylinolap/job/BuildIIWithEngineTest.java
----------------------------------------------------------------------
diff --git a/job/src/test/java/com/kylinolap/job/BuildIIWithEngineTest.java b/job/src/test/java/com/kylinolap/job/BuildIIWithEngineTest.java
index 4663b28..0de60fe 100644
--- a/job/src/test/java/com/kylinolap/job/BuildIIWithEngineTest.java
+++ b/job/src/test/java/com/kylinolap/job/BuildIIWithEngineTest.java
@@ -27,11 +27,11 @@ import com.kylinolap.invertedindex.IIManager;
import com.kylinolap.invertedindex.IISegment;
import com.kylinolap.job.engine.JobEngineConfig;
import com.kylinolap.job.execution.ExecutableState;
-import com.kylinolap.job.impl.threadpool.AbstractExecutable;
+import com.kylinolap.job.execution.AbstractExecutable;
import com.kylinolap.job.impl.threadpool.DefaultScheduler;
import com.kylinolap.job.invertedindex.IIJob;
import com.kylinolap.job.invertedindex.IIJobBuilder;
-import com.kylinolap.job.service.ExecutableManager;
+import com.kylinolap.job.manager.ExecutableManager;
import com.kylinolap.metadata.realization.RealizationStatusEnum;
/**
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/ccfecdb3/job/src/test/java/com/kylinolap/job/ErrorTestExecutable.java
----------------------------------------------------------------------
diff --git a/job/src/test/java/com/kylinolap/job/ErrorTestExecutable.java b/job/src/test/java/com/kylinolap/job/ErrorTestExecutable.java
index 64ea05b..2af9fdd 100644
--- a/job/src/test/java/com/kylinolap/job/ErrorTestExecutable.java
+++ b/job/src/test/java/com/kylinolap/job/ErrorTestExecutable.java
@@ -1,8 +1,6 @@
package com.kylinolap.job;
-import com.kylinolap.job.BaseTestExecutable;
-import com.kylinolap.job.dao.JobOutputPO;
-import com.kylinolap.job.dao.JobPO;
+import com.kylinolap.job.dao.ExecutablePO;
import com.kylinolap.job.exception.ExecuteException;
import com.kylinolap.job.execution.ExecutableContext;
import com.kylinolap.job.execution.ExecuteResult;
@@ -15,7 +13,7 @@ public class ErrorTestExecutable extends BaseTestExecutable {
public ErrorTestExecutable() {
}
- public ErrorTestExecutable(JobPO job) {
+ public ErrorTestExecutable(ExecutablePO job) {
super(job);
}
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/ccfecdb3/job/src/test/java/com/kylinolap/job/FailedTestExecutable.java
----------------------------------------------------------------------
diff --git a/job/src/test/java/com/kylinolap/job/FailedTestExecutable.java b/job/src/test/java/com/kylinolap/job/FailedTestExecutable.java
index 2d89069..43d3878 100644
--- a/job/src/test/java/com/kylinolap/job/FailedTestExecutable.java
+++ b/job/src/test/java/com/kylinolap/job/FailedTestExecutable.java
@@ -1,7 +1,6 @@
package com.kylinolap.job;
-import com.kylinolap.job.dao.JobOutputPO;
-import com.kylinolap.job.dao.JobPO;
+import com.kylinolap.job.dao.ExecutablePO;
import com.kylinolap.job.exception.ExecuteException;
import com.kylinolap.job.execution.ExecutableContext;
import com.kylinolap.job.execution.ExecuteResult;
@@ -14,7 +13,7 @@ public class FailedTestExecutable extends BaseTestExecutable {
public FailedTestExecutable() {
}
- public FailedTestExecutable(JobPO job) {
+ public FailedTestExecutable(ExecutablePO job) {
super(job);
}
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/ccfecdb3/job/src/test/java/com/kylinolap/job/SelfStopExecutable.java
----------------------------------------------------------------------
diff --git a/job/src/test/java/com/kylinolap/job/SelfStopExecutable.java b/job/src/test/java/com/kylinolap/job/SelfStopExecutable.java
index 6cd911c..afdb5ea 100644
--- a/job/src/test/java/com/kylinolap/job/SelfStopExecutable.java
+++ b/job/src/test/java/com/kylinolap/job/SelfStopExecutable.java
@@ -1,9 +1,8 @@
package com.kylinolap.job;
-import com.kylinolap.job.dao.JobPO;
+import com.kylinolap.job.dao.ExecutablePO;
import com.kylinolap.job.exception.ExecuteException;
import com.kylinolap.job.execution.ExecutableContext;
-import com.kylinolap.job.execution.ExecutableState;
import com.kylinolap.job.execution.ExecuteResult;
/**
@@ -14,7 +13,7 @@ public class SelfStopExecutable extends BaseTestExecutable {
public SelfStopExecutable() {
}
- public SelfStopExecutable(JobPO job) {
+ public SelfStopExecutable(ExecutablePO job) {
super(job);
}
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/ccfecdb3/job/src/test/java/com/kylinolap/job/SucceedTestExecutable.java
----------------------------------------------------------------------
diff --git a/job/src/test/java/com/kylinolap/job/SucceedTestExecutable.java b/job/src/test/java/com/kylinolap/job/SucceedTestExecutable.java
index 8d633b1..b7a739b 100644
--- a/job/src/test/java/com/kylinolap/job/SucceedTestExecutable.java
+++ b/job/src/test/java/com/kylinolap/job/SucceedTestExecutable.java
@@ -1,7 +1,6 @@
package com.kylinolap.job;
-import com.kylinolap.job.dao.JobOutputPO;
-import com.kylinolap.job.dao.JobPO;
+import com.kylinolap.job.dao.ExecutablePO;
import com.kylinolap.job.exception.ExecuteException;
import com.kylinolap.job.execution.ExecutableContext;
import com.kylinolap.job.execution.ExecuteResult;
@@ -14,7 +13,7 @@ public class SucceedTestExecutable extends BaseTestExecutable {
public SucceedTestExecutable() {
}
- public SucceedTestExecutable(JobPO job) {
+ public SucceedTestExecutable(ExecutablePO job) {
super(job);
}
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/ccfecdb3/job/src/test/java/com/kylinolap/job/impl/threadpool/BaseSchedulerTest.java
----------------------------------------------------------------------
diff --git a/job/src/test/java/com/kylinolap/job/impl/threadpool/BaseSchedulerTest.java b/job/src/test/java/com/kylinolap/job/impl/threadpool/BaseSchedulerTest.java
index 4e9a941..81ae0bf 100644
--- a/job/src/test/java/com/kylinolap/job/impl/threadpool/BaseSchedulerTest.java
+++ b/job/src/test/java/com/kylinolap/job/impl/threadpool/BaseSchedulerTest.java
@@ -4,8 +4,9 @@ import com.kylinolap.common.KylinConfig;
import com.kylinolap.common.util.LocalFileMetadataTestCase;
import com.kylinolap.job.constant.ExecutableConstants;
import com.kylinolap.job.engine.JobEngineConfig;
+import com.kylinolap.job.execution.AbstractExecutable;
import com.kylinolap.job.execution.ExecutableState;
-import com.kylinolap.job.service.ExecutableManager;
+import com.kylinolap.job.manager.ExecutableManager;
import org.junit.After;
import org.junit.Before;
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/ccfecdb3/job/src/test/java/com/kylinolap/job/impl/threadpool/DefaultSchedulerTest.java
----------------------------------------------------------------------
diff --git a/job/src/test/java/com/kylinolap/job/impl/threadpool/DefaultSchedulerTest.java b/job/src/test/java/com/kylinolap/job/impl/threadpool/DefaultSchedulerTest.java
index 7616895..5ba3f94 100644
--- a/job/src/test/java/com/kylinolap/job/impl/threadpool/DefaultSchedulerTest.java
+++ b/job/src/test/java/com/kylinolap/job/impl/threadpool/DefaultSchedulerTest.java
@@ -1,6 +1,7 @@
package com.kylinolap.job.impl.threadpool;
import com.kylinolap.job.*;
+import com.kylinolap.job.execution.DefaultChainedExecutable;
import com.kylinolap.job.execution.ExecutableState;
import org.junit.Test;
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/ccfecdb3/job/src/test/java/com/kylinolap/job/manager/ExecutableManagerTest.java
----------------------------------------------------------------------
diff --git a/job/src/test/java/com/kylinolap/job/manager/ExecutableManagerTest.java b/job/src/test/java/com/kylinolap/job/manager/ExecutableManagerTest.java
new file mode 100644
index 0000000..5908624
--- /dev/null
+++ b/job/src/test/java/com/kylinolap/job/manager/ExecutableManagerTest.java
@@ -0,0 +1,126 @@
+package com.kylinolap.job.manager;
+
+import com.kylinolap.common.KylinConfig;
+import com.kylinolap.common.util.LocalFileMetadataTestCase;
+import com.kylinolap.job.BaseTestExecutable;
+import com.kylinolap.job.SucceedTestExecutable;
+import com.kylinolap.job.exception.IllegalStateTranferException;
+import com.kylinolap.job.execution.ChainedExecutable;
+import com.kylinolap.job.execution.Executable;
+import com.kylinolap.job.execution.ExecutableState;
+import com.kylinolap.job.execution.AbstractExecutable;
+import com.kylinolap.job.execution.DefaultChainedExecutable;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+
+import java.util.List;
+
+import static org.junit.Assert.*;
+
+/**
+ * Created by qianzhou on 12/16/14.
+ */
+public class ExecutableManagerTest extends LocalFileMetadataTestCase {
+
+ private ExecutableManager service;
+
+ @Before
+ public void setup() throws Exception {
+ createTestMetadata();
+ service = ExecutableManager.getInstance(KylinConfig.getInstanceFromEnv());
+
+ for (String jobId: service.getAllJobIds()) {
+ System.out.println("deleting " + jobId);
+ service.deleteJob(jobId);
+ }
+
+ }
+
+ @After
+ public void after() throws Exception {
+ cleanupTestMetadata();
+ }
+
+ @Test
+ public void test() throws Exception {
+ assertNotNull(service);
+ BaseTestExecutable executable = new SucceedTestExecutable();
+ executable.setParam("test1", "test1");
+ executable.setParam("test2", "test2");
+ executable.setParam("test3", "test3");
+ service.addJob(executable);
+ List<AbstractExecutable> result = service.getAllExecutables();
+ assertEquals(1, result.size());
+ AbstractExecutable another = service.getJob(executable.getId());
+ assertJobEqual(executable, another);
+
+ service.updateJobOutput(executable.getId(), ExecutableState.RUNNING, null, "test output");
+ assertJobEqual(executable, service.getJob(executable.getId()));
+ }
+
+ @Test
+ public void testDefaultChainedExecutable() throws Exception {
+ DefaultChainedExecutable job = new DefaultChainedExecutable();
+ job.addTask(new SucceedTestExecutable());
+ job.addTask(new SucceedTestExecutable());
+
+ service.addJob(job);
+ assertEquals(2, job.getTasks().size());
+ AbstractExecutable anotherJob = service.getJob(job.getId());
+ assertEquals(DefaultChainedExecutable.class, anotherJob.getClass());
+ assertEquals(2, ((DefaultChainedExecutable) anotherJob).getTasks().size());
+ assertJobEqual(job, anotherJob);
+ }
+
+ @Test
+ public void testValidStateTransfer() throws Exception {
+ SucceedTestExecutable job = new SucceedTestExecutable();
+ String id = job.getId();
+ service.addJob(job);
+ service.updateJobOutput(id, ExecutableState.RUNNING, null, null);
+ service.updateJobOutput(id, ExecutableState.ERROR, null, null);
+ service.updateJobOutput(id, ExecutableState.READY, null, null);
+ service.updateJobOutput(id, ExecutableState.RUNNING, null, null);
+ service.updateJobOutput(id, ExecutableState.READY, null, null);
+ service.updateJobOutput(id, ExecutableState.RUNNING, null, null);
+ service.updateJobOutput(id, ExecutableState.SUCCEED, null, null);
+ }
+
+ @Test(expected = IllegalStateTranferException.class)
+ public void testInvalidStateTransfer(){
+ SucceedTestExecutable job = new SucceedTestExecutable();
+ service.addJob(job);
+ service.updateJobOutput(job.getId(), ExecutableState.RUNNING, null, null);
+ service.updateJobOutput(job.getId(), ExecutableState.STOPPED, null, null);
+ }
+
+
+
+ private static void assertJobEqual(Executable one, Executable another) {
+ assertEquals(one.getClass(), another.getClass());
+ assertEquals(one.getId(), another.getId());
+ assertEquals(one.getStatus(), another.getStatus());
+ assertEquals(one.isRunnable(), another.isRunnable());
+ assertEquals(one.getOutput(), another.getOutput());
+ assertTrue((one.getParams() == null && another.getParams() == null) || (one.getParams() != null && another.getParams() != null));
+ if (one.getParams() != null) {
+ assertEquals(one.getParams().size(), another.getParams().size());
+ for (String key : one.getParams().keySet()) {
+ assertEquals(one.getParams().get(key), another.getParams().get(key));
+ }
+ }
+ if (one instanceof ChainedExecutable) {
+ assertTrue(another instanceof ChainedExecutable);
+ List<? extends Executable> onesSubs = ((ChainedExecutable) one).getTasks();
+ List<? extends Executable> anotherSubs = ((ChainedExecutable) another).getTasks();
+ assertTrue((onesSubs == null && anotherSubs == null) || (onesSubs != null && anotherSubs != null));
+ if (onesSubs != null) {
+ assertEquals(onesSubs.size(), anotherSubs.size());
+ for (int i = 0; i < onesSubs.size(); ++i) {
+ assertJobEqual(onesSubs.get(i), anotherSubs.get(i));
+ }
+ }
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/ccfecdb3/job/src/test/java/com/kylinolap/job/service/ExecutableManagerTest.java
----------------------------------------------------------------------
diff --git a/job/src/test/java/com/kylinolap/job/service/ExecutableManagerTest.java b/job/src/test/java/com/kylinolap/job/service/ExecutableManagerTest.java
deleted file mode 100644
index 3d3cfce..0000000
--- a/job/src/test/java/com/kylinolap/job/service/ExecutableManagerTest.java
+++ /dev/null
@@ -1,126 +0,0 @@
-package com.kylinolap.job.service;
-
-import com.kylinolap.common.KylinConfig;
-import com.kylinolap.common.util.LocalFileMetadataTestCase;
-import com.kylinolap.job.BaseTestExecutable;
-import com.kylinolap.job.SucceedTestExecutable;
-import com.kylinolap.job.exception.IllegalStateTranferException;
-import com.kylinolap.job.execution.ChainedExecutable;
-import com.kylinolap.job.execution.Executable;
-import com.kylinolap.job.execution.ExecutableState;
-import com.kylinolap.job.impl.threadpool.AbstractExecutable;
-import com.kylinolap.job.impl.threadpool.DefaultChainedExecutable;
-import org.junit.After;
-import org.junit.Before;
-import org.junit.Test;
-
-import java.util.List;
-
-import static org.junit.Assert.*;
-
-/**
- * Created by qianzhou on 12/16/14.
- */
-public class ExecutableManagerTest extends LocalFileMetadataTestCase {
-
- private ExecutableManager service;
-
- @Before
- public void setup() throws Exception {
- createTestMetadata();
- service = ExecutableManager.getInstance(KylinConfig.getInstanceFromEnv());
-
- for (String jobId: service.getAllJobIds()) {
- System.out.println("deleting " + jobId);
- service.deleteJob(jobId);
- }
-
- }
-
- @After
- public void after() throws Exception {
- cleanupTestMetadata();
- }
-
- @Test
- public void test() throws Exception {
- assertNotNull(service);
- BaseTestExecutable executable = new SucceedTestExecutable();
- executable.setParam("test1", "test1");
- executable.setParam("test2", "test2");
- executable.setParam("test3", "test3");
- service.addJob(executable);
- List<AbstractExecutable> result = service.getAllExecutables();
- assertEquals(1, result.size());
- AbstractExecutable another = service.getJob(executable.getId());
- assertJobEqual(executable, another);
-
- service.updateJobOutput(executable.getId(), ExecutableState.RUNNING, null, "test output");
- assertJobEqual(executable, service.getJob(executable.getId()));
- }
-
- @Test
- public void testDefaultChainedExecutable() throws Exception {
- DefaultChainedExecutable job = new DefaultChainedExecutable();
- job.addTask(new SucceedTestExecutable());
- job.addTask(new SucceedTestExecutable());
-
- service.addJob(job);
- assertEquals(2, job.getTasks().size());
- AbstractExecutable anotherJob = service.getJob(job.getId());
- assertEquals(DefaultChainedExecutable.class, anotherJob.getClass());
- assertEquals(2, ((DefaultChainedExecutable) anotherJob).getTasks().size());
- assertJobEqual(job, anotherJob);
- }
-
- @Test
- public void testValidStateTransfer() throws Exception {
- SucceedTestExecutable job = new SucceedTestExecutable();
- String id = job.getId();
- service.addJob(job);
- service.updateJobOutput(id, ExecutableState.RUNNING, null, null);
- service.updateJobOutput(id, ExecutableState.ERROR, null, null);
- service.updateJobOutput(id, ExecutableState.READY, null, null);
- service.updateJobOutput(id, ExecutableState.RUNNING, null, null);
- service.updateJobOutput(id, ExecutableState.READY, null, null);
- service.updateJobOutput(id, ExecutableState.RUNNING, null, null);
- service.updateJobOutput(id, ExecutableState.SUCCEED, null, null);
- }
-
- @Test(expected = IllegalStateTranferException.class)
- public void testInvalidStateTransfer(){
- SucceedTestExecutable job = new SucceedTestExecutable();
- service.addJob(job);
- service.updateJobOutput(job.getId(), ExecutableState.RUNNING, null, null);
- service.updateJobOutput(job.getId(), ExecutableState.STOPPED, null, null);
- }
-
-
-
- private static void assertJobEqual(Executable one, Executable another) {
- assertEquals(one.getClass(), another.getClass());
- assertEquals(one.getId(), another.getId());
- assertEquals(one.getStatus(), another.getStatus());
- assertEquals(one.isRunnable(), another.isRunnable());
- assertEquals(one.getOutput(), another.getOutput());
- assertTrue((one.getParams() == null && another.getParams() == null) || (one.getParams() != null && another.getParams() != null));
- if (one.getParams() != null) {
- assertEquals(one.getParams().size(), another.getParams().size());
- for (String key : one.getParams().keySet()) {
- assertEquals(one.getParams().get(key), another.getParams().get(key));
- }
- }
- if (one instanceof ChainedExecutable) {
- assertTrue(another instanceof ChainedExecutable);
- List<? extends Executable> onesSubs = ((ChainedExecutable) one).getTasks();
- List<? extends Executable> anotherSubs = ((ChainedExecutable) another).getTasks();
- assertTrue((onesSubs == null && anotherSubs == null) || (onesSubs != null && anotherSubs != null));
- if (onesSubs != null) {
- assertEquals(onesSubs.size(), anotherSubs.size());
- for (int i = 0; i < onesSubs.size(); ++i) {
- assertJobEqual(onesSubs.get(i), anotherSubs.get(i));
- }
- }
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/ccfecdb3/server/src/main/java/com/kylinolap/rest/service/BasicService.java
----------------------------------------------------------------------
diff --git a/server/src/main/java/com/kylinolap/rest/service/BasicService.java b/server/src/main/java/com/kylinolap/rest/service/BasicService.java
index c8f259d..25edc92 100644
--- a/server/src/main/java/com/kylinolap/rest/service/BasicService.java
+++ b/server/src/main/java/com/kylinolap/rest/service/BasicService.java
@@ -27,8 +27,8 @@ import com.kylinolap.invertedindex.IIDescManager;
import com.kylinolap.invertedindex.IIManager;
import com.kylinolap.job.cube.CubingJob;
import com.kylinolap.job.execution.ExecutableState;
-import com.kylinolap.job.impl.threadpool.AbstractExecutable;
-import com.kylinolap.job.service.ExecutableManager;
+import com.kylinolap.job.execution.AbstractExecutable;
+import com.kylinolap.job.manager.ExecutableManager;
import com.kylinolap.metadata.MetadataManager;
import com.kylinolap.metadata.project.ProjectInstance;
import com.kylinolap.metadata.project.ProjectManager;
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/ccfecdb3/server/src/main/java/com/kylinolap/rest/service/CubeService.java
----------------------------------------------------------------------
diff --git a/server/src/main/java/com/kylinolap/rest/service/CubeService.java b/server/src/main/java/com/kylinolap/rest/service/CubeService.java
index 45987fa..f6d76b4 100644
--- a/server/src/main/java/com/kylinolap/rest/service/CubeService.java
+++ b/server/src/main/java/com/kylinolap/rest/service/CubeService.java
@@ -55,7 +55,7 @@ import com.kylinolap.job.exception.JobException;
import com.kylinolap.job.execution.ExecutableState;
import com.kylinolap.job.hadoop.cardinality.HiveColumnCardinalityJob;
import com.kylinolap.job.hadoop.cardinality.HiveColumnCardinalityUpdateJob;
-import com.kylinolap.job.impl.threadpool.DefaultChainedExecutable;
+import com.kylinolap.job.execution.DefaultChainedExecutable;
import com.kylinolap.metadata.MetadataConstances;
import com.kylinolap.metadata.MetadataManager;
import com.kylinolap.metadata.model.SegmentStatusEnum;
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/ccfecdb3/server/src/main/java/com/kylinolap/rest/service/JobService.java
----------------------------------------------------------------------
diff --git a/server/src/main/java/com/kylinolap/rest/service/JobService.java b/server/src/main/java/com/kylinolap/rest/service/JobService.java
index ed547ff..f151136 100644
--- a/server/src/main/java/com/kylinolap/rest/service/JobService.java
+++ b/server/src/main/java/com/kylinolap/rest/service/JobService.java
@@ -48,7 +48,7 @@ import com.kylinolap.job.engine.JobEngineConfig;
import com.kylinolap.job.exception.JobException;
import com.kylinolap.job.execution.ExecutableState;
import com.kylinolap.job.execution.Output;
-import com.kylinolap.job.impl.threadpool.AbstractExecutable;
+import com.kylinolap.job.execution.AbstractExecutable;
import com.kylinolap.metadata.model.SegmentStatusEnum;
import com.kylinolap.rest.constant.Constant;
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/ccfecdb3/server/src/test/java/com/kylinolap/rest/controller/JobControllerTest.java
----------------------------------------------------------------------
diff --git a/server/src/test/java/com/kylinolap/rest/controller/JobControllerTest.java b/server/src/test/java/com/kylinolap/rest/controller/JobControllerTest.java
index e13491f..7fbc46d 100644
--- a/server/src/test/java/com/kylinolap/rest/controller/JobControllerTest.java
+++ b/server/src/test/java/com/kylinolap/rest/controller/JobControllerTest.java
@@ -33,7 +33,7 @@ import com.kylinolap.cube.CubeInstance;
import com.kylinolap.cube.CubeManager;
import com.kylinolap.cube.model.CubeDesc;
import com.kylinolap.job.JobInstance;
-import com.kylinolap.job.dao.JobDao;
+import com.kylinolap.job.dao.ExecutableDao;
import com.kylinolap.job.exception.PersistentException;
import com.kylinolap.rest.request.JobBuildRequest;
import com.kylinolap.rest.request.JobListRequest;
@@ -57,7 +57,7 @@ public class JobControllerTest extends ServiceTestBase {
private CubeManager cubeManager;
private CubeDescManager cubeDescManager;
- private JobDao jobDAO;
+ private ExecutableDao executableDAO;
@Before
public void setup() throws Exception {
@@ -72,7 +72,7 @@ public class JobControllerTest extends ServiceTestBase {
cubeManager = CubeManager.getInstance(testConfig);
cubeDescManager = CubeDescManager.getInstance(testConfig);
- jobDAO = JobDao.getInstance(testConfig);
+ executableDAO = ExecutableDao.getInstance(testConfig);
}
@@ -99,7 +99,7 @@ public class JobControllerTest extends ServiceTestBase {
JobInstance job = cubeController.rebuild(CUBE_NAME, jobBuildRequest);
Assert.assertNotNull(jobSchedulerController.get(job.getId()));
- jobDAO.deleteJob(job.getId());
+ executableDAO.deleteJob(job.getId());
if (cubeManager.getCube(CUBE_NAME) != null) {
cubeManager.dropCube(CUBE_NAME, false);
}