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.List; 023 024import org.apache.oozie.CoordinatorActionBean; 025import org.apache.oozie.CoordinatorJobBean; 026import org.apache.oozie.ErrorCode; 027import org.apache.oozie.XException; 028import org.apache.oozie.client.CoordinatorJob; 029import org.apache.oozie.client.Job; 030import org.apache.oozie.command.CommandException; 031import org.apache.oozie.command.PreconditionException; 032import org.apache.oozie.command.ResumeTransitionXCommand; 033import org.apache.oozie.command.bundle.BundleStatusUpdateXCommand; 034import org.apache.oozie.command.wf.ResumeXCommand; 035import org.apache.oozie.executor.jpa.BatchQueryExecutor; 036import org.apache.oozie.executor.jpa.BatchQueryExecutor.UpdateEntry; 037import org.apache.oozie.executor.jpa.CoordActionQueryExecutor.CoordActionQuery; 038import org.apache.oozie.executor.jpa.CoordJobGetActionsSuspendedJPAExecutor; 039import org.apache.oozie.executor.jpa.CoordJobGetJPAExecutor; 040import org.apache.oozie.executor.jpa.JPAExecutorException; 041import org.apache.oozie.executor.jpa.CoordJobQueryExecutor.CoordJobQuery; 042import org.apache.oozie.service.EventHandlerService; 043import org.apache.oozie.service.JPAService; 044import org.apache.oozie.service.Services; 045import org.apache.oozie.util.InstrumentUtils; 046import org.apache.oozie.util.LogUtils; 047import org.apache.oozie.util.ParamChecker; 048import org.apache.oozie.util.XLog; 049 050/** 051 * Resume coordinator job and actions. 052 * 053 */ 054public class CoordResumeXCommand extends ResumeTransitionXCommand { 055 private final String jobId; 056 private CoordinatorJobBean coordJob = null; 057 private JPAService jpaService = null; 058 private boolean exceptionOccured = false; 059 CoordinatorJob.Status prevStatus; 060 061 public CoordResumeXCommand(String id) { 062 super("coord_resume", "coord_resume", 1); 063 this.jobId = ParamChecker.notEmpty(id, "id"); 064 } 065 066 @Override 067 public String getEntityKey() { 068 return jobId; 069 } 070 071 @Override 072 public String getKey() { 073 return getName() + "_" + this.jobId; 074 } 075 076 @Override 077 protected boolean isLockRequired() { 078 return true; 079 } 080 081 @Override 082 protected void loadState() throws CommandException { 083 jpaService = Services.get().get(JPAService.class); 084 if (jpaService == null) { 085 throw new CommandException(ErrorCode.E0610); 086 } 087 try { 088 coordJob = jpaService.execute(new CoordJobGetJPAExecutor(jobId)); 089 } 090 catch (JPAExecutorException e) { 091 throw new CommandException(e); 092 } 093 setJob(coordJob); 094 prevStatus = coordJob.getStatus(); 095 LogUtils.setLogInfo(coordJob); 096 } 097 098 @Override 099 protected void verifyPrecondition() throws CommandException, PreconditionException { 100 if (coordJob.getStatus() != CoordinatorJob.Status.SUSPENDED && coordJob.getStatus() != CoordinatorJob.Status.SUSPENDEDWITHERROR && coordJob.getStatus() != Job.Status.PREPSUSPENDED) { 101 throw new PreconditionException(ErrorCode.E1100, "CoordResumeXCommand not Resumed - " 102 + "job not in SUSPENDED/SUSPENDEDWITHERROR/PREPSUSPENDED state, job = " + jobId); 103 } 104 } 105 106 @Override 107 public void updateJob() { 108 InstrumentUtils.incrJobCounter(getName(), 1, getInstrumentation()); 109 coordJob.setSuspendedTime(null); 110 coordJob.setLastModifiedTime(new Date()); 111 LOG.debug("Resume coordinator job id = " + jobId + ", status = " + coordJob.getStatus() + ", pending = " 112 + coordJob.isPending()); 113 updateList.add(new UpdateEntry<CoordJobQuery>(CoordJobQuery.UPDATE_COORD_JOB_STATUS_PENDING_TIME, coordJob)); 114 } 115 116 @Override 117 public void resumeChildren() throws CommandException { 118 try { 119 // Get all suspended actions to resume them 120 List<CoordinatorActionBean> actionList = jpaService.execute(new CoordJobGetActionsSuspendedJPAExecutor( 121 jobId)); 122 123 for (CoordinatorActionBean action : actionList) { 124 if (action.getExternalId() != null) { 125 // queue a ResumeXCommand 126 queue(new ResumeXCommand(action.getExternalId())); 127 updateCoordAction(action); 128 LOG.debug( 129 "Resume coord action = [{0}], new status = [{1}], pending = [{2}] and queue ResumeXCommand for [{3}]", 130 action.getId(), action.getStatus(), action.getPending(), action.getExternalId()); 131 } 132 else { 133 updateCoordAction(action); 134 LOG.debug( 135 "Resume coord action = [{0}], new status = [{1}], pending = [{2}] and external id is null", 136 action.getId(), action.getStatus(), action.getPending()); 137 } 138 } 139 140 } 141 catch (XException ex) { 142 exceptionOccured = true; 143 throw new CommandException(ex); 144 } 145 finally { 146 if (exceptionOccured) { 147 coordJob.setStatus(CoordinatorJob.Status.FAILED); 148 coordJob.resetPending(); 149 LOG.warn("Resume children failed so fail coordinator, coordinator job id = " + jobId + ", status = " 150 + coordJob.getStatus()); 151 updateList.add(new UpdateEntry<CoordJobQuery>(CoordJobQuery.UPDATE_COORD_JOB_STATUS_PENDING_TIME, 152 coordJob)); 153 } 154 } 155 } 156 157 @Override 158 public void notifyParent() throws CommandException { 159 // update bundle action 160 if (this.coordJob.getBundleId() != null) { 161 BundleStatusUpdateXCommand bundleStatusUpdate = new BundleStatusUpdateXCommand(coordJob, prevStatus); 162 bundleStatusUpdate.call(); 163 } 164 } 165 166 @Override 167 public void performWrites() throws CommandException { 168 try { 169 BatchQueryExecutor.getInstance().executeBatchInsertUpdateDelete(null, updateList, null); 170 if (EventHandlerService.isEnabled()) { 171 // good enough to set event start time as coord's last modified time 172 // updated when set to running 173 generateEvents(coordJob, coordJob.getLastModifiedTime()); 174 } 175 } 176 catch (JPAExecutorException e) { 177 throw new CommandException(e); 178 } 179 } 180 181 private void updateCoordAction(CoordinatorActionBean action) { 182 action.setStatus(CoordinatorActionBean.Status.RUNNING); 183 action.incrementAndGetPending(); 184 action.setLastModifiedTime(new Date()); 185 updateList.add(new UpdateEntry<CoordActionQuery>(CoordActionQuery.UPDATE_COORD_ACTION_STATUS_PENDING_TIME, action)); 186 } 187 188 @Override 189 public Job getJob() { 190 return coordJob; 191 } 192}