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.coord;
020
021import java.util.Date;
022import java.util.HashMap;
023import java.util.List;
024import java.util.Map;
025
026import org.apache.oozie.CoordinatorActionBean;
027import org.apache.oozie.CoordinatorJobBean;
028import org.apache.oozie.ErrorCode;
029import org.apache.oozie.client.CoordinatorAction;
030import org.apache.oozie.client.Job;
031import org.apache.oozie.client.Job.Status;
032import org.apache.oozie.command.CommandException;
033import org.apache.oozie.command.PreconditionException;
034import org.apache.oozie.command.StatusTransitXCommand;
035import org.apache.oozie.command.bundle.BundleStatusUpdateXCommand;
036import org.apache.oozie.executor.jpa.CoordActionQueryExecutor;
037import org.apache.oozie.executor.jpa.CoordActionQueryExecutor.CoordActionQuery;
038import org.apache.oozie.executor.jpa.CoordJobQueryExecutor;
039import org.apache.oozie.executor.jpa.CoordJobQueryExecutor.CoordJobQuery;
040import org.apache.oozie.executor.jpa.JPAExecutorException;
041import org.apache.oozie.service.SchemaService;
042import org.apache.oozie.service.Services;
043import org.apache.oozie.service.StatusTransitService;
044import org.apache.oozie.util.LogUtils;
045import org.apache.oozie.util.StatusUtils;
046
047/**
048 * CoordStatusTransitXCommand update coord job's status according to its child actions' status. If all child actions'
049 * pending flag equals 0 (job done), we reset the job's pending flag to 0. If all child actions are succeeded, we set
050 * the job's status to SUCCEEDED.
051 */
052public class CoordStatusTransitXCommand extends StatusTransitXCommand {
053
054    private final String jobId;
055    private CoordinatorJobBean coordJob;
056    int coordActionCount;
057    private final Map<CoordinatorAction.Status, Integer> coordActionStatus = new HashMap<CoordinatorAction.Status, Integer>();
058    boolean isPending = false;
059
060    final boolean backwardSupportForCoordStatus = Services.get().getConf()
061            .getBoolean(StatusTransitService.CONF_BACKWARD_SUPPORT_FOR_COORD_STATUS, false);
062
063    public CoordStatusTransitXCommand(String jobId) {
064        super("coord_status_transit", "coord_status_transit", 0);
065        this.jobId = jobId;
066    }
067
068    @Override
069    public String getEntityKey() {
070        return jobId;
071    }
072
073    @Override
074    protected void loadState() throws CommandException {
075        try {
076            coordJob = CoordJobQueryExecutor.getInstance().get(CoordJobQuery.GET_COORD_JOB, jobId);
077            List<CoordinatorActionBean> coordActionStatusList = CoordActionQueryExecutor.getInstance().getList(
078                    CoordActionQuery.GET_COORD_ACTIONS_STATUS_UNIGNORED, jobId);
079
080            long count = (Long) CoordActionQueryExecutor.getInstance().getSingleValue(
081                    CoordActionQuery.GET_COORD_ACTIONS_PENDING_COUNT, jobId);
082            if (count > 0) {
083                isPending = true;
084            }
085
086            for (CoordinatorAction coordAction : coordActionStatusList) {
087                int counter = 0;
088                if (coordActionStatus.containsKey(coordAction.getStatus())) {
089                    counter = getStatusCount(coordAction.getStatus()) + 1;
090                }
091                else {
092                    ++counter;
093                }
094                coordActionStatus.put(coordAction.getStatus(), counter);
095            }
096            coordActionCount = coordActionStatusList.size();
097        }
098        catch (JPAExecutorException jpae) {
099            throw new CommandException(ErrorCode.E1025, jpae);
100        }
101        LogUtils.setLogInfo(this.coordJob);
102    }
103
104    @Override
105    protected void verifyPrecondition() throws CommandException, PreconditionException {
106        // if namespace 0.1 is used and backward support is true, then ignore this coord job
107        if (backwardSupportForCoordStatus == true && coordJob.getAppNamespace() != null
108                && coordJob.getAppNamespace().equals(SchemaService.COORDINATOR_NAMESPACE_URI_1)) {
109            throw new CommandException(ErrorCode.E1025,
110                    " Coord namespace is 0.1 and backward.support.for.coord.status is set");
111        }
112
113    }
114
115    @Override
116    protected Job.Status getJobStatus() throws CommandException {
117        Job.Status jobStatus = super.getJobStatus();
118        if (jobStatus == null) {
119            jobStatus = coordJob.getStatus();
120        }
121
122        return jobStatus;
123    }
124
125    @Override
126    protected boolean isTerminalState() {
127        return (coordJob.isDoneMaterialization() || coordJob.getStatus() == Job.Status.FAILED ||
128                coordJob.getStatus() == Job.Status.KILLED) && isCoordTerminalStatus(coordActionCount);
129    }
130
131    @Override
132    protected Status getTerminalStatus() {
133
134        // If all coord action is done and coord is killed, then don't change the status.
135        if (coordJob.getStatus().equals(Job.Status.KILLED)) {
136            return Job.Status.KILLED;
137
138        }
139        // If all the coordinator actions are succeeded then coordinator job should be succeeded.
140        if (coordActionCount == (getStatusCount(CoordinatorAction.Status.SUCCEEDED)
141                + getStatusCount(CoordinatorAction.Status.SKIPPED)) && coordJob.isDoneMaterialization()) {
142            return Job.Status.SUCCEEDED;
143
144        }
145        else if (coordActionCount == getStatusCount(CoordinatorAction.Status.KILLED)) {
146            // If all the coordinator actions are KILLED then coordinator job should be KILLED.
147            return Job.Status.KILLED;
148
149        }
150        else if (coordActionCount == getStatusCount(CoordinatorAction.Status.FAILED)) {
151            // If all the coordinator actions are FAILED then coordinator job should be FAILED.
152            return Job.Status.FAILED;
153
154        }
155        else {
156            return Job.Status.DONEWITHERROR;
157        }
158    }
159
160    @Override
161    protected boolean isPausedState() {
162        return coordJob.getStatus().equals(Job.Status.PAUSED)
163                || coordJob.getStatus().equals(Job.Status.PAUSEDWITHERROR);
164    }
165
166    @Override
167    protected Status getPausedState() {
168        return hasTerminatedActions() ? Job.Status.PAUSEDWITHERROR : Job.Status.PAUSED;
169    }
170
171    @Override
172    protected boolean isSuspendedState() {
173        if (coordJob.getStatus() == Job.Status.SUSPENDED || coordJob.getStatus() == Job.Status.SUSPENDEDWITHERROR) {
174            return true;
175        }
176        else {
177            return getBottomUpSuspendedState() != null;
178        }
179    }
180
181    @Override
182    protected Status getSuspendedStatus() {
183        if (coordJob.getStatus() == Job.Status.SUSPENDED || coordJob.getStatus() == Job.Status.SUSPENDEDWITHERROR) {
184            return hasTerminatedActions() ? Job.Status.SUSPENDEDWITHERROR : Job.Status.SUSPENDED;
185        }
186        else {
187            return getBottomUpSuspendedState();
188        }
189    }
190
191    @Override
192    protected boolean isRunningState() {
193        return coordJob.getStatus() != Job.Status.PREP;
194    }
195
196    @Override
197    protected Status getRunningState() {
198        return hasTerminatedActions() ? Job.Status.RUNNINGWITHERROR : Job.Status.RUNNING;
199    }
200
201    @Override
202    protected void updateJobStatus(Status coordStatus) throws JPAExecutorException, CommandException {
203        final Job.Status prevStatus = coordJob.getStatus();
204
205        boolean prevPending = coordJob.isPending();
206        if (isPending) {
207            coordJob.setPending();
208        }
209        else {
210            coordJob.resetPending();
211        }
212        boolean isPendingStateChanged = prevPending != coordJob.isPending();
213
214        // Update the Coord Job
215        if (coordJob.isTerminalStatus()
216                && (coordStatus == Job.Status.SUSPENDED || coordStatus == Job.Status.SUSPENDEDWITHERROR)) {
217            LOG.info("Coord Job [" + coordJob.getId() + "] status to " + coordStatus
218                    + " can not be updated as its already in Terminal state");
219            if (isPendingStateChanged) {
220                LOG.info("Pending for job  [" + coordJob.getId() + "] is changed to to '" + coordJob.isPending()
221                        + "' from '" + prevStatus + "'");
222                coordJob.setLastModifiedTime(new Date());
223                CoordJobQueryExecutor.getInstance().executeUpdate(
224                        CoordJobQuery.UPDATE_COORD_JOB_STATUS_PENDING_MODTIME, coordJob);
225            }
226            return;
227
228        }
229
230        // Check for backward support when RUNNINGWITHERROR, SUSPENDEDWITHERROR and PAUSEDWITHERROR is
231        // not supported
232        coordJob.setStatus(StatusUtils.getStatusIfBackwardSupportTrue(coordStatus));
233        // Backward support when coordinator namespace is 0.1
234        coordJob.setStatus(StatusUtils.getStatus(coordJob));
235        if (coordJob.getStatus() != prevStatus || isPendingStateChanged) {
236            LOG.info("Set coordinator job [" + coordJob.getId() + "] status to '" + coordJob.getStatus() + "' from '"
237                    + prevStatus + "'");
238            coordJob.setLastModifiedTime(new Date());
239            CoordJobQueryExecutor.getInstance().executeUpdate(CoordJobQuery.UPDATE_COORD_JOB_STATUS_PENDING_MODTIME,
240                    coordJob);
241        }
242        // update bundle action only when status changes in coord job
243        if (coordJob.getBundleId() != null) {
244            if (!prevStatus.equals(coordJob.getStatus())) {
245                new BundleStatusUpdateXCommand(coordJob, prevStatus).call();
246            }
247        }
248    }
249
250    /**
251     * Bottom up look for children to check the parent's status only if materialization is done and all actions are
252     * non-pending.
253     *
254     * @return the bottom up suspended state
255     */
256    protected Job.Status getBottomUpSuspendedState() {
257        if (coordJob.isDoneMaterialization() && !isPending
258                && coordActionStatus.containsKey(CoordinatorAction.Status.SUSPENDED)) {
259
260            if (coordActionCount == getStatusCount(CoordinatorAction.Status.SUSPENDED)
261                    + getStatusCount(CoordinatorAction.Status.SUCCEEDED)) {
262                return Job.Status.SUSPENDED;
263
264            }
265            else if (coordActionCount == getStatusCount(CoordinatorAction.Status.SUSPENDED)
266                    + getStatusCount(CoordinatorAction.Status.SUCCEEDED)
267                    + getStatusCount(CoordinatorAction.Status.KILLED) + getStatusCount(CoordinatorAction.Status.FAILED)
268                    + getStatusCount(CoordinatorAction.Status.TIMEDOUT)) {
269                return Job.Status.SUSPENDEDWITHERROR;
270
271            }
272        }
273        return null;
274    }
275
276    private boolean isCoordTerminalStatus(int coordActionsCount) {
277        return coordActionsCount == getStatusCount(CoordinatorAction.Status.SUCCEEDED)
278                + getStatusCount(CoordinatorAction.Status.FAILED) + getStatusCount(CoordinatorAction.Status.KILLED)
279                + getStatusCount(CoordinatorAction.Status.TIMEDOUT) + getStatusCount(CoordinatorAction.Status.SKIPPED);
280
281    }
282
283    private int getStatusCount(CoordinatorAction.Status status) {
284        int statusCount = 0;
285        if (coordActionStatus.containsKey(status)) {
286            statusCount = coordActionStatus.get(status);
287        }
288        return statusCount;
289    }
290
291    private boolean hasTerminatedActions() {
292        return coordActionStatus.containsKey(CoordinatorAction.Status.KILLED)
293                || coordActionStatus.containsKey(CoordinatorAction.Status.FAILED)
294                || coordActionStatus.containsKey(CoordinatorAction.Status.TIMEDOUT);
295    }
296
297}