001/**
002 * Licensed to the Apache Software Foundation (ASF) under one
003 * or more contributor license agreements.  See the NOTICE file
004 * distributed with this work for additional information
005 * regarding copyright ownership.  The ASF licenses this file
006 * to you under the Apache License, Version 2.0 (the
007 * "License"); you may not use this file except in compliance
008 * with the License.  You may obtain a copy of the License at
009 *
010 *      http://www.apache.org/licenses/LICENSE-2.0
011 *
012 * Unless required by applicable law or agreed to in writing, software
013 * distributed under the License is distributed on an "AS IS" BASIS,
014 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
015 * See the License for the specific language governing permissions and
016 * limitations under the License.
017 */
018
019package org.apache.oozie.command.bundle;
020
021import java.util.Date;
022import java.util.List;
023
024import org.apache.oozie.BundleActionBean;
025import org.apache.oozie.BundleJobBean;
026import org.apache.oozie.ErrorCode;
027import org.apache.oozie.client.Job;
028import org.apache.oozie.command.CommandException;
029import org.apache.oozie.command.PreconditionException;
030import org.apache.oozie.command.SuspendTransitionXCommand;
031import org.apache.oozie.command.coord.CoordSuspendXCommand;
032import org.apache.oozie.executor.jpa.BundleActionQueryExecutor.BundleActionQuery;
033import org.apache.oozie.executor.jpa.BatchQueryExecutor;
034import org.apache.oozie.executor.jpa.BundleActionQueryExecutor;
035import org.apache.oozie.executor.jpa.BundleJobQueryExecutor;
036import org.apache.oozie.executor.jpa.BundleJobQueryExecutor.BundleJobQuery;
037import org.apache.oozie.executor.jpa.JPAExecutorException;
038import org.apache.oozie.executor.jpa.BatchQueryExecutor.UpdateEntry;
039import org.apache.oozie.util.InstrumentUtils;
040import org.apache.oozie.util.ParamChecker;
041import org.apache.oozie.util.LogUtils;
042
043public class BundleJobSuspendXCommand extends SuspendTransitionXCommand {
044    private final String jobId;
045    private List<BundleActionBean> bundleActions;
046    private BundleJobBean bundleJob;
047
048    public BundleJobSuspendXCommand(String id) {
049        super("bundle_suspend", "bundle_suspend", 1);
050        this.jobId = ParamChecker.notEmpty(id, "id");
051    }
052
053    /* (non-Javadoc)
054     * @see org.apache.oozie.command.TransitionXCommand#getJob()
055     */
056    @Override
057    public Job getJob() {
058        return bundleJob;
059    }
060
061    /* (non-Javadoc)
062     * @see org.apache.oozie.command.TransitionXCommand#notifyParent()
063     */
064    @Override
065    public void notifyParent() throws CommandException {
066    }
067
068    /* (non-Javadoc)
069     * @see org.apache.oozie.command.TransitionXCommand#setJob(org.apache.oozie.client.Job)
070     */
071    @Override
072    public void setJob(Job job) {
073    }
074
075    /* (non-Javadoc)
076     * @see org.apache.oozie.command.SuspendTransitionXCommand#performWrites()
077     */
078    @Override
079    public void performWrites() throws CommandException {
080        try {
081            BatchQueryExecutor.getInstance().executeBatchInsertUpdateDelete(null, updateList, null);
082        }
083        catch (JPAExecutorException e) {
084            throw new CommandException(e);
085        }
086    }
087
088    /* (non-Javadoc)
089     * @see org.apache.oozie.command.XCommand#getEntityKey()
090     */
091    @Override
092    public String getEntityKey() {
093        return this.jobId;
094    }
095
096    /* (non-Javadoc)
097     * @see org.apache.oozie.command.XCommand#isLockRequired()
098     */
099    @Override
100    protected boolean isLockRequired() {
101        return true;
102    }
103
104    /* (non-Javadoc)
105     * @see org.apache.oozie.command.XCommand#loadState()
106     */
107    @Override
108    protected void loadState() throws CommandException {
109        try {
110            bundleJob = BundleJobQueryExecutor.getInstance().get(BundleJobQuery.GET_BUNDLE_JOB, jobId);
111        }
112        catch (Exception Ex) {
113            throw new CommandException(ErrorCode.E0604, jobId);
114        }
115
116        try {
117            bundleActions = BundleActionQueryExecutor.getInstance().getList(
118                    BundleActionQuery.GET_BUNDLE_ACTIONS_STATUS_UNIGNORED_FOR_BUNDLE, bundleJob.getId());
119        }
120        catch (Exception Ex) {
121            throw new CommandException(ErrorCode.E1311, jobId);
122        }
123
124        LogUtils.setLogInfo(bundleJob);
125    }
126
127    /* (non-Javadoc)
128     * @see org.apache.oozie.command.XCommand#verifyPrecondition()
129     */
130    @Override
131    protected void verifyPrecondition() throws CommandException, PreconditionException {
132        if (bundleJob.getStatus() == Job.Status.SUCCEEDED || bundleJob.getStatus() == Job.Status.FAILED
133                || bundleJob.getStatus() == Job.Status.KILLED || bundleJob.getStatus() == Job.Status.DONEWITHERROR) {
134            LOG.info("BundleJobSuspendXCommand is not going to execute because job either succeeded, failed, killed, or donewitherror; id = "
135                            + jobId + ", status = " + bundleJob.getStatus());
136            throw new PreconditionException(ErrorCode.E1312, jobId, bundleJob.getStatus().toString());
137        }
138    }
139
140    /* (non-Javadoc)
141     * @see org.apache.oozie.command.TransitionXCommand#updateJob()
142     */
143    @Override
144    public void updateJob() {
145        InstrumentUtils.incrJobCounter("bundle_suspend", 1, null);
146        bundleJob.setSuspendedTime(new Date());
147        bundleJob.setLastModifiedTime(new Date());
148
149        LOG.debug("Suspend bundle job id = " + jobId + ", status = " + bundleJob.getStatus() + ", pending = " + bundleJob.isPending());
150        updateList.add(new UpdateEntry<BundleJobQuery>(BundleJobQuery.UPDATE_BUNDLE_JOB_STATUS_PENDING_SUSP_MOD_TIME, bundleJob));
151    }
152
153    @Override
154    public void suspendChildren() throws CommandException {
155        for (BundleActionBean action : this.bundleActions) {
156            if (action.getStatus() == Job.Status.RUNNING || action.getStatus() == Job.Status.RUNNINGWITHERROR
157                    || action.getStatus() == Job.Status.PREP || action.getStatus() == Job.Status.PAUSED
158                    || action.getStatus() == Job.Status.PAUSEDWITHERROR) {
159                // queue a CoordSuspendXCommand
160                if (action.getCoordId() != null) {
161                    queue(new CoordSuspendXCommand(action.getCoordId()));
162                    updateBundleAction(action);
163                    LOG.debug("Suspend bundle action = [{0}], new status = [{1}], pending = [{2}] and queue CoordSuspendXCommand for [{3}]",
164                            action.getBundleActionId(), action.getStatus(), action.getPending(), action.getCoordId());
165                } else {
166                    updateBundleAction(action);
167                    LOG.debug("Suspend bundle action = [{0}], new status = [{1}], pending = [{2}] and coord id is null",
168                            action.getBundleActionId(), action.getStatus(), action.getPending());
169                }
170
171            }
172        }
173        LOG.debug("Suspended bundle actions for the bundle=[{0}]", jobId);
174    }
175
176    private void updateBundleAction(BundleActionBean action) {
177        if (action.getStatus() == Job.Status.PREP) {
178            action.setStatus(Job.Status.PREPSUSPENDED);
179        }
180        else if (action.getStatus() == Job.Status.RUNNING) {
181            action.setStatus(Job.Status.SUSPENDED);
182        }
183        else if (action.getStatus() == Job.Status.RUNNINGWITHERROR) {
184            action.setStatus(Job.Status.SUSPENDEDWITHERROR);
185        }
186        else if (action.getStatus() == Job.Status.PAUSED) {
187            action.setStatus(Job.Status.SUSPENDED);
188        }
189        else if (action.getStatus() == Job.Status.PAUSEDWITHERROR) {
190            action.setStatus(Job.Status.SUSPENDEDWITHERROR);
191        }
192
193        action.incrementAndGetPending();
194        action.setLastModifiedTime(new Date());
195        updateList.add(new UpdateEntry<BundleActionQuery>(
196                BundleActionQuery.UPDATE_BUNDLE_ACTION_STATUS_PENDING_MODTIME, action));
197    }
198}