import org.apache.kylin.job.dao.ExecutableOutputPO; //导入依赖的package包/类
public void forceKillJob(String jobId) {
    try {
        final ExecutableOutputPO jobOutput = executableDao.getJobOutput(jobId);
        jobOutput.setStatus(ExecutableState.ERROR.toString());
        List<ExecutablePO> tasks = executableDao.getJob(jobId).getTasks();

        for (ExecutablePO task : tasks) {
            if (executableDao.getJobOutput(task.getId()).getStatus().equals("SUCCEED")) {
                continue;
            } else if (executableDao.getJobOutput(task.getId()).getStatus().equals("RUNNING")) {
                updateJobOutput(task.getId(), ExecutableState.READY, Maps.<String, String> newHashMap(), "");
            }
            break;
        }
        executableDao.updateJobOutput(jobOutput);
    } catch (PersistentException e) {
        throw new RuntimeException(e);
    }
}
 

import org.apache.kylin.job.dao.ExecutableOutputPO; //导入依赖的package包/类
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);
    }
}
 

import org.apache.kylin.job.dao.ExecutableOutputPO; //导入依赖的package包/类
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);
        }
    }
}
 

import org.apache.kylin.job.dao.ExecutableOutputPO; //导入依赖的package包/类
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);
        return parseOutput(jobOutput);
    } catch (PersistentException e) {
        logger.error("fail to get job output:" + uuid, e);
        throw new RuntimeException(e);
    }
}
 

import org.apache.kylin.job.dao.ExecutableOutputPO; //导入依赖的package包/类
private DefaultOutput parseOutput(ExecutableOutputPO jobOutput) {
    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;
}
 

import org.apache.kylin.job.dao.ExecutableOutputPO; //导入依赖的package包/类
public Map<String, Output> getAllOutputs() {
    try {
        final List<ExecutableOutputPO> jobOutputs = executableDao.getJobOutputs();
        HashMap<String, Output> result = Maps.newHashMap();
        for (ExecutableOutputPO jobOutput : jobOutputs) {
            result.put(jobOutput.getId(), parseOutput(jobOutput));
        }
        return result;
    } catch (PersistentException e) {
        logger.error("fail to get all job output:", e);
        throw new RuntimeException(e);
    }
}
 

import org.apache.kylin.job.dao.ExecutableOutputPO; //导入依赖的package包/类
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);
    }
}
 

import org.apache.kylin.job.dao.ExecutableOutputPO; //导入依赖的package包/类
public void resumeAllRunningJobs() {
    try {
        final List<ExecutableOutputPO> jobOutputs = executableDao.getJobOutputs();
        for (ExecutableOutputPO executableOutputPO : jobOutputs) {
            if (executableOutputPO.getStatus().equalsIgnoreCase(ExecutableState.RUNNING.toString())) {
                executableOutputPO.setStatus(ExecutableState.READY.toString());
                executableDao.updateJobOutput(executableOutputPO);
            }
        }
    } catch (PersistentException e) {
        logger.error("error reset job status from RUNNING to READY", e);
        throw new RuntimeException(e);
    }
}
 

import org.apache.kylin.job.dao.ExecutableOutputPO; //导入依赖的package包/类
public ExecutableOutputPO getJobOutput(String jobId) {
    try {
        return executableDao.getJobOutput(jobId);
    } catch (PersistentException e) {
        logger.error("Can't get output of Job " + jobId);
        throw new RuntimeException(e);
    }
}
 

import org.apache.kylin.job.dao.ExecutableOutputPO; //导入依赖的package包/类
public void updateJobOutput(String jobId, ExecutableState newStatus, Map<String, String> info, String output) {
    // when 
    if (Thread.currentThread().isInterrupted()) {
        throw new RuntimeException("Current thread is interruptted, aborting");
    }

    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 + ", job id: " + jobId);
            }
            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);
        throw new RuntimeException(e);
    }
}
 

import org.apache.kylin.job.dao.ExecutableOutputPO; //导入依赖的package包/类
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);
    }
}
 

import org.apache.kylin.job.dao.ExecutableOutputPO; //导入依赖的package包/类
public void addJobInfo(String id, Map<String, String> info) {
    if (info == null) {
        return;
    }

    // post process
    if (info.containsKey(MR_JOB_ID) && !info.containsKey(ExecutableConstants.YARN_APP_ID)) {
        String jobId = info.get(MR_JOB_ID);
        if (jobId.startsWith("job_")) {
            info.put(YARN_APP_ID, jobId.replace("job_", "application_"));
        }
    }

    if (info.containsKey(YARN_APP_ID) && !StringUtils.isEmpty(config.getJobTrackingURLPattern())) {
        String pattern = config.getJobTrackingURLPattern();
        try {
            String newTrackingURL = String.format(pattern, info.get(YARN_APP_ID));
            info.put(YARN_APP_URL, newTrackingURL);
        } catch (IllegalFormatException ife) {
            logger.error("Illegal tracking url pattern: " + config.getJobTrackingURLPattern());
        }
    }

    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);
    }
}
 

import org.apache.kylin.job.dao.ExecutableOutputPO; //导入依赖的package包/类
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);
        }
    }
}
 

import org.apache.kylin.job.dao.ExecutableOutputPO; //导入依赖的package包/类
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);
    }
}
 

import org.apache.kylin.job.dao.ExecutableOutputPO; //导入依赖的package包/类
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);
    }
}
 

import org.apache.kylin.job.dao.ExecutableOutputPO; //导入依赖的package包/类
public void cleanup() throws Exception {
    CubeManager cubeManager = CubeManager.getInstance(config);

    List<String> toDeleteResource = Lists.newArrayList();

    // two level resources, snapshot tables and cube statistics
    for (String resourceRoot : new String[] { ResourceStore.SNAPSHOT_RESOURCE_ROOT, ResourceStore.CUBE_STATISTICS_ROOT }) {
        NavigableSet<String> snapshotTables = getStore().listResources(resourceRoot);

        if (snapshotTables != null) {
            for (String snapshotTable : snapshotTables) {
                NavigableSet<String> snapshotNames = getStore().listResources(snapshotTable);
                if (snapshotNames != null)
                    for (String snapshot : snapshotNames) {
                        if (isOlderThanThreshold(getStore().getResourceTimestamp(snapshot)))
                            toDeleteResource.add(snapshot);
                    }
            }
        }
    }

    // three level resources, only dictionaries
    NavigableSet<String> dictTables = getStore().listResources(ResourceStore.DICT_RESOURCE_ROOT);

    if (dictTables != null) {
        for (String table : dictTables) {
            NavigableSet<String> tableColNames = getStore().listResources(table);
            if (tableColNames != null)
                for (String tableCol : tableColNames) {
                    NavigableSet<String> dictionaries = getStore().listResources(tableCol);
                    if (dictionaries != null)
                        for (String dict : dictionaries)
                            if (isOlderThanThreshold(getStore().getResourceTimestamp(dict)))
                                toDeleteResource.add(dict);
                }
        }
    }

    Set<String> activeResourceList = Sets.newHashSet();
    for (org.apache.kylin.cube.CubeInstance cube : cubeManager.listAllCubes()) {
        for (org.apache.kylin.cube.CubeSegment segment : cube.getSegments()) {
            activeResourceList.addAll(segment.getSnapshotPaths());
            activeResourceList.addAll(segment.getDictionaryPaths());
            activeResourceList.add(segment.getStatisticsResourcePath());
        }
    }

    toDeleteResource.removeAll(activeResourceList);

    // delete old and completed jobs
    ExecutableDao executableDao = ExecutableDao.getInstance(KylinConfig.getInstanceFromEnv());
    List<ExecutablePO> allExecutable = executableDao.getJobs();
    for (ExecutablePO executable : allExecutable) {
        long lastModified = executable.getLastModified();
        ExecutableOutputPO output = executableDao.getJobOutput(executable.getUuid());
        if (System.currentTimeMillis() - lastModified > TIME_THREADSHOLD_FOR_JOB && (ExecutableState.SUCCEED.toString().equals(output.getStatus()) || ExecutableState.DISCARDED.toString().equals(output.getStatus()))) {
            toDeleteResource.add(ResourceStore.EXECUTE_RESOURCE_ROOT + "/" + executable.getUuid());
            toDeleteResource.add(ResourceStore.EXECUTE_OUTPUT_RESOURCE_ROOT + "/" + executable.getUuid());

            for (ExecutablePO task : executable.getTasks()) {
                toDeleteResource.add(ResourceStore.EXECUTE_OUTPUT_RESOURCE_ROOT + "/" + task.getUuid());
            }
        }
    }

    if (toDeleteResource.size() > 0) {
        logger.info("The following resources have no reference or is too old, will be cleaned from metadata store: \n");

        for (String s : toDeleteResource) {
            logger.info(s);
            if (delete == true) {
                getStore().deleteResource(s);
            }
        }
    } else {
        logger.info("No resource to be cleaned up from metadata store;");
    }

}