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    package org.apache.oozie.command.wf;
019    
020    import java.io.IOException;
021    import java.net.URISyntaxException;
022    import java.util.ArrayList;
023    import java.util.Date;
024    import java.util.List;
025    
026    import org.apache.oozie.ErrorCode;
027    import org.apache.oozie.WorkflowActionBean;
028    import org.apache.oozie.WorkflowJobBean;
029    import org.apache.oozie.action.control.EndActionExecutor;
030    import org.apache.oozie.action.control.ForkActionExecutor;
031    import org.apache.oozie.action.control.JoinActionExecutor;
032    import org.apache.oozie.action.control.KillActionExecutor;
033    import org.apache.oozie.action.control.StartActionExecutor;
034    import org.apache.oozie.client.WorkflowJob;
035    import org.apache.oozie.client.rest.JsonBean;
036    import org.apache.oozie.command.CommandException;
037    import org.apache.oozie.command.PreconditionException;
038    import org.apache.oozie.command.coord.CoordActionUpdateXCommand;
039    import org.apache.oozie.command.wf.ActionXCommand.ActionExecutorContext;
040    import org.apache.oozie.executor.jpa.BulkUpdateInsertJPAExecutor;
041    import org.apache.oozie.executor.jpa.JPAExecutorException;
042    import org.apache.oozie.executor.jpa.WorkflowJobGetActionsJPAExecutor;
043    import org.apache.oozie.executor.jpa.WorkflowJobGetJPAExecutor;
044    import org.apache.oozie.service.HadoopAccessorException;
045    import org.apache.oozie.service.JPAService;
046    import org.apache.oozie.service.Services;
047    import org.apache.oozie.util.InstrumentUtils;
048    import org.apache.oozie.util.LogUtils;
049    import org.apache.oozie.util.ParamChecker;
050    import org.apache.oozie.workflow.WorkflowException;
051    import org.apache.oozie.workflow.WorkflowInstance;
052    import org.apache.oozie.workflow.lite.LiteWorkflowInstance;
053    
054    public class ResumeXCommand extends WorkflowXCommand<Void> {
055    
056        private String id;
057        private JPAService jpaService = null;
058        private WorkflowJobBean workflow = null;
059        private List<JsonBean> updateList = new ArrayList<JsonBean>();
060    
061        public ResumeXCommand(String id) {
062            super("resume", "resume", 1);
063            this.id = ParamChecker.notEmpty(id, "id");
064        }
065    
066        @Override
067        protected Void execute() throws CommandException {
068            try {
069                if (workflow.getStatus() == WorkflowJob.Status.SUSPENDED) {
070                    InstrumentUtils.incrJobCounter(getName(), 1, getInstrumentation());
071                    workflow.getWorkflowInstance().resume();
072                    WorkflowInstance wfInstance = workflow.getWorkflowInstance();
073                    ((LiteWorkflowInstance) wfInstance).setStatus(WorkflowInstance.Status.RUNNING);
074                    workflow.setWorkflowInstance(wfInstance);
075                    workflow.setStatus(WorkflowJob.Status.RUNNING);
076    
077    
078                    //for (WorkflowActionBean action : store.getActionsForWorkflow(id, false)) {
079                    for (WorkflowActionBean action : jpaService.execute(new WorkflowJobGetActionsJPAExecutor(id))) {
080    
081                        // Set pending flag to true for the actions that are START_RETRY or
082                        // START_MANUAL or END_RETRY or END_MANUAL
083                        if (action.isRetryOrManual()) {
084                            action.setPendingOnly();
085                            updateList.add(action);
086                        }
087    
088                        if (action.isPending()) {
089                            if (action.getStatus() == WorkflowActionBean.Status.PREP
090                                    || action.getStatus() == WorkflowActionBean.Status.START_MANUAL) {
091                                // When resuming a workflow that was programatically suspended (via ActionCheckXCommand) because of
092                                // a repeated transient error, we have to clean up the action dir
093                                if (!action.getType().equals(StartActionExecutor.TYPE) &&       // The control actions have invalid
094                                    !action.getType().equals(ForkActionExecutor.TYPE) &&        // action dir paths because they
095                                    !action.getType().equals(JoinActionExecutor.TYPE) &&        // contain ":" (colons)
096                                    !action.getType().equals(KillActionExecutor.TYPE) &&
097                                    !action.getType().equals(EndActionExecutor.TYPE)) {
098                                    ActionExecutorContext context =
099                                            new ActionXCommand.ActionExecutorContext(workflow, action, false, false);
100                                    if (context.getAppFileSystem().exists(context.getActionDir())) {
101                                        context.getAppFileSystem().delete(context.getActionDir(), true);
102                                    }
103                                }
104                                queue(new ActionStartXCommand(action.getId(), action.getType()));
105                            }
106                            else {
107                                if (action.getStatus() == WorkflowActionBean.Status.START_RETRY) {
108                                    Date nextRunTime = action.getPendingAge();
109                                    queue(new ActionStartXCommand(action.getId(), action.getType()),
110                                                  nextRunTime.getTime() - System.currentTimeMillis());
111                                }
112                                else {
113                                    if (action.getStatus() == WorkflowActionBean.Status.DONE
114                                            || action.getStatus() == WorkflowActionBean.Status.END_MANUAL) {
115                                        queue(new ActionEndXCommand(action.getId(), action.getType()));
116                                    }
117                                    else {
118                                        if (action.getStatus() == WorkflowActionBean.Status.END_RETRY) {
119                                            Date nextRunTime = action.getPendingAge();
120                                            queue(new ActionEndXCommand(action.getId(), action.getType()),
121                                                          nextRunTime.getTime() - System.currentTimeMillis());
122                                        }
123                                    }
124                                }
125                            }
126    
127                        }
128                    }
129    
130                    workflow.setLastModifiedTime(new Date());
131                    updateList.add(workflow);
132                    jpaService.execute(new BulkUpdateInsertJPAExecutor(updateList, null));
133                    queue(new NotificationXCommand(workflow));
134                }
135                return null;
136            }
137            catch (WorkflowException ex) {
138                throw new CommandException(ex);
139            }
140            catch (JPAExecutorException e) {
141                throw new CommandException(e);
142            }
143            catch (HadoopAccessorException e) {
144                throw new CommandException(e);
145            }
146            catch (IOException e) {
147                throw new CommandException(ErrorCode.E0902, e.getMessage(), e);
148            }
149            catch (URISyntaxException e) {
150                throw new CommandException(ErrorCode.E0902, e.getMessage(), e);
151            }
152            finally {
153                // update coordinator action
154                new CoordActionUpdateXCommand(workflow).call();
155            }
156        }
157    
158        @Override
159        public String getEntityKey() {
160            return id;
161        }
162    
163        @Override
164        protected boolean isLockRequired() {
165            return true;
166        }
167    
168        @Override
169        protected void loadState() throws CommandException {
170            jpaService = Services.get().get(JPAService.class);
171            if (jpaService == null) {
172                throw new CommandException(ErrorCode.E0610);
173            }
174            try {
175                workflow = jpaService.execute(new WorkflowJobGetJPAExecutor(id));
176            }
177            catch (JPAExecutorException e) {
178                throw new CommandException(e);
179            }
180            LogUtils.setLogInfo(workflow, logInfo);
181        }
182    
183        @Override
184        protected void verifyPrecondition() throws CommandException, PreconditionException {
185            if (workflow.getStatus() != WorkflowJob.Status.SUSPENDED) {
186                throw new PreconditionException(ErrorCode.E1100, "workflow's status is " + workflow.getStatusStr() + " is not SUSPENDED");
187            }
188        }
189    }