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;
024import java.util.Map;
025
026import org.apache.oozie.BundleActionBean;
027import org.apache.oozie.BundleJobBean;
028import org.apache.oozie.CoordinatorJobBean;
029import org.apache.oozie.XException;
030import org.apache.oozie.client.Job;
031import org.apache.oozie.client.rest.RestConstants;
032import org.apache.oozie.command.CommandException;
033import org.apache.oozie.command.RerunTransitionXCommand;
034import org.apache.oozie.command.coord.CoordRerunXCommand;
035import org.apache.oozie.executor.jpa.BatchQueryExecutor;
036import org.apache.oozie.executor.jpa.BundleActionQueryExecutor;
037import org.apache.oozie.executor.jpa.BundleJobQueryExecutor;
038import org.apache.oozie.executor.jpa.BundleJobQueryExecutor.BundleJobQuery;
039import org.apache.oozie.executor.jpa.CoordJobQueryExecutor;
040import org.apache.oozie.executor.jpa.CoordJobQueryExecutor.CoordJobQuery;
041import org.apache.oozie.executor.jpa.JPAExecutorException;
042import org.apache.oozie.executor.jpa.BundleActionQueryExecutor.BundleActionQuery;
043import org.apache.oozie.executor.jpa.BatchQueryExecutor.UpdateEntry;
044import org.apache.oozie.util.DateUtils;
045import org.apache.oozie.util.LogUtils;
046import org.apache.oozie.util.ParamChecker;
047import org.apache.oozie.util.XLog;
048
049/**
050 * Rerun bundle coordinator jobs by a list of coordinator names or dates. User can specify if refresh or noCleanup.
051 * <p/>
052 * The "refresh" is used to indicate if user wants to refresh an action's input/outpur dataset urls
053 * <p/>
054 * The "noCleanup" is used to indicate if user wants to cleanup output events for given rerun actions
055 */
056public class BundleRerunXCommand extends RerunTransitionXCommand<Void> {
057
058    private final String coordScope;
059    private final String dateScope;
060    private final boolean refresh;
061    private final boolean noCleanup;
062    private BundleJobBean bundleJob;
063    private List<BundleActionBean> bundleActions;
064    protected boolean prevPending;
065
066    /**
067     * The constructor for class {@link BundleRerunXCommand}
068     *
069     * @param jobId the bundle job id
070     * @param coordScope the rerun scope for coordinator job names separated by ","
071     * @param dateScope the rerun scope for coordinator nominal times separated by ","
072     * @param refresh true if user wants to refresh input/outpur dataset urls
073     * @param noCleanup false if user wants to cleanup output events for given rerun actions
074     */
075    public BundleRerunXCommand(String jobId, String coordScope, String dateScope, boolean refresh, boolean noCleanup) {
076        super("bundle_rerun", "bundle_rerun", 1);
077        this.jobId = ParamChecker.notEmpty(jobId, "jobId");
078        this.coordScope = coordScope;
079        this.dateScope = dateScope;
080        this.refresh = refresh;
081        this.noCleanup = noCleanup;
082    }
083
084    /* (non-Javadoc)
085     * @see org.apache.oozie.command.XCommand#loadState()
086     */
087    @Override
088    protected void loadState() throws CommandException {
089        try {
090            this.bundleJob = BundleJobQueryExecutor.getInstance().get(BundleJobQuery.GET_BUNDLE_JOB, jobId);
091            this.bundleActions = BundleActionQueryExecutor.getInstance().getList(
092                    BundleActionQuery.GET_BUNDLE_ACTIONS_STATUS_UNIGNORED_FOR_BUNDLE, jobId);
093            LogUtils.setLogInfo(bundleJob);
094            super.setJob(bundleJob);
095            prevPending = bundleJob.isPending();
096        }
097        catch (XException ex) {
098            throw new CommandException(ex);
099        }
100
101    }
102
103    /* (non-Javadoc)
104     * @see org.apache.oozie.command.RerunTransitionXCommand#rerunChildren()
105     */
106    @Override
107    public void rerunChildren() throws CommandException {
108        boolean isUpdateActionDone = false;
109        Map<String, BundleActionBean> coordNameToBAMapping = new HashMap<String, BundleActionBean>();
110        if (bundleActions != null) {
111            for (BundleActionBean action : bundleActions) {
112                if (action.getCoordName() != null) {
113                    coordNameToBAMapping.put(action.getCoordName(), action);
114                }
115            }
116        }
117
118        if (coordScope != null && !coordScope.isEmpty()) {
119            String[] list = coordScope.split(",");
120            for (String coordName : list) {
121                coordName = coordName.trim();
122                if (coordNameToBAMapping.keySet().contains(coordName)) {
123                    String coordId = coordNameToBAMapping.get(coordName).getCoordId();
124                    if (coordId == null) {
125                        LOG.info("No coord id found. Therefore, nothing to queue for coord rerun for coordname: " + coordName);
126                        continue;
127                    }
128                    CoordinatorJobBean coordJob = getCoordJob(coordId);
129
130                    String rerunDateScope;
131                    if (dateScope != null && !dateScope.isEmpty()) {
132                        rerunDateScope = dateScope;
133                    }
134                    else {
135                        String coordStart = DateUtils.formatDateOozieTZ(coordJob.getStartTime());
136                        String coordEnd = DateUtils.formatDateOozieTZ(coordJob.getEndTime());
137                        rerunDateScope = coordStart + "::" + coordEnd;
138                    }
139                    LOG.debug("Queuing rerun range [" + rerunDateScope + "] for coord id " + coordId + " of bundle "
140                            + bundleJob.getId());
141                    queue(new CoordRerunXCommand(coordId, RestConstants.JOB_COORD_SCOPE_DATE, rerunDateScope, refresh,
142                            noCleanup, false));
143                    updateBundleAction(coordNameToBAMapping.get(coordName));
144                    isUpdateActionDone = true;
145                }
146                else {
147                    LOG.info("Rerun for coord " + coordName + " NOT performed because it is not in bundle ", bundleJob.getId());
148                }
149            }
150        }
151        else if (dateScope != null && !dateScope.isEmpty()) {
152            if (bundleActions != null) {
153                for (BundleActionBean action : bundleActions) {
154                    if (action.getCoordId() == null) {
155                        LOG.info("No coord id found. Therefore nothing to queue for coord rerun with coord name "
156                                + action.getCoordName());
157                        continue;
158                    }
159                    LOG.debug("Queuing rerun range [" + dateScope + "] for coord id " + action.getCoordId() + " of bundle "
160                            + bundleJob.getId());
161                    queue(new CoordRerunXCommand(action.getCoordId(), RestConstants.JOB_COORD_SCOPE_DATE, dateScope,
162                            refresh, noCleanup, false));
163                    updateBundleAction(action);
164                    isUpdateActionDone = true;
165                }
166            }
167        }
168        if (!isUpdateActionDone) {
169            transitToPrevious();
170        }
171        LOG.info("Rerun coord jobs for the bundle=[{0}]", jobId);
172    }
173
174    private final void transitToPrevious() throws CommandException {
175        bundleJob.setStatus(getPrevStatus());
176        if (!prevPending) {
177            bundleJob.resetPending();
178        }
179        else {
180            bundleJob.setPending();
181        }
182        updateJob();
183    }
184
185    /**
186     * Update bundle action
187     *
188     * @param action the bundle action
189     * @throws CommandException thrown if failed to update bundle action
190     */
191    private void updateBundleAction(BundleActionBean action) {
192        action.incrementAndGetPending();
193        action.setLastModifiedTime(new Date());
194        updateList.add(new UpdateEntry<BundleActionQuery>(BundleActionQuery.UPDATE_BUNDLE_ACTION_PENDING_MODTIME, action));
195    }
196
197    /* (non-Javadoc)
198     * @see org.apache.oozie.command.TransitionXCommand#updateJob()
199     */
200    @Override
201    public void updateJob() {
202        // rerun a paused bundle job will keep job status at paused and pending at previous pending
203        if (getPrevStatus() != null) {
204            Job.Status bundleJobStatus = getPrevStatus();
205            if (bundleJobStatus.equals(Job.Status.PAUSED) || bundleJobStatus.equals(Job.Status.PAUSEDWITHERROR)) {
206                bundleJob.setStatus(bundleJobStatus);
207                if (prevPending) {
208                    bundleJob.setPending();
209                }
210                else {
211                    bundleJob.resetPending();
212                }
213            }
214        }
215        updateList.add(new UpdateEntry<BundleJobQuery>(BundleJobQuery.UPDATE_BUNDLE_JOB_STATUS_PENDING, bundleJob));
216    }
217
218    /* (non-Javadoc)
219     * @see org.apache.oozie.command.RerunTransitionXCommand#performWrites()
220     */
221    @Override
222    public void performWrites() throws CommandException {
223        try {
224            BatchQueryExecutor.getInstance().executeBatchInsertUpdateDelete(null, updateList, null);
225        }
226        catch (JPAExecutorException e) {
227            throw new CommandException(e);
228        }
229    }
230
231    /* (non-Javadoc)
232     * @see org.apache.oozie.command.XCommand#getEntityKey()
233     */
234    @Override
235    public String getEntityKey() {
236        return jobId;
237    }
238
239    /* (non-Javadoc)
240     * @see org.apache.oozie.command.XCommand#isLockRequired()
241     */
242    @Override
243    protected boolean isLockRequired() {
244        return true;
245    }
246
247    /*
248     * (non-Javadoc)
249     * @see org.apache.oozie.command.TransitionXCommand#getJob()
250     */
251    @Override
252    public Job getJob() {
253        return bundleJob;
254    }
255
256    /* (non-Javadoc)
257     * @see org.apache.oozie.command.TransitionXCommand#notifyParent()
258     */
259    @Override
260    public void notifyParent() throws CommandException {
261
262    }
263
264    /* (non-Javadoc)
265     * @see org.apache.oozie.command.RerunTransitionXCommand#getLog()
266     */
267    @Override
268    public XLog getLog() {
269        return LOG;
270    }
271
272    private final CoordinatorJobBean getCoordJob(String coordId) throws CommandException {
273        try {
274            CoordinatorJobBean job = CoordJobQueryExecutor.getInstance().get(CoordJobQuery.GET_COORD_JOB, coordId);
275            return job;
276        }
277        catch (JPAExecutorException je) {
278            throw new CommandException(je);
279        }
280
281    }
282
283}