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/07/23 13:21:56 UTC
[19/23] incubator-kylin git commit: KYLIN-875 half way
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/7663fff4/core-job/src/main/java/org/apache/kylin/job/common/ShellExecutable.java
----------------------------------------------------------------------
diff --git a/core-job/src/main/java/org/apache/kylin/job/common/ShellExecutable.java b/core-job/src/main/java/org/apache/kylin/job/common/ShellExecutable.java
new file mode 100644
index 0000000..786698e
--- /dev/null
+++ b/core-job/src/main/java/org/apache/kylin/job/common/ShellExecutable.java
@@ -0,0 +1,143 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 org.apache.kylin.job.common;
+
+import java.io.IOException;
+import java.util.Map;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+
+import org.apache.kylin.common.util.Pair;
+
+import com.google.common.collect.Maps;
+import org.apache.kylin.common.util.Logger;
+import org.apache.kylin.job.constant.ExecutableConstants;
+import org.apache.kylin.job.exception.ExecuteException;
+import org.apache.kylin.job.execution.AbstractExecutable;
+import org.apache.kylin.job.execution.ExecutableContext;
+import org.apache.kylin.job.execution.ExecuteResult;
+
+/**
+ */
+public class ShellExecutable extends AbstractExecutable {
+
+ private static final String CMD = "cmd";
+
+ public ShellExecutable() {
+ super();
+ }
+
+ @Override
+ protected ExecuteResult doWork(ExecutableContext context) throws ExecuteException {
+ try {
+ logger.info("executing:" + getCmd());
+ final ShellExecutableLogger logger = new ShellExecutableLogger();
+ final Pair<Integer, String> result = context.getConfig().getCliCommandExecutor().execute(getCmd(), logger);
+ executableManager.addJobInfo(getId(), logger.getInfo());
+ return new ExecuteResult(result.getFirst() == 0? ExecuteResult.State.SUCCEED: ExecuteResult.State.FAILED, result.getSecond());
+ } catch (IOException e) {
+ logger.error("job:" + getId() + " execute finished with exception", e);
+ return new ExecuteResult(ExecuteResult.State.ERROR, e.getLocalizedMessage());
+ }
+ }
+
+ public void setCmd(String cmd) {
+ setParam(CMD, cmd);
+ }
+
+ public String getCmd() {
+ return getParam(CMD);
+ }
+
+ private static class ShellExecutableLogger implements Logger {
+
+ private final Map<String, String> info = Maps.newHashMap();
+
+ private static final Pattern PATTERN_APP_ID = Pattern.compile("Submitted application (.*?) to ResourceManager");
+ private static final Pattern PATTERN_APP_URL = Pattern.compile("The url to track the job: (.*)");
+ private static final Pattern PATTERN_JOB_ID = Pattern.compile("Running job: (.*)");
+ private static final Pattern PATTERN_HDFS_BYTES_WRITTEN = Pattern.compile("(?:HD|MAPR)FS: Number of bytes written=(\\d+)");
+ private static final Pattern PATTERN_SOURCE_RECORDS_COUNT = Pattern.compile("Map input records=(\\d+)");
+ private static final Pattern PATTERN_SOURCE_RECORDS_SIZE = Pattern.compile("(?:HD|MAPR)FS Read: (\\d+) HDFS Write");
+
+ // hive
+ private static final Pattern PATTERN_HIVE_APP_ID_URL = Pattern.compile("Starting Job = (.*?), Tracking URL = (.*)");
+ private static final Pattern PATTERN_HIVE_BYTES_WRITTEN = Pattern.compile("(?:HD|MAPR)FS Read: (\\d+) HDFS Write: (\\d+) SUCCESS");
+
+ @Override
+ public void log(String message) {
+ Matcher matcher = PATTERN_APP_ID.matcher(message);
+ if (matcher.find()) {
+ String appId = matcher.group(1);
+ info.put(ExecutableConstants.YARN_APP_ID, appId);
+ }
+
+ matcher = PATTERN_APP_URL.matcher(message);
+ if (matcher.find()) {
+ String appTrackingUrl = matcher.group(1);
+ info.put(ExecutableConstants.YARN_APP_URL, appTrackingUrl);
+ }
+
+ matcher = PATTERN_JOB_ID.matcher(message);
+ if (matcher.find()) {
+ String mrJobID = matcher.group(1);
+ info.put(ExecutableConstants.MR_JOB_ID, mrJobID);
+ }
+
+ matcher = PATTERN_HDFS_BYTES_WRITTEN.matcher(message);
+ if (matcher.find()) {
+ String hdfsWritten = matcher.group(1);
+ info.put(ExecutableConstants.HDFS_BYTES_WRITTEN, hdfsWritten);
+ }
+
+ matcher = PATTERN_SOURCE_RECORDS_COUNT.matcher(message);
+ if (matcher.find()) {
+ String sourceCount = matcher.group(1);
+ info.put(ExecutableConstants.SOURCE_RECORDS_COUNT, sourceCount);
+ }
+
+ matcher = PATTERN_SOURCE_RECORDS_SIZE.matcher(message);
+ if (matcher.find()) {
+ String sourceSize = matcher.group(1);
+ info.put(ExecutableConstants.SOURCE_RECORDS_SIZE, sourceSize);
+ }
+
+ // hive
+ matcher = PATTERN_HIVE_APP_ID_URL.matcher(message);
+ if (matcher.find()) {
+ String jobId = matcher.group(1);
+ String trackingUrl = matcher.group(2);
+ info.put(ExecutableConstants.MR_JOB_ID, jobId);
+ info.put(ExecutableConstants.YARN_APP_URL, trackingUrl);
+ }
+
+ matcher = PATTERN_HIVE_BYTES_WRITTEN.matcher(message);
+ if (matcher.find()) {
+ // String hdfsRead = matcher.group(1);
+ String hdfsWritten = matcher.group(2);
+ info.put(ExecutableConstants.HDFS_BYTES_WRITTEN, hdfsWritten);
+ }
+ }
+
+ Map<String, String> getInfo() {
+ return info;
+ }
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/7663fff4/core-job/src/main/java/org/apache/kylin/job/constant/ExecutableConstants.java
----------------------------------------------------------------------
diff --git a/core-job/src/main/java/org/apache/kylin/job/constant/ExecutableConstants.java b/core-job/src/main/java/org/apache/kylin/job/constant/ExecutableConstants.java
new file mode 100644
index 0000000..fdcfdbe
--- /dev/null
+++ b/core-job/src/main/java/org/apache/kylin/job/constant/ExecutableConstants.java
@@ -0,0 +1,80 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 org.apache.kylin.job.constant;
+
+/**
+ */
+public final class ExecutableConstants {
+
+ private ExecutableConstants(){}
+
+ public static final String YARN_APP_ID = "yarn_application_id";
+
+ public static final String YARN_APP_URL = "yarn_application_tracking_url";
+ public static final String MR_JOB_ID = "mr_job_id";
+ public static final String HDFS_BYTES_WRITTEN = "hdfs_bytes_written";
+ public static final String SOURCE_RECORDS_COUNT = "source_records_count";
+ public static final String SOURCE_RECORDS_SIZE = "source_records_size";
+ public static final String GLOBAL_LISTENER_NAME = "ChainListener";
+
+
+
+
+ public static final int DEFAULT_SCHEDULER_INTERVAL_SECONDS = 60;
+
+ public static final String CUBE_JOB_GROUP_NAME = "cube_job_group";
+
+ public static final String DAEMON_JOB_GROUP_NAME = "daemon_job_group";
+ public static final String STEP_NAME_BUILD_DICTIONARY = "Build Dimension Dictionary";
+
+ public static final String STEP_NAME_CREATE_FLAT_HIVE_TABLE = "Create Intermediate Flat Hive Table";
+ public static final String STEP_NAME_FACT_DISTINCT_COLUMNS = "Extract Fact Table Distinct Columns";
+ public static final String STEP_NAME_BUILD_BASE_CUBOID = "Build Base Cuboid Data";
+ public static final String STEP_NAME_BUILD_IN_MEM_CUBE = "Build Cube";
+ public static final String STEP_NAME_BUILD_N_D_CUBOID = "Build N-Dimension Cuboid Data";
+ public static final String STEP_NAME_GET_CUBOID_KEY_DISTRIBUTION = "Calculate HTable Region Splits";
+ public static final String STEP_NAME_CREATE_HBASE_TABLE = "Create HTable";
+ public static final String STEP_NAME_CONVERT_CUBOID_TO_HFILE = "Convert Cuboid Data to HFile";
+ public static final String STEP_NAME_BULK_LOAD_HFILE = "Load HFile to HBase Table";
+ public static final String STEP_NAME_MERGE_DICTIONARY = "Merge Cuboid Dictionary";
+ public static final String STEP_NAME_MERGE_STATISTICS = "Merge Cuboid Statistics";
+ public static final String STEP_NAME_SAVE_STATISTICS = "Save Cuboid Statistics";
+ public static final String STEP_NAME_MERGE_CUBOID = "Merge Cuboid Data";
+ public static final String STEP_NAME_UPDATE_CUBE_INFO = "Update Cube Info";
+ public static final String STEP_NAME_GARBAGE_COLLECTION = "Garbage Collection";
+
+ public static final String STEP_NAME_BUILD_II = "Build Inverted Index";
+ public static final String STEP_NAME_CONVERT_II_TO_HFILE = "Convert Inverted Index Data to HFile";
+
+ public static final String PROP_ENGINE_CONTEXT = "jobengineConfig";
+ public static final String PROP_JOB_FLOW = "jobFlow";
+ public static final String PROP_JOBINSTANCE_UUID = "jobInstanceUuid";
+ public static final String PROP_JOBSTEP_SEQ_ID = "jobStepSequenceID";
+ public static final String PROP_COMMAND = "command";
+ // public static final String PROP_STORAGE_LOCATION =
+ // "storageLocationIdentifier";
+ public static final String PROP_JOB_ASYNC = "jobAsync";
+ public static final String PROP_JOB_CMD_EXECUTOR = "jobCmdExecutor";
+ public static final String PROP_JOB_CMD_OUTPUT = "jobCmdOutput";
+ public static final String PROP_JOB_KILLED = "jobKilled";
+ public static final String PROP_JOB_RUNTIME_FLOWS = "jobFlows";
+
+ public static final String NOTIFY_EMAIL_TEMPLATE = "<div><b>Build Result of Job ${job_name}</b><pre><ul>" + "<li>Build Result: <b>${result}</b></li>" + "<li>Job Engine: ${job_engine}</li>" + "<li>Cube Name: ${cube_name}</li>" + "<li>Start Time: ${start_time}</li>" + "<li>Duration: ${duration}</li>" + "<li>MR Waiting: ${mr_waiting}</li>" + "<li>Last Update Time: ${last_update_time}</li>" + "<li>Submitter: ${submitter}</li>" + "<li>Error Log: ${error_log}</li>" + "</ul></pre><div/>";
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/7663fff4/core-job/src/main/java/org/apache/kylin/job/constant/JobStatusEnum.java
----------------------------------------------------------------------
diff --git a/core-job/src/main/java/org/apache/kylin/job/constant/JobStatusEnum.java b/core-job/src/main/java/org/apache/kylin/job/constant/JobStatusEnum.java
new file mode 100644
index 0000000..a4ef564
--- /dev/null
+++ b/core-job/src/main/java/org/apache/kylin/job/constant/JobStatusEnum.java
@@ -0,0 +1,49 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 org.apache.kylin.job.constant;
+
+public enum JobStatusEnum {
+
+ NEW(0), PENDING(1), RUNNING(2), FINISHED(4), ERROR(8), DISCARDED(16);
+
+ private final int code;
+
+ private JobStatusEnum(int statusCode) {
+ this.code = statusCode;
+ }
+
+ public static JobStatusEnum getByCode(int statusCode) {
+ for (JobStatusEnum status : values()) {
+ if (status.getCode() == statusCode) {
+ return status;
+ }
+ }
+
+ return null;
+ }
+
+ public int getCode() {
+ return this.code;
+ }
+
+ public boolean isComplete() {
+ return code == JobStatusEnum.FINISHED.getCode() || code == JobStatusEnum.ERROR.getCode() || code == JobStatusEnum.DISCARDED.getCode();
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/7663fff4/core-job/src/main/java/org/apache/kylin/job/constant/JobStepCmdTypeEnum.java
----------------------------------------------------------------------
diff --git a/core-job/src/main/java/org/apache/kylin/job/constant/JobStepCmdTypeEnum.java b/core-job/src/main/java/org/apache/kylin/job/constant/JobStepCmdTypeEnum.java
new file mode 100644
index 0000000..02b40a3
--- /dev/null
+++ b/core-job/src/main/java/org/apache/kylin/job/constant/JobStepCmdTypeEnum.java
@@ -0,0 +1,27 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 org.apache.kylin.job.constant;
+
+/**
+ * @author xduo, ysong1
+ *
+ */
+public enum JobStepCmdTypeEnum {
+ SHELL_CMD, SHELL_CMD_HADOOP, JAVA_CMD_HADOOP_FACTDISTINCT, JAVA_CMD_HADOOP_BASECUBOID, JAVA_CMD_HADOOP_NDCUBOID, JAVA_CMD_HADOOP_RANGEKEYDISTRIBUTION, JAVA_CMD_HADOOP_CONVERTHFILE, JAVA_CMD_HADOOP_MERGECUBOID, JAVA_CMD_HADOOP_NO_MR_DICTIONARY, JAVA_CMD_HADDOP_NO_MR_CREATEHTABLE, JAVA_CMD_HADOOP_NO_MR_BULKLOAD
+}
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/7663fff4/core-job/src/main/java/org/apache/kylin/job/constant/JobStepStatusEnum.java
----------------------------------------------------------------------
diff --git a/core-job/src/main/java/org/apache/kylin/job/constant/JobStepStatusEnum.java b/core-job/src/main/java/org/apache/kylin/job/constant/JobStepStatusEnum.java
new file mode 100644
index 0000000..08ee79a
--- /dev/null
+++ b/core-job/src/main/java/org/apache/kylin/job/constant/JobStepStatusEnum.java
@@ -0,0 +1,51 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 org.apache.kylin.job.constant;
+
+public enum JobStepStatusEnum {
+ NEW(0), PENDING(1), RUNNING(2), FINISHED(4), ERROR(8), DISCARDED(16), WAITING(32), KILLED(64);
+
+ private final int code;
+
+ private JobStepStatusEnum(int statusCode) {
+ this.code = statusCode;
+ }
+
+ public static JobStepStatusEnum getByCode(int statusCode) {
+ for (JobStepStatusEnum status : values()) {
+ if (status.getCode() == statusCode) {
+ return status;
+ }
+ }
+
+ return null;
+ }
+
+ public int getCode() {
+ return this.code;
+ }
+
+ public boolean isComplete() {
+ return code == JobStepStatusEnum.FINISHED.getCode() || code == JobStepStatusEnum.ERROR.getCode() || code == JobStepStatusEnum.DISCARDED.getCode();
+ }
+
+ public boolean isRunable() {
+ return code == JobStepStatusEnum.PENDING.getCode() || code == JobStepStatusEnum.ERROR.getCode();
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/7663fff4/core-job/src/main/java/org/apache/kylin/job/dao/ExecutableDao.java
----------------------------------------------------------------------
diff --git a/core-job/src/main/java/org/apache/kylin/job/dao/ExecutableDao.java b/core-job/src/main/java/org/apache/kylin/job/dao/ExecutableDao.java
new file mode 100644
index 0000000..4862bb1
--- /dev/null
+++ b/core-job/src/main/java/org/apache/kylin/job/dao/ExecutableDao.java
@@ -0,0 +1,221 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 org.apache.kylin.job.dao;
+
+import com.google.common.collect.Lists;
+import org.apache.commons.io.IOUtils;
+import org.apache.kylin.common.KylinConfig;
+import org.apache.kylin.common.persistence.JsonSerializer;
+import org.apache.kylin.common.persistence.RawResource;
+import org.apache.kylin.common.persistence.ResourceStore;
+import org.apache.kylin.common.persistence.Serializer;
+import org.apache.kylin.job.exception.PersistentException;
+import org.apache.kylin.metadata.MetadataManager;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.DataInputStream;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.concurrent.ConcurrentHashMap;
+
+/**
+ */
+public class ExecutableDao {
+
+ private static final Serializer<ExecutablePO> JOB_SERIALIZER = new JsonSerializer<ExecutablePO>(ExecutablePO.class);
+ private static final Serializer<ExecutableOutputPO> JOB_OUTPUT_SERIALIZER = new JsonSerializer<ExecutableOutputPO>(ExecutableOutputPO.class);
+ private static final Logger logger = LoggerFactory.getLogger(ExecutableDao.class);
+ private static final ConcurrentHashMap<KylinConfig, ExecutableDao> CACHE = new ConcurrentHashMap<KylinConfig, ExecutableDao>();
+ public static final String JOB_PATH_ROOT = "/execute";
+ public static final String JOB_OUTPUT_ROOT = "/execute_output";
+
+ private ResourceStore store;
+
+ public static ExecutableDao getInstance(KylinConfig config) {
+ ExecutableDao r = CACHE.get(config);
+ if (r == null) {
+ r = new ExecutableDao(config);
+ CACHE.put(config, r);
+ if (CACHE.size() > 1) {
+ logger.warn("More than one singleton exist");
+ }
+
+ }
+ return r;
+ }
+
+ private ExecutableDao(KylinConfig config) {
+ logger.info("Using metadata url: " + config);
+ this.store = MetadataManager.getInstance(config).getStore();
+ }
+
+ private String pathOfJob(ExecutablePO job) {
+ return pathOfJob(job.getUuid());
+ }
+ private String pathOfJob(String uuid) {
+ return JOB_PATH_ROOT + "/" + uuid;
+ }
+
+ private String pathOfJobOutput(String uuid) {
+ return JOB_OUTPUT_ROOT + "/" + uuid;
+ }
+
+ private ExecutablePO readJobResource(String path) throws IOException {
+ return store.getResource(path, ExecutablePO.class, JOB_SERIALIZER);
+ }
+
+ private void writeJobResource(String path, ExecutablePO job) throws IOException {
+ store.putResource(path, job, JOB_SERIALIZER);
+ }
+
+ private ExecutableOutputPO readJobOutputResource(String path) throws IOException {
+ return store.getResource(path, ExecutableOutputPO.class, JOB_OUTPUT_SERIALIZER);
+ }
+
+ private long writeJobOutputResource(String path, ExecutableOutputPO output) throws IOException {
+ return store.putResource(path, output, JOB_OUTPUT_SERIALIZER);
+ }
+
+ public List<ExecutableOutputPO> getJobOutputs() throws PersistentException {
+ try {
+ ArrayList<String> resources = store.listResources(JOB_OUTPUT_ROOT);
+ if (resources == null || resources.isEmpty()) {
+ return Collections.emptyList();
+ }
+ Collections.sort(resources);
+ String rangeStart = resources.get(0);
+ String rangeEnd = resources.get(resources.size() - 1);
+ return store.getAllResources(rangeStart, rangeEnd, ExecutableOutputPO.class, JOB_OUTPUT_SERIALIZER);
+ } catch (IOException e) {
+ logger.error("error get all Jobs:", e);
+ throw new PersistentException(e);
+ }
+ }
+
+ public List<ExecutablePO> getJobs() throws PersistentException {
+ try {
+ final List<String> jobIds = store.listResources(JOB_PATH_ROOT);
+ if (jobIds == null || jobIds.isEmpty()) {
+ return Collections.emptyList();
+ }
+ Collections.sort(jobIds);
+ String rangeStart = jobIds.get(0);
+ String rangeEnd = jobIds.get(jobIds.size() - 1);
+ return store.getAllResources(rangeStart, rangeEnd, ExecutablePO.class, JOB_SERIALIZER);
+ } catch (IOException e) {
+ logger.error("error get all Jobs:", e);
+ throw new PersistentException(e);
+ }
+ }
+
+ public List<String> getJobIds() throws PersistentException {
+ try {
+ ArrayList<String> resources = store.listResources(JOB_PATH_ROOT);
+ if (resources == null) {
+ return Collections.emptyList();
+ }
+ ArrayList<String> result = Lists.newArrayListWithExpectedSize(resources.size());
+ for (String path : resources) {
+ result.add(path.substring(path.lastIndexOf("/") + 1));
+ }
+ return result;
+ } catch (IOException e) {
+ logger.error("error get all Jobs:", e);
+ throw new PersistentException(e);
+ }
+ }
+
+ public ExecutablePO getJob(String uuid) throws PersistentException {
+ try {
+ return readJobResource(pathOfJob(uuid));
+ } catch (IOException e) {
+ logger.error("error get job:" + uuid, e);
+ throw new PersistentException(e);
+ }
+ }
+
+ public ExecutablePO addJob(ExecutablePO job) throws PersistentException {
+ try {
+ if (getJob(job.getUuid()) != null) {
+ throw new IllegalArgumentException("job id:" + job.getUuid() + " already exists");
+ }
+ writeJobResource(pathOfJob(job), job);
+ return job;
+ } catch (IOException e) {
+ logger.error("error save job:" + job.getUuid(), e);
+ throw new PersistentException(e);
+ }
+ }
+
+ public void deleteJob(String uuid) throws PersistentException {
+ try {
+ store.deleteResource(pathOfJob(uuid));
+ } catch (IOException e) {
+ logger.error("error delete job:" + uuid, e);
+ throw new PersistentException(e);
+ }
+ }
+
+ public ExecutableOutputPO getJobOutput(String uuid) throws PersistentException {
+ try {
+ ExecutableOutputPO result = readJobOutputResource(pathOfJobOutput(uuid));
+ if (result == null) {
+ result = new ExecutableOutputPO();
+ result.setUuid(uuid);
+ return result;
+ }
+ return result;
+ } catch (IOException e) {
+ logger.error("error get job output id:" + uuid, e);
+ throw new PersistentException(e);
+ }
+ }
+
+ public void addJobOutput(ExecutableOutputPO output) throws PersistentException {
+ try {
+ output.setLastModified(0);
+ writeJobOutputResource(pathOfJobOutput(output.getUuid()), output);
+ } catch (IOException e) {
+ logger.error("error update job output id:" + output.getUuid(), e);
+ throw new PersistentException(e);
+ }
+ }
+
+ public void updateJobOutput(ExecutableOutputPO output) throws PersistentException {
+ try {
+ final long ts = writeJobOutputResource(pathOfJobOutput(output.getUuid()), output);
+ output.setLastModified(ts);
+ } catch (IOException e) {
+ logger.error("error update job output id:" + output.getUuid(), e);
+ throw new PersistentException(e);
+ }
+ }
+
+ public void deleteJobOutput(String uuid) throws PersistentException {
+ try {
+ store.deleteResource(pathOfJobOutput(uuid));
+ } catch (IOException e) {
+ logger.error("error delete job:" + uuid, e);
+ throw new PersistentException(e);
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/7663fff4/core-job/src/main/java/org/apache/kylin/job/dao/ExecutableOutputPO.java
----------------------------------------------------------------------
diff --git a/core-job/src/main/java/org/apache/kylin/job/dao/ExecutableOutputPO.java b/core-job/src/main/java/org/apache/kylin/job/dao/ExecutableOutputPO.java
new file mode 100644
index 0000000..4dacd8a
--- /dev/null
+++ b/core-job/src/main/java/org/apache/kylin/job/dao/ExecutableOutputPO.java
@@ -0,0 +1,65 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 org.apache.kylin.job.dao;
+
+import java.util.Map;
+
+import com.fasterxml.jackson.annotation.JsonAutoDetect;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.google.common.collect.Maps;
+import org.apache.kylin.common.persistence.RootPersistentEntity;
+
+/**
+ */
+@JsonAutoDetect(fieldVisibility = JsonAutoDetect.Visibility.NONE, getterVisibility = JsonAutoDetect.Visibility.NONE, isGetterVisibility = JsonAutoDetect.Visibility.NONE, setterVisibility = JsonAutoDetect.Visibility.NONE)
+public class ExecutableOutputPO extends RootPersistentEntity {
+
+ @JsonProperty("content")
+ private String content;
+
+ @JsonProperty("status")
+ private String status = "READY";
+
+ @JsonProperty("info")
+ private Map<String, String> info = Maps.newHashMap();
+
+ public String getContent() {
+ return content;
+ }
+
+ public void setContent(String content) {
+ this.content = content;
+ }
+
+ public String getStatus() {
+ return status;
+ }
+
+ public void setStatus(String status) {
+ this.status = status;
+ }
+
+ public Map<String, String> getInfo() {
+ return info;
+ }
+
+ public void setInfo(Map<String, String> info) {
+ this.info = info;
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/7663fff4/core-job/src/main/java/org/apache/kylin/job/dao/ExecutablePO.java
----------------------------------------------------------------------
diff --git a/core-job/src/main/java/org/apache/kylin/job/dao/ExecutablePO.java b/core-job/src/main/java/org/apache/kylin/job/dao/ExecutablePO.java
new file mode 100644
index 0000000..6a17b29
--- /dev/null
+++ b/core-job/src/main/java/org/apache/kylin/job/dao/ExecutablePO.java
@@ -0,0 +1,78 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 org.apache.kylin.job.dao;
+
+import java.util.List;
+import java.util.Map;
+
+import com.fasterxml.jackson.annotation.JsonAutoDetect;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.google.common.collect.Maps;
+import org.apache.kylin.common.persistence.RootPersistentEntity;
+
+/**
+ */
+@JsonAutoDetect(fieldVisibility = JsonAutoDetect.Visibility.NONE, getterVisibility = JsonAutoDetect.Visibility.NONE, isGetterVisibility = JsonAutoDetect.Visibility.NONE, setterVisibility = JsonAutoDetect.Visibility.NONE)
+public class ExecutablePO extends RootPersistentEntity {
+
+ @JsonProperty("name")
+ private String name;
+
+ @JsonProperty("tasks")
+ private List<ExecutablePO> tasks;
+
+ @JsonProperty("type")
+ private String type;
+
+ @JsonProperty("params")
+ private Map<String, String> params = Maps.newHashMap();
+
+ public String getName() {
+ return name;
+ }
+
+ public void setName(String name) {
+ this.name = name;
+ }
+
+ public List<ExecutablePO> getTasks() {
+ return tasks;
+ }
+
+ public void setTasks(List<ExecutablePO> tasks) {
+ this.tasks = tasks;
+ }
+
+ public String getType() {
+ return type;
+ }
+
+ public void setType(String type) {
+ this.type = type;
+ }
+
+ public Map<String, String> getParams() {
+ return params;
+ }
+
+ public void setParams(Map<String, String> params) {
+ this.params = params;
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/7663fff4/core-job/src/main/java/org/apache/kylin/job/engine/JobEngineConfig.java
----------------------------------------------------------------------
diff --git a/core-job/src/main/java/org/apache/kylin/job/engine/JobEngineConfig.java b/core-job/src/main/java/org/apache/kylin/job/engine/JobEngineConfig.java
new file mode 100644
index 0000000..4d4d944
--- /dev/null
+++ b/core-job/src/main/java/org/apache/kylin/job/engine/JobEngineConfig.java
@@ -0,0 +1,185 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 org.apache.kylin.job.engine;
+
+import java.io.File;
+import java.io.IOException;
+
+import org.apache.commons.lang.StringUtils;
+import org.apache.kylin.common.KylinConfig;
+import org.apache.kylin.job.common.OptionsHelper;
+import org.apache.kylin.metadata.model.DataModelDesc.RealizationCapacity;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * @author ysong1
+ */
+public class JobEngineConfig {
+ private static final Logger logger = LoggerFactory.getLogger(JobEngineConfig.class);
+ public static String HADOOP_JOB_CONF_FILENAME = "kylin_job_conf";
+ public static String HIVE_CONF_FILENAME = "kylin_hive_conf";
+
+ private static File getJobConfig(String fileName) {
+ String path = System.getProperty(KylinConfig.KYLIN_CONF);
+ if (StringUtils.isNotEmpty(path)) {
+ return new File(path, fileName);
+ }
+
+ path = KylinConfig.getKylinHome();
+ if (StringUtils.isNotEmpty(path)) {
+ return new File(path + File.separator + "conf", fileName);
+ }
+ return null;
+ }
+
+ private String getHadoopJobConfFilePath(RealizationCapacity capaticy, boolean appendSuffix) throws IOException {
+ String hadoopJobConfFile;
+ if (appendSuffix) {
+ hadoopJobConfFile = (HADOOP_JOB_CONF_FILENAME + "_" + capaticy.toString().toLowerCase() + ".xml");
+ } else {
+ hadoopJobConfFile = (HADOOP_JOB_CONF_FILENAME + ".xml");
+ }
+
+ File jobConfig = getJobConfig(hadoopJobConfFile);
+ if (jobConfig == null || !jobConfig.exists()) {
+ logger.warn("fail to locate " + hadoopJobConfFile + ", trying to locate " + HADOOP_JOB_CONF_FILENAME + ".xml");
+ jobConfig = getJobConfig(HADOOP_JOB_CONF_FILENAME + ".xml");
+ if (jobConfig == null || !jobConfig.exists()) {
+ logger.error("fail to locate " + HADOOP_JOB_CONF_FILENAME + ".xml");
+ throw new RuntimeException("fail to locate " + HADOOP_JOB_CONF_FILENAME + ".xml");
+ }
+ }
+ return OptionsHelper.convertToFileURL(jobConfig.getAbsolutePath());
+ }
+
+ public String getHadoopJobConfFilePath(RealizationCapacity capaticy) throws IOException {
+ String path = getHadoopJobConfFilePath(capaticy, true);
+ if (!StringUtils.isEmpty(path)) {
+ logger.info("Chosen job conf is : " + path);
+ return path;
+ } else {
+ path = getHadoopJobConfFilePath(capaticy, false);
+ if (!StringUtils.isEmpty(path)) {
+ logger.info("Chosen job conf is : " + path);
+ return path;
+ }
+ }
+ return "";
+ }
+
+
+ public String getHiveConfFilePath() throws IOException {
+ String hiveConfFile = (HIVE_CONF_FILENAME + ".xml");
+
+ File jobConfig = getJobConfig(hiveConfFile);
+ if (jobConfig == null || !jobConfig.exists()) {
+
+ logger.error("fail to locate " + HIVE_CONF_FILENAME + ".xml");
+ throw new RuntimeException("fail to locate " + HIVE_CONF_FILENAME + ".xml");
+ }
+ return OptionsHelper.convertToFileURL(jobConfig.getAbsolutePath());
+ }
+
+ // there should be no setters
+ private final KylinConfig config;
+
+ public JobEngineConfig(KylinConfig config) {
+ this.config = config;
+ }
+
+ public KylinConfig getConfig() {
+ return config;
+ }
+
+ public String getHdfsWorkingDirectory() {
+ return config.getHdfsWorkingDirectory();
+ }
+
+ /**
+ * @return the maxConcurrentJobLimit
+ */
+ public int getMaxConcurrentJobLimit() {
+ return config.getMaxConcurrentJobLimit();
+ }
+
+ /**
+ * @return the timeZone
+ */
+ public String getTimeZone() {
+ return config.getTimeZone();
+ }
+
+ /**
+ * @return the adminDls
+ */
+ public String getAdminDls() {
+ return config.getAdminDls();
+ }
+
+ /**
+ * @return the jobStepTimeout
+ */
+ public long getJobStepTimeout() {
+ return config.getJobStepTimeout();
+ }
+
+ /**
+ * @return the asyncJobCheckInterval
+ */
+ public int getAsyncJobCheckInterval() {
+ return config.getYarnStatusCheckIntervalSeconds();
+ }
+
+ /*
+ * (non-Javadoc)
+ *
+ * @see java.lang.Object#hashCode()
+ */
+ @Override
+ public int hashCode() {
+ final int prime = 31;
+ int result = 1;
+ result = prime * result + ((config == null) ? 0 : config.hashCode());
+ return result;
+ }
+
+ /*
+ * (non-Javadoc)
+ *
+ * @see java.lang.Object#equals(java.lang.Object)
+ */
+ @Override
+ public boolean equals(Object obj) {
+ if (this == obj)
+ return true;
+ if (obj == null)
+ return false;
+ if (getClass() != obj.getClass())
+ return false;
+ JobEngineConfig other = (JobEngineConfig) obj;
+ if (config == null) {
+ if (other.config != null)
+ return false;
+ } else if (!config.equals(other.config))
+ return false;
+ return true;
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/7663fff4/core-job/src/main/java/org/apache/kylin/job/exception/ExecuteException.java
----------------------------------------------------------------------
diff --git a/core-job/src/main/java/org/apache/kylin/job/exception/ExecuteException.java b/core-job/src/main/java/org/apache/kylin/job/exception/ExecuteException.java
new file mode 100644
index 0000000..8544fff
--- /dev/null
+++ b/core-job/src/main/java/org/apache/kylin/job/exception/ExecuteException.java
@@ -0,0 +1,41 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 org.apache.kylin.job.exception;
+
+/**
+ */
+public class ExecuteException extends Exception {
+
+ private static final long serialVersionUID = 5677121412192984281L;
+
+ public ExecuteException() {
+ }
+
+ public ExecuteException(String message) {
+ super(message);
+ }
+
+ public ExecuteException(String message, Throwable cause) {
+ super(message, cause);
+ }
+
+ public ExecuteException(Throwable cause) {
+ super(cause);
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/7663fff4/core-job/src/main/java/org/apache/kylin/job/exception/IllegalStateTranferException.java
----------------------------------------------------------------------
diff --git a/core-job/src/main/java/org/apache/kylin/job/exception/IllegalStateTranferException.java b/core-job/src/main/java/org/apache/kylin/job/exception/IllegalStateTranferException.java
new file mode 100644
index 0000000..f19b0ca
--- /dev/null
+++ b/core-job/src/main/java/org/apache/kylin/job/exception/IllegalStateTranferException.java
@@ -0,0 +1,45 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 org.apache.kylin.job.exception;
+
+/**
+ */
+public class IllegalStateTranferException extends RuntimeException {
+
+ private static final long serialVersionUID = 8466551519300132702L;
+
+ public IllegalStateTranferException() {
+ }
+
+ public IllegalStateTranferException(String message) {
+ super(message);
+ }
+
+ public IllegalStateTranferException(String message, Throwable cause) {
+ super(message, cause);
+ }
+
+ public IllegalStateTranferException(Throwable cause) {
+ super(cause);
+ }
+
+ public IllegalStateTranferException(String message, Throwable cause, boolean enableSuppression, boolean writableStackTrace) {
+ super(message, cause, enableSuppression, writableStackTrace);
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/7663fff4/core-job/src/main/java/org/apache/kylin/job/exception/JobException.java
----------------------------------------------------------------------
diff --git a/core-job/src/main/java/org/apache/kylin/job/exception/JobException.java b/core-job/src/main/java/org/apache/kylin/job/exception/JobException.java
new file mode 100644
index 0000000..ba4c52a
--- /dev/null
+++ b/core-job/src/main/java/org/apache/kylin/job/exception/JobException.java
@@ -0,0 +1,58 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 org.apache.kylin.job.exception;
+
+/**
+ * @author xduo
+ *
+ */
+public class JobException extends Exception {
+
+ private static final long serialVersionUID = 1L;
+
+ /**
+ *
+ */
+ public JobException() {
+ super();
+ }
+
+ /**
+ * @param message
+ * @param cause
+ */
+ public JobException(String message, Throwable cause) {
+ super(message, cause);
+ }
+
+ /**
+ * @param message
+ */
+ public JobException(String message) {
+ super(message);
+ }
+
+ /**
+ * @param cause
+ */
+ public JobException(Throwable cause) {
+ super(cause);
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/7663fff4/core-job/src/main/java/org/apache/kylin/job/exception/LockException.java
----------------------------------------------------------------------
diff --git a/core-job/src/main/java/org/apache/kylin/job/exception/LockException.java b/core-job/src/main/java/org/apache/kylin/job/exception/LockException.java
new file mode 100644
index 0000000..cf43ac9
--- /dev/null
+++ b/core-job/src/main/java/org/apache/kylin/job/exception/LockException.java
@@ -0,0 +1,44 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 org.apache.kylin.job.exception;
+
+/**
+ */
+public class LockException extends Exception {
+ private static final long serialVersionUID = 2072745879281754945L;
+
+ public LockException() {
+ }
+
+ public LockException(String message) {
+ super(message);
+ }
+
+ public LockException(String message, Throwable cause) {
+ super(message, cause);
+ }
+
+ public LockException(Throwable cause) {
+ super(cause);
+ }
+
+ public LockException(String message, Throwable cause, boolean enableSuppression, boolean writableStackTrace) {
+ super(message, cause, enableSuppression, writableStackTrace);
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/7663fff4/core-job/src/main/java/org/apache/kylin/job/exception/PersistentException.java
----------------------------------------------------------------------
diff --git a/core-job/src/main/java/org/apache/kylin/job/exception/PersistentException.java b/core-job/src/main/java/org/apache/kylin/job/exception/PersistentException.java
new file mode 100644
index 0000000..8507a53
--- /dev/null
+++ b/core-job/src/main/java/org/apache/kylin/job/exception/PersistentException.java
@@ -0,0 +1,44 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 org.apache.kylin.job.exception;
+
+/**
+ */
+public class PersistentException extends Exception {
+ private static final long serialVersionUID = -4239863858506718998L;
+
+ public PersistentException() {
+ }
+
+ public PersistentException(String message) {
+ super(message);
+ }
+
+ public PersistentException(String message, Throwable cause) {
+ super(message, cause);
+ }
+
+ public PersistentException(Throwable cause) {
+ super(cause);
+ }
+
+ public PersistentException(String message, Throwable cause, boolean enableSuppression, boolean writableStackTrace) {
+ super(message, cause, enableSuppression, writableStackTrace);
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/7663fff4/core-job/src/main/java/org/apache/kylin/job/exception/SchedulerException.java
----------------------------------------------------------------------
diff --git a/core-job/src/main/java/org/apache/kylin/job/exception/SchedulerException.java b/core-job/src/main/java/org/apache/kylin/job/exception/SchedulerException.java
new file mode 100644
index 0000000..057bd4a
--- /dev/null
+++ b/core-job/src/main/java/org/apache/kylin/job/exception/SchedulerException.java
@@ -0,0 +1,44 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 org.apache.kylin.job.exception;
+
+/**
+ */
+public class SchedulerException extends Exception {
+ private static final long serialVersionUID = 349041244824274861L;
+
+ public SchedulerException() {
+ }
+
+ public SchedulerException(String message) {
+ super(message);
+ }
+
+ public SchedulerException(String message, Throwable cause) {
+ super(message, cause);
+ }
+
+ public SchedulerException(Throwable cause) {
+ super(cause);
+ }
+
+ public SchedulerException(String message, Throwable cause, boolean enableSuppression, boolean writableStackTrace) {
+ super(message, cause, enableSuppression, writableStackTrace);
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/7663fff4/core-job/src/main/java/org/apache/kylin/job/execution/AbstractExecutable.java
----------------------------------------------------------------------
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
new file mode 100644
index 0000000..be82b3a
--- /dev/null
+++ b/core-job/src/main/java/org/apache/kylin/job/execution/AbstractExecutable.java
@@ -0,0 +1,302 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 org.apache.kylin.job.execution;
+
+import com.google.common.base.Objects;
+import com.google.common.base.Preconditions;
+import com.google.common.collect.Lists;
+import com.google.common.collect.Maps;
+import org.apache.commons.lang.StringUtils;
+import org.apache.commons.lang3.tuple.Pair;
+import org.apache.kylin.common.KylinConfig;
+import org.apache.kylin.common.util.LogTitlePrinter;
+import org.apache.kylin.common.util.MailService;
+import org.apache.kylin.job.exception.ExecuteException;
+import org.apache.kylin.job.impl.threadpool.DefaultContext;
+import org.apache.kylin.job.manager.ExecutableManager;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.PrintWriter;
+import java.io.StringWriter;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+
+/**
+ */
+public abstract class AbstractExecutable implements Executable, Idempotent {
+
+ protected static final String SUBMITTER = "submitter";
+ 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 Logger logger = LoggerFactory.getLogger(AbstractExecutable.class);
+
+ private String name;
+ private String id;
+ private Map<String, String> params = Maps.newHashMap();
+
+ protected static ExecutableManager executableManager = ExecutableManager.getInstance(KylinConfig.getInstanceFromEnv());
+
+ public AbstractExecutable() {
+ setId(UUID.randomUUID().toString());
+ }
+
+ protected void onExecuteStart(ExecutableContext executableContext) {
+ Map<String, String> info = Maps.newHashMap();
+ info.put(START_TIME, Long.toString(System.currentTimeMillis()));
+ executableManager.updateJobOutput(getId(), ExecutableState.RUNNING, info, null);
+ }
+
+ protected void onExecuteFinished(ExecuteResult result, ExecutableContext executableContext) {
+ setEndTime(System.currentTimeMillis());
+ if (!isDiscarded()) {
+ if (result.succeed()) {
+ executableManager.updateJobOutput(getId(), ExecutableState.SUCCEED, null, result.output());
+ } else {
+ executableManager.updateJobOutput(getId(), ExecutableState.ERROR, null, result.output());
+ }
+ } else {
+ }
+ }
+
+ protected void onExecuteError(Throwable exception, ExecutableContext executableContext) {
+ if (!isDiscarded()) {
+ executableManager.addJobInfo(getId(), END_TIME, Long.toString(System.currentTimeMillis()));
+ String output = null;
+ if (exception != null) {
+ final StringWriter out = new StringWriter();
+ exception.printStackTrace(new PrintWriter(out));
+ output = out.toString();
+ }
+ executableManager.updateJobOutput(getId(), ExecutableState.ERROR, null, output);
+ } else {
+ }
+ }
+
+ @Override
+ public final ExecuteResult execute(ExecutableContext executableContext) throws ExecuteException {
+
+ //print a eye-catching title in log
+ LogTitlePrinter.printTitle(this.getName());
+
+ Preconditions.checkArgument(executableContext instanceof DefaultContext);
+ ExecuteResult result;
+ try {
+ onExecuteStart(executableContext);
+ result = doWork(executableContext);
+ } catch (Throwable e) {
+ logger.error("error running Executable", e);
+ onExecuteError(e, executableContext);
+ throw new ExecuteException(e);
+ }
+ onExecuteFinished(result, executableContext);
+ return result;
+ }
+
+ protected abstract ExecuteResult doWork(ExecutableContext context) throws ExecuteException;
+
+ @Override
+ public void cleanup() throws ExecuteException {
+
+ }
+
+ @Override
+ public boolean isRunnable() {
+ return this.getStatus() == ExecutableState.READY;
+ }
+
+ @Override
+ public String getName() {
+ return name;
+ }
+
+ public void setName(String name) {
+ this.name = name;
+ }
+
+ @Override
+ public final String getId() {
+ return this.id;
+ }
+
+ public final void setId(String id) {
+ this.id = id;
+ }
+
+ @Override
+ public final ExecutableState getStatus() {
+ return executableManager.getOutput(this.getId()).getState();
+ }
+
+ @Override
+ public final Map<String, String> getParams() {
+ return Collections.unmodifiableMap(this.params);
+ }
+
+ public final String getParam(String key) {
+ return this.params.get(key);
+ }
+
+ public final void setParam(String key, String value) {
+ this.params.put(key, value);
+ }
+
+ public final void setParams(Map<String, String> params) {
+ this.params.putAll(params);
+ }
+
+ public final long getLastModified() {
+ return executableManager.getOutput(getId()).getLastModified();
+ }
+
+ public final void setSubmitter(String submitter) {
+ setParam(SUBMITTER, submitter);
+ }
+
+ public final List<String> getNotifyList() {
+ final String str = getParam(NOTIFY_LIST);
+ if (str != null) {
+ return Lists.newArrayList(StringUtils.split(str, ","));
+ } else {
+ return Collections.emptyList();
+ }
+ }
+
+ public final void setNotifyList(String notifications) {
+ setParam(NOTIFY_LIST, notifications);
+ }
+
+ public final void setNotifyList(List<String> notifications) {
+ setNotifyList(StringUtils.join(notifications, ","));
+ }
+
+ protected Pair<String, String> formatNotifications(ExecutableContext executableContext, ExecutableState state) {
+ return null;
+ }
+
+ protected final void notifyUserStatusChange(ExecutableContext context, ExecutableState state) {
+ try {
+ List<String> users = Lists.newArrayList();
+ users.addAll(getNotifyList());
+ final String adminDls = KylinConfig.getInstanceFromEnv().getAdminDls();
+ if (null != adminDls) {
+ for (String adminDl : adminDls.split(",")) {
+ users.add(adminDl);
+ }
+ }
+ if (users.isEmpty()) {
+ return;
+ }
+ final Pair<String, String> email = formatNotifications(context, state);
+ if (email == null) {
+ return;
+ }
+ logger.info("prepare to send email to:" + users);
+ logger.info("job name:" + getName());
+ logger.info("submitter:" + getSubmitter());
+ logger.info("notify list:" + users);
+ new MailService().sendMail(users, email.getLeft(), email.getRight());
+ } catch (Exception e) {
+ logger.error(e.getLocalizedMessage(), e);
+ }
+ }
+
+ public final String getSubmitter() {
+ return getParam(SUBMITTER);
+ }
+
+ @Override
+ public final Output getOutput() {
+ return executableManager.getOutput(getId());
+ }
+
+ protected long getExtraInfoAsLong(String key, long defaultValue) {
+ return getExtraInfoAsLong(executableManager.getOutput(getId()), key, defaultValue);
+ }
+
+ public static long getStartTime(Output output) {
+ return getExtraInfoAsLong(output, START_TIME, 0L);
+ }
+
+ public static long getEndTime(Output output) {
+ return getExtraInfoAsLong(output, END_TIME, 0L);
+ }
+
+ public static long getDuration(long startTime, long endTime) {
+ if (startTime == 0) {
+ return 0;
+ }
+ if (endTime == 0) {
+ return System.currentTimeMillis() - startTime;
+ } else {
+ return endTime - startTime;
+ }
+ }
+
+ public static long getExtraInfoAsLong(Output output, String key, long defaultValue) {
+ final String str = output.getExtra().get(key);
+ if (str != null) {
+ return Long.parseLong(str);
+ } else {
+ return defaultValue;
+ }
+ }
+
+ protected final void addExtraInfo(String key, String value) {
+ executableManager.addJobInfo(getId(), key, value);
+ }
+
+ public final void setStartTime(long time) {
+ addExtraInfo(START_TIME, time + "");
+ }
+
+ public final void setEndTime(long time) {
+ addExtraInfo(END_TIME, time + "");
+ }
+
+ public final long getStartTime() {
+ return getExtraInfoAsLong(START_TIME, 0L);
+ }
+
+ public final long getEndTime() {
+ return getExtraInfoAsLong(END_TIME, 0L);
+ }
+
+ public final long getDuration() {
+ return getDuration(getStartTime(), getEndTime());
+ }
+
+ /*
+ * discarded is triggered by JobService, the Scheduler is not awake of that
+ *
+ * */
+ protected final boolean isDiscarded() {
+ final ExecutableState status = executableManager.getOutput(getId()).getState();
+ return status == ExecutableState.DISCARDED;
+ }
+
+ @Override
+ public String toString() {
+ return Objects.toStringHelper(this).add("id", getId()).add("name", getName()).add("state", getStatus()).toString();
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/7663fff4/core-job/src/main/java/org/apache/kylin/job/execution/ChainedExecutable.java
----------------------------------------------------------------------
diff --git a/core-job/src/main/java/org/apache/kylin/job/execution/ChainedExecutable.java b/core-job/src/main/java/org/apache/kylin/job/execution/ChainedExecutable.java
new file mode 100644
index 0000000..d5a7aae
--- /dev/null
+++ b/core-job/src/main/java/org/apache/kylin/job/execution/ChainedExecutable.java
@@ -0,0 +1,31 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 org.apache.kylin.job.execution;
+
+import java.util.List;
+
+/**
+ */
+public interface ChainedExecutable extends Executable {
+
+ List<? extends AbstractExecutable> getTasks();
+
+ void addTask(AbstractExecutable executable);
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/7663fff4/core-job/src/main/java/org/apache/kylin/job/execution/DefaultChainedExecutable.java
----------------------------------------------------------------------
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
new file mode 100644
index 0000000..6443762
--- /dev/null
+++ b/core-job/src/main/java/org/apache/kylin/job/execution/DefaultChainedExecutable.java
@@ -0,0 +1,122 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 org.apache.kylin.job.execution;
+
+import java.util.List;
+import java.util.Map;
+
+import com.google.common.collect.Lists;
+import com.google.common.collect.Maps;
+import org.apache.kylin.common.KylinConfig;
+import org.apache.kylin.job.exception.ExecuteException;
+import org.apache.kylin.job.manager.ExecutableManager;
+
+/**
+ */
+public class DefaultChainedExecutable extends AbstractExecutable implements ChainedExecutable {
+
+ private final List<AbstractExecutable> subTasks = Lists.newArrayList();
+
+ protected final ExecutableManager jobService = ExecutableManager.getInstance(KylinConfig.getInstanceFromEnv());
+
+ public DefaultChainedExecutable(){
+ super();
+ }
+
+ @Override
+ protected ExecuteResult doWork(ExecutableContext context) throws ExecuteException {
+ List<? extends Executable> executables = getTasks();
+ final int size = executables.size();
+ for (int i = 0; i < size; ++i) {
+ Executable subTask = executables.get(i);
+ if (subTask.isRunnable()) {
+ return subTask.execute(context);
+ }
+ }
+ return new ExecuteResult(ExecuteResult.State.SUCCEED, null);
+ }
+
+ @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) {
+ jobService.updateJobOutput(getId(), ExecutableState.RUNNING, null, null);
+ } else {
+ jobService.updateJobOutput(getId(), ExecutableState.RUNNING, info, null);
+ }
+ }
+
+ @Override
+ protected void onExecuteFinished(ExecuteResult result, ExecutableContext executableContext) {
+ if (isDiscarded()) {
+ setEndTime(System.currentTimeMillis());
+ notifyUserStatusChange(executableContext, ExecutableState.DISCARDED);
+ } else if (result.succeed()) {
+ List<? extends Executable> jobs = getTasks();
+ boolean allSucceed = true;
+ boolean hasError = false;
+ for (Executable task: jobs) {
+ final ExecutableState status = task.getStatus();
+ if (status == ExecutableState.ERROR) {
+ hasError = true;
+ }
+ if (status != ExecutableState.SUCCEED) {
+ allSucceed = false;
+ }
+ }
+ if (allSucceed) {
+ setEndTime(System.currentTimeMillis());
+ jobService.updateJobOutput(getId(), ExecutableState.SUCCEED, null, null);
+ notifyUserStatusChange(executableContext, ExecutableState.SUCCEED);
+ } else if (hasError) {
+ setEndTime(System.currentTimeMillis());
+ jobService.updateJobOutput(getId(), ExecutableState.ERROR, null, null);
+ notifyUserStatusChange(executableContext, ExecutableState.ERROR);
+ } else {
+ jobService.updateJobOutput(getId(), ExecutableState.READY, null, null);
+ }
+ } else {
+ setEndTime(System.currentTimeMillis());
+ jobService.updateJobOutput(getId(), ExecutableState.ERROR, null, null);
+ notifyUserStatusChange(executableContext, ExecutableState.ERROR);
+ }
+ }
+
+ @Override
+ public List<AbstractExecutable> getTasks() {
+ return subTasks;
+ }
+
+ public final AbstractExecutable getTaskByName(String name) {
+ for (AbstractExecutable task : subTasks) {
+ if (task.getName() != null && task.getName().equalsIgnoreCase(name)) {
+ return task;
+ }
+ }
+ return null;
+ }
+
+ @Override
+ public void addTask(AbstractExecutable executable) {
+ executable.setId(getId() + "-" + String.format("%02d", subTasks.size()));
+ this.subTasks.add(executable);
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/7663fff4/core-job/src/main/java/org/apache/kylin/job/execution/DefaultOutput.java
----------------------------------------------------------------------
diff --git a/core-job/src/main/java/org/apache/kylin/job/execution/DefaultOutput.java b/core-job/src/main/java/org/apache/kylin/job/execution/DefaultOutput.java
new file mode 100644
index 0000000..6bc3281
--- /dev/null
+++ b/core-job/src/main/java/org/apache/kylin/job/execution/DefaultOutput.java
@@ -0,0 +1,97 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 org.apache.kylin.job.execution;
+
+import org.apache.commons.lang3.StringUtils;
+
+import java.util.Map;
+
+/**
+ */
+public class DefaultOutput implements Output {
+
+ private ExecutableState state;
+ private Map<String, String> extra;
+ private String verboseMsg;
+ private long lastModified;
+
+ @Override
+ public Map<String, String> getExtra() {
+ return extra;
+ }
+
+ @Override
+ public String getVerboseMsg() {
+ return verboseMsg;
+ }
+
+ @Override
+ public ExecutableState getState() {
+ return state;
+ }
+
+ @Override
+ public long getLastModified() {
+ return lastModified;
+ }
+
+ public void setState(ExecutableState state) {
+ this.state = state;
+ }
+
+ public void setExtra(Map<String, String> extra) {
+ this.extra = extra;
+ }
+
+ public void setVerboseMsg(String verboseMsg) {
+ this.verboseMsg = verboseMsg;
+ }
+
+ public void setLastModified(long lastModified) {
+ this.lastModified = lastModified;
+ }
+
+ @Override
+ public int hashCode() {
+ final int prime = 31;
+ int hashCode = state.hashCode();
+ hashCode = hashCode * prime + extra.hashCode();
+ hashCode = hashCode * prime + verboseMsg.hashCode();
+ hashCode = hashCode * prime + Long.valueOf(lastModified).hashCode();
+ return hashCode;
+ }
+
+ @Override
+ public boolean equals(Object obj) {
+ if (!(obj instanceof DefaultOutput)) {
+ return false;
+ }
+ DefaultOutput another = ((DefaultOutput) obj);
+ if (this.state != another.state) {
+ return false;
+ }
+ if (!extra.equals(another.extra)) {
+ return false;
+ }
+ if (this.lastModified != another.lastModified) {
+ return false;
+ }
+ return StringUtils.equals(verboseMsg, another.verboseMsg);
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/7663fff4/core-job/src/main/java/org/apache/kylin/job/execution/Executable.java
----------------------------------------------------------------------
diff --git a/core-job/src/main/java/org/apache/kylin/job/execution/Executable.java b/core-job/src/main/java/org/apache/kylin/job/execution/Executable.java
new file mode 100644
index 0000000..a7f1358
--- /dev/null
+++ b/core-job/src/main/java/org/apache/kylin/job/execution/Executable.java
@@ -0,0 +1,42 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 org.apache.kylin.job.execution;
+
+import org.apache.kylin.job.exception.ExecuteException;
+
+import java.util.Map;
+
+/**
+ */
+public interface Executable {
+
+ String getId();
+
+ String getName();
+
+ ExecuteResult execute(ExecutableContext executableContext) throws ExecuteException;
+
+ ExecutableState getStatus();
+
+ Output getOutput();
+
+ boolean isRunnable();
+
+ Map<String, String> getParams();
+}
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/7663fff4/core-job/src/main/java/org/apache/kylin/job/execution/ExecutableContext.java
----------------------------------------------------------------------
diff --git a/core-job/src/main/java/org/apache/kylin/job/execution/ExecutableContext.java b/core-job/src/main/java/org/apache/kylin/job/execution/ExecutableContext.java
new file mode 100644
index 0000000..e3f99ca
--- /dev/null
+++ b/core-job/src/main/java/org/apache/kylin/job/execution/ExecutableContext.java
@@ -0,0 +1,30 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 org.apache.kylin.job.execution;
+
+import org.apache.kylin.common.KylinConfig;
+
+/**
+ */
+public interface ExecutableContext {
+
+ Object getSchedulerContext();
+
+ KylinConfig getConfig();
+}
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/7663fff4/core-job/src/main/java/org/apache/kylin/job/execution/ExecutableState.java
----------------------------------------------------------------------
diff --git a/core-job/src/main/java/org/apache/kylin/job/execution/ExecutableState.java b/core-job/src/main/java/org/apache/kylin/job/execution/ExecutableState.java
new file mode 100644
index 0000000..5dad4b3
--- /dev/null
+++ b/core-job/src/main/java/org/apache/kylin/job/execution/ExecutableState.java
@@ -0,0 +1,82 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 org.apache.kylin.job.execution;
+
+import java.util.Collection;
+import java.util.Set;
+import java.util.concurrent.CopyOnWriteArraySet;
+
+import com.google.common.base.Supplier;
+import com.google.common.collect.Maps;
+import com.google.common.collect.Multimap;
+import com.google.common.collect.Multimaps;
+
+/**
+ */
+public enum ExecutableState {
+
+ READY,
+ RUNNING,
+ ERROR,
+ STOPPED,
+ DISCARDED,
+ SUCCEED;
+
+ private static Multimap<ExecutableState, ExecutableState> VALID_STATE_TRANSFER;
+
+ static {
+ VALID_STATE_TRANSFER = Multimaps.newSetMultimap(Maps.<ExecutableState, Collection<ExecutableState>>newEnumMap(ExecutableState.class), new Supplier<Set<ExecutableState>>() {
+ @Override
+ public Set<ExecutableState> get() {
+ return new CopyOnWriteArraySet<ExecutableState>();
+ }
+ });
+
+ //scheduler
+ VALID_STATE_TRANSFER.put(ExecutableState.READY, ExecutableState.RUNNING);
+ VALID_STATE_TRANSFER.put(ExecutableState.READY, ExecutableState.ERROR);
+ //user
+ VALID_STATE_TRANSFER.put(ExecutableState.READY, ExecutableState.DISCARDED);
+
+ //job
+ VALID_STATE_TRANSFER.put(ExecutableState.RUNNING, ExecutableState.READY);
+ //job
+ VALID_STATE_TRANSFER.put(ExecutableState.RUNNING, ExecutableState.SUCCEED);
+ //user
+ VALID_STATE_TRANSFER.put(ExecutableState.RUNNING, ExecutableState.DISCARDED);
+ //scheduler,job
+ VALID_STATE_TRANSFER.put(ExecutableState.RUNNING, ExecutableState.ERROR);
+
+
+ VALID_STATE_TRANSFER.put(ExecutableState.STOPPED, ExecutableState.DISCARDED);
+ VALID_STATE_TRANSFER.put(ExecutableState.STOPPED, ExecutableState.READY);
+
+ VALID_STATE_TRANSFER.put(ExecutableState.ERROR, ExecutableState.DISCARDED);
+ VALID_STATE_TRANSFER.put(ExecutableState.ERROR, ExecutableState.READY);
+ }
+
+ public boolean isFinalState() {
+ return this == SUCCEED || this == DISCARDED;
+ }
+
+ public static boolean isValidStateTransfer(ExecutableState from, ExecutableState to) {
+ return VALID_STATE_TRANSFER.containsEntry(from, to);
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/7663fff4/core-job/src/main/java/org/apache/kylin/job/execution/ExecuteResult.java
----------------------------------------------------------------------
diff --git a/core-job/src/main/java/org/apache/kylin/job/execution/ExecuteResult.java b/core-job/src/main/java/org/apache/kylin/job/execution/ExecuteResult.java
new file mode 100644
index 0000000..cddc0f7
--- /dev/null
+++ b/core-job/src/main/java/org/apache/kylin/job/execution/ExecuteResult.java
@@ -0,0 +1,54 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 org.apache.kylin.job.execution;
+
+import com.google.common.base.Preconditions;
+
+/**
+ */
+public final class ExecuteResult {
+
+ public static enum State {SUCCEED, FAILED, ERROR, DISCARDED, STOPPED}
+
+ private final State state;
+ private final String output;
+
+ public ExecuteResult(State state) {
+ this(state, "");
+ }
+
+ public ExecuteResult(State state, String output) {
+ Preconditions.checkArgument(state != null, "state cannot be null");
+ this.state = state;
+ this.output = output;
+ }
+
+ public State state() {
+ return state;
+ }
+
+ public boolean succeed() {
+ return state == State.SUCCEED;
+ }
+
+
+ public String output() {
+ return output;
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/7663fff4/core-job/src/main/java/org/apache/kylin/job/execution/Idempotent.java
----------------------------------------------------------------------
diff --git a/core-job/src/main/java/org/apache/kylin/job/execution/Idempotent.java b/core-job/src/main/java/org/apache/kylin/job/execution/Idempotent.java
new file mode 100644
index 0000000..98c950e
--- /dev/null
+++ b/core-job/src/main/java/org/apache/kylin/job/execution/Idempotent.java
@@ -0,0 +1,28 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 org.apache.kylin.job.execution;
+
+import org.apache.kylin.job.exception.ExecuteException;
+
+/**
+ */
+public interface Idempotent {
+
+ void cleanup() throws ExecuteException;
+}
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/7663fff4/core-job/src/main/java/org/apache/kylin/job/execution/Output.java
----------------------------------------------------------------------
diff --git a/core-job/src/main/java/org/apache/kylin/job/execution/Output.java b/core-job/src/main/java/org/apache/kylin/job/execution/Output.java
new file mode 100644
index 0000000..4d93132
--- /dev/null
+++ b/core-job/src/main/java/org/apache/kylin/job/execution/Output.java
@@ -0,0 +1,34 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 org.apache.kylin.job.execution;
+
+import java.util.Map;
+
+/**
+ */
+public interface Output {
+
+ Map<String, String> getExtra();
+
+ String getVerboseMsg();
+
+ ExecutableState getState();
+
+ long getLastModified();
+}
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/7663fff4/core-job/src/main/java/org/apache/kylin/job/impl/threadpool/DefaultContext.java
----------------------------------------------------------------------
diff --git a/core-job/src/main/java/org/apache/kylin/job/impl/threadpool/DefaultContext.java b/core-job/src/main/java/org/apache/kylin/job/impl/threadpool/DefaultContext.java
new file mode 100644
index 0000000..e959ae2
--- /dev/null
+++ b/core-job/src/main/java/org/apache/kylin/job/impl/threadpool/DefaultContext.java
@@ -0,0 +1,61 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 org.apache.kylin.job.impl.threadpool;
+
+import org.apache.kylin.common.KylinConfig;
+import org.apache.kylin.job.execution.Executable;
+import org.apache.kylin.job.execution.ExecutableContext;
+
+import java.util.Collections;
+import java.util.Map;
+import java.util.concurrent.ConcurrentMap;
+
+/**
+ */
+public class DefaultContext implements ExecutableContext {
+
+ private final ConcurrentMap<String, Executable> runningJobs;
+ private final KylinConfig kylinConfig;
+
+ public DefaultContext(ConcurrentMap<String, Executable> runningJobs, KylinConfig kylinConfig) {
+ this.runningJobs = runningJobs;
+ this.kylinConfig = kylinConfig;
+ }
+ @Override
+ public Object getSchedulerContext() {
+ return null;
+ }
+
+ @Override
+ public KylinConfig getConfig() {
+ return kylinConfig;
+ }
+
+ void addRunningJob(Executable executable) {
+ runningJobs.put(executable.getId(), executable);
+ }
+
+ void removeRunningJob(Executable executable) {
+ runningJobs.remove(executable.getId());
+ }
+
+ public Map<String, Executable> getRunningJobs() {
+ return Collections.unmodifiableMap(runningJobs);
+ }
+}