/**
 * 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.oozie.command.coord;

import org.apache.commons.lang3.StringUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.oozie.AppType;
import org.apache.oozie.CoordinatorActionBean;
import org.apache.oozie.CoordinatorJobBean;
import org.apache.oozie.ErrorCode;
import org.apache.oozie.SLAEventBean;
import org.apache.oozie.client.CoordinatorJob;
import org.apache.oozie.client.Job;
import org.apache.oozie.client.SLAEvent.SlaAppType;
import org.apache.oozie.client.rest.JsonBean;
import org.apache.oozie.command.CommandException;
import org.apache.oozie.command.MaterializeTransitionXCommand;
import org.apache.oozie.command.PreconditionException;
import org.apache.oozie.command.bundle.BundleStatusUpdateXCommand;
import org.apache.oozie.coord.CoordUtils;
import org.apache.oozie.coord.TimeUnit;
import org.apache.oozie.executor.jpa.BatchQueryExecutor;
import org.apache.oozie.executor.jpa.BatchQueryExecutor.UpdateEntry;
import org.apache.oozie.executor.jpa.CoordActionsActiveCountJPAExecutor;
import org.apache.oozie.executor.jpa.CoordJobQueryExecutor;
import org.apache.oozie.executor.jpa.CoordJobQueryExecutor.CoordJobQuery;
import org.apache.oozie.executor.jpa.JPAExecutorException;
import org.apache.oozie.service.ConfigurationService;
import org.apache.oozie.service.CoordMaterializeTriggerService;
import org.apache.oozie.service.EventHandlerService;
import org.apache.oozie.service.JPAService;
import org.apache.oozie.service.Service;
import org.apache.oozie.service.Services;
import org.apache.oozie.sla.SLAOperations;
import org.apache.oozie.util.DateUtils;
import org.apache.oozie.util.Instrumentation;
import org.apache.oozie.util.LogUtils;
import org.apache.oozie.util.ParamChecker;
import org.apache.oozie.util.StatusUtils;
import org.apache.oozie.util.XConfiguration;
import org.apache.oozie.util.XmlUtils;
import org.apache.oozie.util.db.SLADbOperations;
import org.jdom.Element;
import org.jdom.JDOMException;

import java.io.IOException;
import java.io.StringReader;
import java.sql.Timestamp;
import java.util.Calendar;
import java.util.Date;
import java.util.TimeZone;

/**
 * Materialize actions for specified start and end time for coordinator job.
 */
@SuppressWarnings("deprecation")
public class CoordMaterializeTransitionXCommand extends MaterializeTransitionXCommand {

    private JPAService jpaService = null;
    private CoordinatorJobBean coordJob = null;
    private String jobId = null;
    private Date startMatdTime = null;
    private Date endMatdTime = null;
    private final int materializationWindow;
    private int lastActionNumber = 1; // over-ride by DB value
    private CoordinatorJob.Status prevStatus = null;

    static final private int lookAheadWindow = ConfigurationService.getInt(CoordMaterializeTriggerService
            .CONF_LOOKUP_INTERVAL);

    /**
     * Default MAX timeout in minutes, after which coordinator input check will timeout
     */
    public static final String CONF_DEFAULT_MAX_TIMEOUT = Service.CONF_PREFIX + "coord.default.max.timeout";

    /**
     * The constructor for class {@link CoordMaterializeTransitionXCommand}
     *
     * @param jobId coordinator job id
     * @param materializationWindow materialization window to calculate end time
     */
    public CoordMaterializeTransitionXCommand(String jobId, int materializationWindow) {
        super("coord_mater", "coord_mater", 1);
        this.jobId = ParamChecker.notEmpty(jobId, "jobId");
        this.materializationWindow = materializationWindow;
    }

    public CoordMaterializeTransitionXCommand(CoordinatorJobBean coordJob, int materializationWindow, Date startTime,
                                              Date endTime) {
        super("coord_mater", "coord_mater", 1);
        this.jobId = ParamChecker.notEmpty(coordJob.getId(), "jobId");
        this.materializationWindow = materializationWindow;
        this.coordJob = coordJob;
        this.startMatdTime = startTime;
        this.endMatdTime = endTime;
    }

    @Override
    protected void setLogInfo() {
        LogUtils.setLogInfo(jobId);
    }

    @Override
    public void transitToNext() throws CommandException {
    }

    @Override
    public void updateJob() throws CommandException {
        updateList.add(new UpdateEntry(CoordJobQuery.UPDATE_COORD_JOB_MATERIALIZE,coordJob));
    }

    @Override
    public void performWrites() throws CommandException {
        try {
            BatchQueryExecutor.getInstance().executeBatchInsertUpdateDelete(insertList, updateList, null);
            // register the partition related dependencies of actions
            for (JsonBean actionBean : insertList) {
                if (actionBean instanceof CoordinatorActionBean) {
                    CoordinatorActionBean coordAction = (CoordinatorActionBean) actionBean;
                    if (EventHandlerService.isEnabled()) {
                        CoordinatorXCommand.generateEvent(coordAction, coordJob.getUser(), coordJob.getAppName(), null);
                    }

                    // TODO: time 100s should be configurable
                    queue(new CoordActionNotificationXCommand(coordAction), 100);

                    //Delay for input check = (nominal time - now)
                    long checkDelay = coordAction.getNominalTime().getTime() - new Date().getTime();
                    queue(new CoordActionInputCheckXCommand(coordAction.getId(), coordAction.getJobId()),
                        Math.max(checkDelay, 0));

                    if (!StringUtils.isEmpty(coordAction.getPushMissingDependencies())) {
                        // TODO: Delay in catchup mode?
                        queue(new CoordPushDependencyCheckXCommand(coordAction.getId(), true), 100);
                    }
                }
            }
        }
        catch (JPAExecutorException jex) {
            throw new CommandException(jex);
        }
    }

    @Override
    public String getEntityKey() {
        return this.jobId;
    }

    @Override
    protected boolean isLockRequired() {
        return true;
    }

    @Override
    protected void loadState() throws CommandException {
        jpaService = Services.get().get(JPAService.class);
        if (jpaService == null) {
            LOG.error(ErrorCode.E0610);
        }

        try {
            coordJob = CoordJobQueryExecutor.getInstance().get(CoordJobQuery.GET_COORD_JOB_MATERIALIZE, jobId);
            prevStatus = coordJob.getStatus();
        }
        catch (JPAExecutorException jex) {
            throw new CommandException(jex);
        }
        LogUtils.setLogInfo(coordJob);
        // calculate start materialize and end materialize time
        calcMatdTime();
    }

    /**
     * Calculate startMatdTime and endMatdTime from job's start time if next materialized time is null
     *
     * @throws CommandException thrown if failed to calculate startMatdTime and endMatdTime
     */
    protected void calcMatdTime() throws CommandException {
        Timestamp startTime = coordJob.getNextMaterializedTimestamp();
        if (startTime == null) {
            startTime = coordJob.getStartTimestamp();
        }
        // calculate end time by adding materializationWindow to start time.
        // need to convert materializationWindow from secs to milliseconds
        long startTimeMilli = startTime.getTime();
        long endTimeMilli = startTimeMilli + (materializationWindow * 1000);

        startMatdTime = DateUtils.toDate(new Timestamp(startTimeMilli));
        endMatdTime = DateUtils.toDate(new Timestamp(endTimeMilli));
        endMatdTime = getMaterializationTimeForCatchUp(endMatdTime);
        // if MaterializationWindow end time is greater than endTime
        // for job, then set it to endTime of job
        Date jobEndTime = coordJob.getEndTime();
        if (endMatdTime.compareTo(jobEndTime) > 0) {
            endMatdTime = jobEndTime;
        }

        LOG.debug("Materializing coord job id=" + jobId + ", start=" + DateUtils.formatDateOozieTZ(startMatdTime) + ", end="
                + DateUtils.formatDateOozieTZ(endMatdTime)
                + ", window=" + materializationWindow);
    }

    /**
     * Get materialization for window for catch-up jobs. for current jobs,it reruns currentMatdate, For catch-up, end
     * Mataterilized Time = startMatdTime + MatThrottling * frequency; unless LAST_ONLY execution order is set, in which
     * case it returns now (to materialize all actions in the past)
     *
     * @param currentMatTime
     * @return Date returns materialization for window for catch-up jobs
     * @throws CommandException
     * @throws JDOMException
     */
    private Date getMaterializationTimeForCatchUp(Date currentMatTime) throws CommandException {
        if (currentMatTime.after(new Date())) {
            return currentMatTime;
        }
        if (coordJob.getExecutionOrder().equals(CoordinatorJob.Execution.LAST_ONLY) ||
                coordJob.getExecutionOrder().equals(CoordinatorJob.Execution.NONE)) {
            return new Date();
        }
        final int frequency;
        try {
            frequency = Integer.parseInt(coordJob.getFrequency());
        }
        catch (final NumberFormatException e) {
            // Cron based frequency: catching up at maximum till the coordinator job's end time,
            // bounded also by the throttle parameter, aka the number of coordinator actions to materialize
            return coordJob.getEndTime();
        }

        TimeZone appTz = DateUtils.getTimeZone(coordJob.getTimeZone());
        TimeUnit freqTU = TimeUnit.valueOf(coordJob.getTimeUnitStr());
        Calendar startInstance = Calendar.getInstance(appTz);
        startInstance.setTime(startMatdTime);
        Calendar endMatInstance = null;
        Calendar previousInstance = startInstance;
        for (int i = 1; i <= coordJob.getMatThrottling(); i++) {
            endMatInstance = (Calendar) startInstance.clone();
            endMatInstance.add(freqTU.getCalendarUnit(), i * frequency);
            if (endMatInstance.getTime().compareTo(new Date()) >= 0) {
                if (previousInstance.getTime().after(currentMatTime)) {
                    return previousInstance.getTime();
                }
                else {
                    return currentMatTime;
                }
            }
            previousInstance = endMatInstance;
        }
        if (endMatInstance == null) {
            return currentMatTime;
        }
        else {
            return endMatInstance.getTime();
        }
    }

    @Override
    protected void verifyPrecondition() throws CommandException, PreconditionException {
        if (!(coordJob.getStatus() == CoordinatorJobBean.Status.PREP || coordJob.getStatus() == CoordinatorJobBean.Status.RUNNING
                || coordJob.getStatus() == CoordinatorJobBean.Status.RUNNINGWITHERROR)) {
            throw new PreconditionException(ErrorCode.E1100, "CoordMaterializeTransitionXCommand for jobId=" + jobId
                    + " job is not in PREP or RUNNING but in " + coordJob.getStatus());
        }

        if (coordJob.isDoneMaterialization()) {
            throw new PreconditionException(ErrorCode.E1100, "CoordMaterializeTransitionXCommand for jobId =" + jobId
                    + " job is already materialized");
        }

        if (coordJob.getNextMaterializedTimestamp() != null
                && coordJob.getNextMaterializedTimestamp().compareTo(coordJob.getEndTimestamp()) >= 0) {
            throw new PreconditionException(ErrorCode.E1100, "CoordMaterializeTransitionXCommand for jobId=" + jobId
                    + " job is already materialized");
        }

        Timestamp startTime = coordJob.getNextMaterializedTimestamp();
        if (startTime == null) {
            startTime = coordJob.getStartTimestamp();

            if (startTime.after(new Timestamp(System.currentTimeMillis() + lookAheadWindow * 1000))) {
                throw new PreconditionException(ErrorCode.E1100, "CoordMaterializeTransitionXCommand for jobId="
                        + jobId + " job's start time is not reached yet - nothing to materialize");
            }
        }

        if (coordJob.getNextMaterializedTimestamp() != null
                && coordJob.getNextMaterializedTimestamp().after(
                        new Timestamp(System.currentTimeMillis() + lookAheadWindow * 1000))) {
            throw new PreconditionException(ErrorCode.E1100, "CoordMaterializeTransitionXCommand for jobId=" + jobId
                    + " Request is for future time. Lookup time is  "
                    + new Timestamp(System.currentTimeMillis() + lookAheadWindow * 1000) + " mat time is "
                    + coordJob.getNextMaterializedTimestamp());
        }

        if (coordJob.getLastActionTime() != null && coordJob.getLastActionTime().compareTo(coordJob.getEndTime()) >= 0) {
            throw new PreconditionException(ErrorCode.E1100, "ENDED Coordinator materialization for jobId = " + jobId
                    + ", all actions have been materialized from start time = " + coordJob.getStartTime()
                    + " to end time = " + coordJob.getEndTime() + ", job status = " + coordJob.getStatusStr());
        }

        if (coordJob.getLastActionTime() != null && coordJob.getLastActionTime().compareTo(endMatdTime) >= 0) {
            throw new PreconditionException(ErrorCode.E1100, "ENDED Coordinator materialization for jobId = " + jobId
                    + ", action is *already* materialized for Materialization start time = " + startMatdTime
                    + ", materialization end time = " + endMatdTime + ", job status = " + coordJob.getStatusStr());
        }

        if (endMatdTime.after(coordJob.getEndTime())) {
            throw new PreconditionException(ErrorCode.E1100, "ENDED Coordinator materialization for jobId = " + jobId
                    + " materialization end time = " + endMatdTime + " surpasses coordinator job's end time = "
                    + coordJob.getEndTime() + " job status = " + coordJob.getStatusStr());
        }

        if (coordJob.getPauseTime() != null && !startMatdTime.before(coordJob.getPauseTime())) {
            throw new PreconditionException(ErrorCode.E1100, "ENDED Coordinator materialization for jobId = " + jobId
                    + ", materialization start time = " + startMatdTime
                    + " is after or equal to coordinator job's pause time = " + coordJob.getPauseTime()
                    + ", job status = " + coordJob.getStatusStr());
        }

    }

    @Override
    protected void materialize() throws CommandException {
        Instrumentation.Cron cron = new Instrumentation.Cron();
        cron.start();
        try {
            materializeActions(false);
            updateJobMaterializeInfo(coordJob);
        }
        catch (CommandException ex) {
            LOG.warn("Exception occurred:" + ex.getMessage() + " Making the job failed ", ex);
            coordJob.setStatus(Job.Status.FAILED);
            coordJob.resetPending();
            // remove any materialized actions and slaEvents
            insertList.clear();
        }
        catch (Exception e) {
            LOG.error("Exception occurred:" + e.getMessage() + " Making the job failed ", e);
            coordJob.setStatus(Job.Status.FAILED);
            try {
                CoordJobQueryExecutor.getInstance().executeUpdate(CoordJobQuery.UPDATE_COORD_JOB_MATERIALIZE, coordJob);
            }
            catch (JPAExecutorException jex) {
                throw new CommandException(ErrorCode.E1011, jex);
            }
            throw new CommandException(ErrorCode.E1012, e.getMessage(), e);
        } finally {
            cron.stop();
            instrumentation.addCron(INSTRUMENTATION_GROUP, getName() + ".materialize", cron);
        }
    }

    /**
     * Create action instances starting from "startMatdTime" to "endMatdTime" and store them into coord action table.
     *
     * @param dryrun if this is a dry run
     * @return materialized action(s)
     * @throws Exception thrown if failed to materialize actions
     */
    protected String materializeActions(boolean dryrun) throws Exception {

        Configuration jobConf = null;
        try {
            jobConf = new XConfiguration(new StringReader(coordJob.getConf()));
        }
        catch (IOException ioe) {
            LOG.warn("Configuration parse error. read from DB :" + coordJob.getConf(), ioe);
            throw new CommandException(ErrorCode.E1005, ioe.getMessage(), ioe);
        }

        String jobXml = coordJob.getJobXml();
        Element eJob = XmlUtils.parseXml(jobXml);
        TimeZone appTz = DateUtils.getTimeZone(coordJob.getTimeZone());

        String frequency = coordJob.getFrequency();
        TimeUnit freqTU = TimeUnit.valueOf(coordJob.getTimeUnitStr());
        TimeUnit endOfFlag = TimeUnit.valueOf(eJob.getAttributeValue("end_of_duration"));
        Calendar start = Calendar.getInstance(appTz);
        start.setTime(startMatdTime);
        DateUtils.moveToEnd(start, endOfFlag);
        Calendar end = Calendar.getInstance(appTz);
        end.setTime(endMatdTime);
        lastActionNumber = coordJob.getLastActionNumber();
        //Intentionally printing dates in their own timezone, not Oozie timezone
        LOG.info("materialize actions for tz=" + appTz.getDisplayName() + ",\n start=" + start.getTime() + ", end="
                + end.getTime() + ",\n timeUnit " + freqTU.getCalendarUnit() + ",\n frequency :" + frequency + ":"
                + freqTU + ",\n lastActionNumber " + lastActionNumber);
        // Keep the actual start time
        Calendar origStart = Calendar.getInstance(appTz);
        origStart.setTime(coordJob.getStartTimestamp());
        // Move to the End of duration, if needed.
        DateUtils.moveToEnd(origStart, endOfFlag);

        StringBuilder actionStrings = new StringBuilder();
        Date jobPauseTime = coordJob.getPauseTime();
        Calendar pause = null;
        if (jobPauseTime != null) {
            pause = Calendar.getInstance(appTz);
            pause.setTime(DateUtils.convertDateToTimestamp(jobPauseTime));
        }

        String action = null;
        int numWaitingActions = dryrun ? 0 : jpaService.execute(new CoordActionsActiveCountJPAExecutor(coordJob.getId()));
        int maxActionToBeCreated = coordJob.getMatThrottling() - numWaitingActions;
        // If LAST_ONLY and all materialization is in the past, ignore maxActionsToBeCreated
        boolean ignoreMaxActions =
                (coordJob.getExecutionOrder().equals(CoordinatorJob.Execution.LAST_ONLY) ||
                        coordJob.getExecutionOrder().equals(CoordinatorJob.Execution.NONE))
                        && endMatdTime.before(new Date());
        LOG.debug("Coordinator job :" + coordJob.getId() + ", maxActionToBeCreated :" + maxActionToBeCreated
                + ", Mat_Throttle :" + coordJob.getMatThrottling() + ", numWaitingActions :" + numWaitingActions);

        boolean isCronFrequency = false;

        Calendar effStart = (Calendar) start.clone();
        try {
            int intFrequency = Integer.parseInt(coordJob.getFrequency());
            effStart = (Calendar) origStart.clone();
            effStart.add(freqTU.getCalendarUnit(), lastActionNumber * intFrequency);
        }
        catch (NumberFormatException e) {
            isCronFrequency = true;
        }

        boolean firstMater = true;

        while (effStart.compareTo(end) < 0 && (ignoreMaxActions || maxActionToBeCreated-- > 0)) {
            if (pause != null && effStart.compareTo(pause) >= 0) {
                break;
            }

            Date nextTime = effStart.getTime();

            if (isCronFrequency) {
                if (effStart.getTime().compareTo(startMatdTime) == 0 && firstMater) {
                    effStart.add(Calendar.MINUTE, -1);
                    firstMater = false;
                }
                nextTime = CoordCommandUtils.getNextValidActionTimeForCronFrequency(effStart.getTime(), coordJob);
                Date prevTime = new Date(effStart.getTimeInMillis());
                effStart.setTime(nextTime);
                addDSTChangeToNominalTime(prevTime, nextTime, coordJob);
            }

            if (effStart.compareTo(end) < 0) {

                if (pause != null && effStart.compareTo(pause) >= 0) {
                    break;
                }
                CoordinatorActionBean actionBean = new CoordinatorActionBean();
                lastActionNumber++;

                int timeout = coordJob.getTimeout();
                LOG.debug("Materializing action for time=" + DateUtils.formatDateOozieTZ(effStart.getTime())
                        + ", lastactionnumber=" + lastActionNumber + " timeout=" + timeout + " minutes");
                Date actualTime = new Date();
                action = CoordCommandUtils.materializeOneInstance(jobId, dryrun, (Element) eJob.clone(),
                        nextTime, actualTime, lastActionNumber, jobConf, actionBean);
                actionBean.setTimeOut(timeout);
                if (!dryrun) {
                    storeToDB(actionBean, action, jobConf); // Storing to table

                }
                else {
                    actionStrings.append("action for new instance");
                    actionStrings.append(action);
                }
            }
            else {
                break;
            }

            if (!isCronFrequency) {
                effStart = (Calendar) origStart.clone();
                effStart.add(freqTU.getCalendarUnit(), lastActionNumber * Integer.parseInt(coordJob.getFrequency()));
            }
        }

        if (isCronFrequency) {
            if (effStart.compareTo(end) < 0 && !(ignoreMaxActions || maxActionToBeCreated-- > 0)) {
                //Since we exceed the throttle, we need to move the nextMadtime forward
                //to avoid creating duplicate actions
                if (!firstMater) {
                    effStart.setTime(CoordCommandUtils.getNextValidActionTimeForCronFrequency(effStart.getTime(), coordJob));
                }
            }
        }

        endMatdTime = effStart.getTime();

        if (!dryrun) {
            return action;
        }
        else {
            return actionStrings.toString();
        }
    }

    /**
     * Apply DST correction according the job`s timezone, if the difference between the previous nominal time and the actual one
     * is greater or equal than 24 hours.
     * Calendar uses a similar approach: applies DST change if the TimeUnit is lower or equal than TimeUnit.DAY. With this approach
     * a similar behaviour can be achieved.
     *
     * @see {@http://oozie.apache.org/docs/5.0.0/CoordinatorFunctionalSpec.html#a7._Handling_Timezones_and_Daylight_Saving_Time}
     *
     * @param prevTime nominal time of the previous coordinator action
     * @param nextTime nominal time of the actual coordinator action
     * @param coordJob the coordinator job
     */
    private void addDSTChangeToNominalTime(Date prevTime, Date nextTime, CoordinatorJobBean coordJob) {
        final long differenceBetweenTwoActionsInSec = java.util.concurrent.TimeUnit.MILLISECONDS.toSeconds
                (Math.abs(prevTime.getTime() - nextTime.getTime()));
        final long oneDayInSeconds = java.util.concurrent.TimeUnit.DAYS.toSeconds(1);
        if (differenceBetweenTwoActionsInSec < oneDayInSeconds) {
            return;
        }
        final long dstOffset = DaylightOffsetCalculator.getDSTOffset(DateUtils
                .getTimeZone(coordJob.getTimeZone()), coordJob.getStartTime(), nextTime);
        LOG.debug("[{0}] ms DST offset applied to nominal time for coordinator job: [{1}]", dstOffset, coordJob.getId());
        nextTime.setTime(nextTime.getTime() + dstOffset);
    }

    private void storeToDB(CoordinatorActionBean actionBean, String actionXml, Configuration jobConf) throws Exception {
        LOG.debug("In storeToDB() coord action id = " + actionBean.getId() + ", size of actionXml = "
                + actionXml.length());
        actionBean.setActionXml(actionXml);
        insertList.add(actionBean);
        writeActionSlaRegistration(actionXml, actionBean, jobConf);
    }

    private void writeActionSlaRegistration(String actionXml, CoordinatorActionBean actionBean, Configuration jobConf)
            throws Exception {
        Element eAction = XmlUtils.parseXml(actionXml);
        Element eSla = eAction.getChild("action", eAction.getNamespace()).getChild("info", eAction.getNamespace("sla"));
                SLAEventBean slaEvent = SLADbOperations.createSlaRegistrationEvent(eSla, actionBean.getId(),
                                 SlaAppType.COORDINATOR_ACTION, coordJob.getUser(), coordJob.getGroup(), LOG);
                         if (slaEvent != null) {
            insertList.add(slaEvent);
        }
        // inserting into new table also
        SLAOperations.createSlaRegistrationEvent(eSla, actionBean.getId(), actionBean.getJobId(),
                AppType.COORDINATOR_ACTION, coordJob.getUser(), coordJob.getAppName(), LOG, false,
                CoordUtils.isSlaAlertDisabled(actionBean, coordJob.getAppName(), jobConf));
    }

    private void updateJobMaterializeInfo(CoordinatorJobBean job) throws CommandException {
        job.setLastActionTime(endMatdTime);
        job.setLastActionNumber(lastActionNumber);
        // if the job endtime == action endtime, we don't need to materialize this job anymore
        Date jobEndTime = job.getEndTime();

        if (job.getStatus() == CoordinatorJob.Status.PREP){
            LOG.info("[" + job.getId() + "]: Update status from " + job.getStatus() + " to RUNNING");
            job.setStatus(Job.Status.RUNNING);
        }
        job.setPending();
        Calendar end = Calendar.getInstance();
        end.setTime(jobEndTime);

        if (end.getTime().compareTo(endMatdTime) <= 0) {
            LOG.info("[" + job.getId() + "]: all actions have been materialized, set pending to true");
            // set doneMaterialization to true when materialization is done
            job.setDoneMaterialization();
        }
        job.setStatus(StatusUtils.getStatus(job));
        LOG.info("Coord Job status updated to = " + job.getStatus());
        job.setNextMaterializedTime(endMatdTime);
    }

    @Override
    public String getKey() {
        return getName() + "_" + jobId;
    }

    @Override
    public void notifyParent() throws CommandException {
        // update bundle action only when status changes in coord job
        if (this.coordJob.getBundleId() != null) {
            if (!prevStatus.equals(coordJob.getStatus())) {
                BundleStatusUpdateXCommand bundleStatusUpdate = new BundleStatusUpdateXCommand(coordJob, prevStatus);
                bundleStatusUpdate.call();
            }
        }
    }
}