Source code

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.HashMap;
023import java.util.List;
024
025import org.apache.oozie.BundleActionBean;
026import org.apache.oozie.BundleJobBean;
027import org.apache.oozie.ErrorCode;
028import org.apache.oozie.client.Job;
029import org.apache.oozie.client.Job.Status;
030import org.apache.oozie.command.CommandException;
031import org.apache.oozie.command.PreconditionException;
032import org.apache.oozie.command.StatusTransitXCommand;
033import org.apache.oozie.executor.jpa.BundleActionQueryExecutor;
034import org.apache.oozie.executor.jpa.BundleJobQueryExecutor;
035import org.apache.oozie.executor.jpa.BundleActionQueryExecutor.BundleActionQuery;
036import org.apache.oozie.executor.jpa.BundleJobQueryExecutor.BundleJobQuery;
037import org.apache.oozie.executor.jpa.JPAExecutorException;
038import org.apache.oozie.util.LogUtils;
039import org.apache.oozie.util.StatusUtils;
040
041/**
042 * BundleStatusTransitXCommand update job's status according to its child actions' status. If all child actions' pending
043 * flag equals 0 (job done), we reset the job's pending flag to 0. If all child actions are succeeded, we set the job's
044 * status to SUCCEEDED.
045 */
046public class BundleStatusTransitXCommand extends StatusTransitXCommand {
047
048    private String jobId;
049    private List<BundleActionBean> bundleActions;
050    private BundleJobBean bundleJob;
051    private boolean foundPending;
052    private HashMap<Job.Status, Integer> bundleActionStatus = new HashMap<Job.Status, Integer>();
053
054    public BundleStatusTransitXCommand(String id) {
055        super("bundle_status_transit", "bundle_status_transit", 0);
056        this.jobId = id;
057    }
058
059    @Override
060    public String getEntityKey() {
061        return jobId;
062    }
063
064    @Override
065    protected void loadState() throws CommandException {
066        try {
067            bundleJob = BundleJobQueryExecutor.getInstance().get(
068                    BundleJobQuery.GET_BUNDLE_JOB_ID_STATUS_PENDING_MOD_PAUSE_SUSPEND_TIME, jobId);
069
070            bundleActions = BundleActionQueryExecutor.getInstance().getList(
071                    BundleActionQuery.GET_BUNDLE_UNIGNORED_ACTION_STATUS_PENDING_FOR_BUNDLE, jobId);
072            for (BundleActionBean bAction : bundleActions) {
073                int counter = 0;
074                if (bundleActionStatus.containsKey(bAction.getStatus())) {
075                    counter = getActionStatusCount(bAction.getStatus()) + 1;
076                }
077                else {
078                    ++counter;
079                }
080                bundleActionStatus.put(bAction.getStatus(), counter);
081                if (bAction.getCoordId() == null
082                        && (bAction.getStatus() == Job.Status.FAILED || bAction.getStatus() == Job.Status.KILLED) ) {
083                    new BundleKillXCommand(jobId).call();
084                    bundleJob = BundleJobQueryExecutor.getInstance().get(
085                            BundleJobQuery.GET_BUNDLE_JOB_ID_STATUS_PENDING_MOD_PAUSE_SUSPEND_TIME, jobId);
086                    bundleJob.setStatus(Job.Status.FAILED);
087                    bundleJob.setLastModifiedTime(new Date());
088                    BundleJobQueryExecutor.getInstance().executeUpdate(BundleJobQuery.UPDATE_BUNDLE_JOB_STATUS,
089                            bundleJob);
090                }
091
092                if (bAction.isPending()) {
093                    foundPending = true;
094                }
095            }
096            LogUtils.setLogInfo(bundleJob);
097        }
098        catch (JPAExecutorException e) {
099            throw new CommandException(ErrorCode.E1322, e);
100        }
101    }
102
103    @Override
104    protected Job.Status getJobStatus() throws CommandException {
105        Job.Status jobStatus = super.getJobStatus();
106        if (jobStatus == null) {
107            if (isPrepRunningState()) {
108                return getPrepRunningStatus();
109            }
110        }
111
112        return jobStatus;
113    }
114
115    @Override
116    protected boolean isTerminalState() {
117        return !foundPending
118                && bundleActions.size() == getActionStatusCount(Job.Status.SUCCEEDED)
119                        + getActionStatusCount(Job.Status.FAILED) + getActionStatusCount(Job.Status.KILLED)
120                        + getActionStatusCount(Job.Status.DONEWITHERROR);
121    }
122
123    @Override
124    protected Job.Status getTerminalStatus() {
125
126        // If all bundle action is done and bundle is killed, then don't change the status.
127        if (bundleJob.getStatus().equals(Job.Status.KILLED)) {
128            return Job.Status.KILLED;
129
130        }
131        // If all the bundle actions are succeeded then bundle job should be succeeded.
132        if (bundleActions.size() == getActionStatusCount(Job.Status.SUCCEEDED)) {
133            return Job.Status.SUCCEEDED;
134
135        }
136        else if (bundleActions.size() == getActionStatusCount(Job.Status.KILLED)) {
137            // If all the bundle actions are KILLED then bundle job should be KILLED.
138            return Job.Status.KILLED;
139        }
140        else if (bundleActions.size() == getActionStatusCount(Job.Status.FAILED)) {
141            // If all the bundle actions are FAILED then bundle job should be FAILED.
142            return Job.Status.FAILED;
143        }
144        else {
145            return Job.Status.DONEWITHERROR;
146
147        }
148    }
149
150    @Override
151    protected boolean isPausedState() {
152        //If bundle is paused then timestamp will be set.
153        //If bundleJob.getPauseTime() is not set, that means that status has to be computed from bottom-up.
154        if (bundleJob.getStatus() == Job.Status.PAUSED || bundleJob.getStatus() == Job.Status.PAUSEDWITHERROR
155                && bundleJob.getPauseTime() != null) {
156            return true;
157        }
158        else {
159            return getBottomUpPauseStatus() != null;
160        }
161
162    }
163
164    @Override
165    protected Job.Status getPausedState() {
166        if (bundleJob.getStatus() == Job.Status.PAUSED || bundleJob.getStatus() == Job.Status.PAUSEDWITHERROR) {
167            if (hasTerminatedActions() || bundleActionStatus.containsKey(Job.Status.SUSPENDEDWITHERROR)
168                    || bundleActionStatus.containsKey(Job.Status.RUNNINGWITHERROR)
169                    || bundleActionStatus.containsKey(Job.Status.PAUSEDWITHERROR)) {
170                return Job.Status.PAUSEDWITHERROR;
171            }
172            else {
173                return Job.Status.PAUSED;
174            }
175        }
176        return getBottomUpPauseStatus();
177
178    }
179
180    @Override
181    protected boolean isSuspendedState() {
182        //If bundle is suspended then timestamp will be set.
183        //If bundleJob.getSuspendedTimestamp() is not set, that means that status has to be computed from bottom-up.
184        if ((bundleJob.getStatus() == Job.Status.SUSPENDED || bundleJob.getStatus() == Job.Status.SUSPENDEDWITHERROR)
185                && bundleJob.getSuspendedTimestamp() != null) {
186            return true;
187        }
188
189        return getBottomUpSuspendedState() != null;
190
191    }
192
193    @Override
194    protected Job.Status getSuspendedStatus() {
195        if (bundleJob.getStatus() == Job.Status.SUSPENDED || bundleJob.getStatus() == Job.Status.SUSPENDEDWITHERROR) {
196            if (hasTerminatedActions() || bundleActionStatus.containsKey(Job.Status.SUSPENDEDWITHERROR)
197                    || bundleActionStatus.containsKey(Job.Status.PAUSEDWITHERROR)) {
198                return Job.Status.SUSPENDEDWITHERROR;
199            }
200            else {
201                return Job.Status.SUSPENDED;
202            }
203
204        }
205        return getBottomUpSuspendedState();
206
207    }
208
209    @Override
210    protected boolean isRunningState() {
211        return true;
212    }
213
214    @Override
215    protected Status getRunningState() {
216        if (bundleJob.getStatus() != Job.Status.PREP) {
217            return getRunningStatus(bundleActionStatus);
218        }
219        else
220            return null;
221    }
222
223    @Override
224    protected void updateJobStatus(Job.Status bundleStatus) throws JPAExecutorException {
225        LOG.info("Set bundle job [" + jobId + "] status to '" + bundleStatus + "' from '" + bundleJob.getStatus() + "'");
226
227        String jobId = bundleJob.getId();
228        // Update the Bundle Job
229        // Check for backward support when RUNNINGWITHERROR, SUSPENDEDWITHERROR and
230        // PAUSEDWITHERROR is not supported
231        bundleJob.setStatus(StatusUtils.getStatusIfBackwardSupportTrue(bundleStatus));
232        bundleJob.setLastModifiedTime(new Date());
233        if (foundPending) {
234            bundleJob.setPending();
235            LOG.info("Bundle job [" + jobId + "] Pending set to TRUE");
236        }
237        else {
238            bundleJob.resetPending();
239            LOG.info("Bundle job [" + jobId + "] Pending set to FALSE");
240        }
241        BundleJobQueryExecutor.getInstance().executeUpdate(BundleJobQuery.UPDATE_BUNDLE_JOB_STATUS_PENDING_MODTIME,
242                bundleJob);
243    }
244
245    /**
246     * bottom up; check the status of parent through their children.
247     *
248     * @return the bottom up pause status
249     */
250    private Job.Status getBottomUpPauseStatus() {
251
252        if (bundleActionStatus.containsKey(Job.Status.PAUSED)
253                && bundleActions.size() == getActionStatusCount(Job.Status.PAUSED)) {
254            return Job.Status.PAUSED;
255
256        }
257        else if (bundleActionStatus.containsKey(Job.Status.PAUSEDWITHERROR)
258                && bundleActions.size() == getActionStatusCount(Job.Status.PAUSED)
259                        + getActionStatusCount(Job.Status.PAUSEDWITHERROR)) {
260            return Job.Status.PAUSEDWITHERROR;
261        }
262
263        return null;
264    }
265
266    /**
267     * Bottom up update status of parent from the status of its children.
268     *
269     * @return the bottom up suspended state
270     */
271    private Job.Status getBottomUpSuspendedState() {
272
273        if (!foundPending && bundleActionStatus.containsKey(Job.Status.SUSPENDED)
274                || bundleActionStatus.containsKey(Job.Status.SUSPENDEDWITHERROR)) {
275
276            if (bundleActions.size() == getActionStatusCount(Job.Status.SUSPENDED)
277                    + getActionStatusCount(Job.Status.SUCCEEDED)) {
278                return Job.Status.SUSPENDED;
279            }
280            else if (bundleActions.size() == getActionStatusCount(Job.Status.SUSPENDEDWITHERROR)
281                    + getActionStatusCount(Job.Status.SUSPENDED) + getActionStatusCount(Job.Status.SUCCEEDED)
282                    + getActionStatusCount(Job.Status.KILLED) + getActionStatusCount(Job.Status.FAILED)
283                    + getActionStatusCount(Job.Status.DONEWITHERROR)) {
284                return Job.Status.SUSPENDEDWITHERROR;
285
286            }
287        }
288        return null;
289    }
290
291    private boolean hasTerminatedActions() {
292        return bundleActionStatus.containsKey(Job.Status.KILLED) || bundleActionStatus.containsKey(Job.Status.FAILED)
293                || bundleActionStatus.containsKey(Job.Status.DONEWITHERROR);
294
295    }
296
297    private boolean isPrepRunningState() {
298        return !foundPending && bundleActionStatus.containsKey(Job.Status.PREP)
299                && bundleActions.size() > getActionStatusCount(Job.Status.PREP);
300    }
301
302    private Status getPrepRunningStatus() {
303        return getRunningStatus(bundleActionStatus);
304
305    }
306
307    private int getActionStatusCount(final Job.Status status) {
308
309        if (bundleActionStatus.containsKey(status)) {
310            return bundleActionStatus.get(status);
311        }
312        else {
313            return 0;
314        }
315    }
316
317    private Job.Status getRunningStatus(HashMap<Job.Status, Integer> actionStatus) {
318        if (actionStatus.containsKey(Job.Status.FAILED) || actionStatus.containsKey(Job.Status.KILLED)
319                || actionStatus.containsKey(Job.Status.DONEWITHERROR)
320                || actionStatus.containsKey(Job.Status.RUNNINGWITHERROR)) {
321            return Job.Status.RUNNINGWITHERROR;
322        }
323        else {
324            return Job.Status.RUNNING;
325        }
326    }
327
328    @Override
329    protected void verifyPrecondition() throws CommandException, PreconditionException {
330    }
331
332}