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