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.io.IOException;
022import java.io.StringReader;
023import java.net.URI;
024import java.util.Arrays;
025import java.util.Date;
026import java.util.List;
027
028import org.apache.hadoop.conf.Configuration;
029import org.apache.oozie.CoordinatorActionBean;
030import org.apache.oozie.CoordinatorJobBean;
031import org.apache.oozie.ErrorCode;
032import org.apache.oozie.client.CoordinatorAction;
033import org.apache.oozie.client.Job;
034import org.apache.oozie.client.OozieClient;
035import org.apache.oozie.command.CommandException;
036import org.apache.oozie.command.PreconditionException;
037import org.apache.oozie.dependency.DependencyChecker;
038import org.apache.oozie.dependency.ActionDependency;
039import org.apache.oozie.dependency.URIHandler;
040import org.apache.oozie.executor.jpa.CoordActionGetForInputCheckJPAExecutor;
041import org.apache.oozie.executor.jpa.CoordActionQueryExecutor;
042import org.apache.oozie.executor.jpa.CoordActionQueryExecutor.CoordActionQuery;
043import org.apache.oozie.executor.jpa.CoordJobGetJPAExecutor;
044import org.apache.oozie.executor.jpa.JPAExecutorException;
045import org.apache.oozie.service.CallableQueueService;
046import org.apache.oozie.service.ConfigurationService;
047import org.apache.oozie.service.EventHandlerService;
048import org.apache.oozie.service.JPAService;
049import org.apache.oozie.service.PartitionDependencyManagerService;
050import org.apache.oozie.service.RecoveryService;
051import org.apache.oozie.service.Service;
052import org.apache.oozie.service.Services;
053import org.apache.oozie.service.URIHandlerService;
054import org.apache.oozie.util.LogUtils;
055import org.apache.oozie.util.StatusUtils;
056import org.apache.oozie.util.XConfiguration;
057import org.apache.oozie.util.XLog;
058import org.apache.oozie.util.DateUtils;
059
060public class CoordPushDependencyCheckXCommand extends CoordinatorXCommand<Void> {
061    protected String actionId;
062    protected JPAService jpaService = null;
063    protected CoordinatorActionBean coordAction = null;
064    protected CoordinatorJobBean coordJob = null;
065
066    /**
067     * Property name of command re-queue interval for coordinator push check in
068     * milliseconds.
069     */
070    public static final String CONF_COORD_PUSH_CHECK_REQUEUE_INTERVAL = Service.CONF_PREFIX
071            + "coord.push.check.requeue.interval";
072    private boolean registerForNotification;
073    private boolean removeAvailDependencies;
074
075    public CoordPushDependencyCheckXCommand(String actionId) {
076        this(actionId, false, true);
077    }
078
079    public CoordPushDependencyCheckXCommand(String actionId, boolean registerForNotification) {
080        this(actionId, registerForNotification, !registerForNotification);
081    }
082
083    public CoordPushDependencyCheckXCommand(String actionId, boolean registerForNotification,
084            boolean removeAvailDependencies) {
085        super("coord_push_dep_check", "coord_push_dep_check", 0);
086        this.actionId = actionId;
087        this.registerForNotification = registerForNotification;
088        this.removeAvailDependencies = removeAvailDependencies;
089    }
090
091    protected CoordPushDependencyCheckXCommand(String actionName, String actionId) {
092        super(actionName, actionName, 0);
093        this.actionId = actionId;
094    }
095
096    @Override
097    protected void setLogInfo() {
098        LogUtils.setLogInfo(actionId);
099    }
100
101    @Override
102    protected Void execute() throws CommandException {
103        String pushMissingDeps = coordAction.getPushMissingDependencies();
104        if (pushMissingDeps == null || pushMissingDeps.length() == 0) {
105            LOG.info("Nothing to check. Empty push missing dependency");
106        }
107        else {
108            String[] missingDepsArray = DependencyChecker.dependenciesAsArray(pushMissingDeps);
109            LOG.info("First Push missing dependency is [{0}] ", missingDepsArray[0]);
110            LOG.trace("Push missing dependencies are [{0}] ", pushMissingDeps);
111            if (registerForNotification) {
112                LOG.debug("Register for notifications is true");
113            }
114
115            try {
116                Configuration actionConf = null;
117                try {
118                    actionConf = new XConfiguration(new StringReader(coordAction.getRunConf()));
119                }
120                catch (IOException e) {
121                    throw new CommandException(ErrorCode.E1307, e.getMessage(), e);
122                }
123
124                // Check all dependencies during materialization to avoid registering in the cache.
125                // But check only first missing one afterwards similar to
126                // CoordActionInputCheckXCommand for efficiency. listPartitions is costly.
127                ActionDependency actionDep = DependencyChecker.checkForAvailability(missingDepsArray, actionConf,
128                        !registerForNotification);
129
130                boolean isChangeInDependency = true;
131                boolean timeout = false;
132                if (actionDep.getMissingDependencies().size() == 0) {
133                    // All push-based dependencies are available
134                    onAllPushDependenciesAvailable();
135                }
136                else {
137                    if (actionDep.getMissingDependencies().size() == missingDepsArray.length) {
138                        isChangeInDependency = false;
139                    }
140                    else {
141                        String stillMissingDeps = DependencyChecker.dependenciesAsString(actionDep
142                                .getMissingDependencies());
143                        coordAction.setPushMissingDependencies(stillMissingDeps);
144                    }
145                    // Checking for timeout
146                    timeout = isTimeout();
147                    if (timeout) {
148                        queue(new CoordActionTimeOutXCommand(coordAction, coordJob.getUser(), coordJob.getAppName()));
149                    }
150                    else {
151                        queue(new CoordPushDependencyCheckXCommand(coordAction.getId()),
152                                getCoordPushCheckRequeueInterval());
153                    }
154                }
155
156                updateCoordAction(coordAction, isChangeInDependency);
157                if (registerForNotification) {
158                    registerForNotification(actionDep.getMissingDependencies(), actionConf);
159                }
160                if (removeAvailDependencies) {
161                    unregisterAvailableDependencies(actionDep.getAvailableDependencies());
162                }
163                if (timeout) {
164                    unregisterMissingDependencies(actionDep.getMissingDependencies(), actionId);
165                }
166            }
167            catch (Exception e) {
168                final CallableQueueService callableQueueService = Services.get().get(CallableQueueService.class);
169                if (isTimeout()) {
170                    LOG.debug("Queueing timeout command");
171                    // XCommand.queue() will not work when there is a Exception
172                    callableQueueService.queue(new CoordActionTimeOutXCommand(coordAction, coordJob.getUser(), coordJob.getAppName()));
173                    unregisterMissingDependencies(Arrays.asList(missingDepsArray), actionId);
174                }
175                else if (coordAction.getMissingDependencies() != null
176                        && coordAction.getMissingDependencies().length() > 0) {
177                    // Queue again on exception as RecoveryService will not queue this again with
178                    // the action being updated regularly by CoordActionInputCheckXCommand
179                    callableQueueService.queue(new CoordPushDependencyCheckXCommand(coordAction.getId(),
180                            registerForNotification, removeAvailDependencies),
181                            Services.get().getConf().getInt(RecoveryService.CONF_COORD_OLDER_THAN, 600) * 1000);
182                }
183                throw new CommandException(ErrorCode.E1021, e.getMessage(), e);
184            }
185        }
186        return null;
187    }
188
189    /**
190     * Return the re-queue interval for coord push dependency check
191     * @return
192     */
193    public long getCoordPushCheckRequeueInterval() {
194        long requeueInterval = ConfigurationService.getLong(CONF_COORD_PUSH_CHECK_REQUEUE_INTERVAL);
195        return requeueInterval;
196    }
197
198    /**
199     * Returns true if timeout period has been reached
200     *
201     * @return true if it is time for timeout else false
202     */
203    protected boolean isTimeout() {
204        long waitingTime = (new Date().getTime() - Math.max(coordAction.getNominalTime().getTime(), coordAction
205                .getCreatedTime().getTime()))
206                / (60 * 1000);
207        int timeOut = coordAction.getTimeOut();
208        return (timeOut >= 0) && (waitingTime > timeOut);
209    }
210
211    protected void onAllPushDependenciesAvailable() throws CommandException {
212        coordAction.setPushMissingDependencies("");
213        Services.get().get(PartitionDependencyManagerService.class)
214                .removeCoordActionWithDependenciesAvailable(coordAction.getId());
215        if (coordAction.getMissingDependencies() == null || coordAction.getMissingDependencies().length() == 0) {
216            Date nominalTime = coordAction.getNominalTime();
217            Date currentTime = new Date();
218            // The action should become READY only if current time > nominal time;
219            // CoordActionInputCheckXCommand will take care of moving it to READY when it is nominal time.
220            if (nominalTime.compareTo(currentTime) > 0) {
221                LOG.info("[" + actionId + "]::ActionInputCheck:: nominal Time is newer than current time. Current="
222                        + DateUtils.formatDateOozieTZ(currentTime) + ", nominal=" + DateUtils.formatDateOozieTZ(nominalTime));
223            }
224            else {
225                String actionXml = resolveCoordConfiguration();
226                coordAction.setActionXml(actionXml);
227                coordAction.setStatus(CoordinatorAction.Status.READY);
228                // pass jobID to the CoordActionReadyXCommand
229                queue(new CoordActionReadyXCommand(coordAction.getJobId()), 100);
230            }
231        }
232        else if (isTimeout()) {
233            // If it is timeout and all push dependencies are available but still some unresolved
234            // missing dependencies queue CoordActionInputCheckXCommand now. Else it will have to
235            // wait till RecoveryService kicks in
236            queue(new CoordActionInputCheckXCommand(coordAction.getId(), coordAction.getJobId()));
237        }
238    }
239
240    private String resolveCoordConfiguration() throws CommandException {
241        try {
242            Configuration actionConf = new XConfiguration(new StringReader(coordAction.getRunConf()));
243            StringBuilder actionXml = new StringBuilder(coordAction.getActionXml());
244            String newActionXml = CoordActionInputCheckXCommand.resolveCoordConfiguration(actionXml, actionConf,
245                    actionId);
246            actionXml.replace(0, actionXml.length(), newActionXml);
247            return actionXml.toString();
248        }
249        catch (Exception e) {
250            throw new CommandException(ErrorCode.E1021, e.getMessage(), e);
251        }
252    }
253
254    protected void updateCoordAction(CoordinatorActionBean coordAction, boolean isChangeInDependency)
255            throws CommandException {
256        coordAction.setLastModifiedTime(new Date());
257        if (jpaService != null) {
258            try {
259                if (isChangeInDependency) {
260                    CoordActionQueryExecutor.getInstance().executeUpdate(
261                            CoordActionQuery.UPDATE_COORD_ACTION_FOR_PUSH_INPUTCHECK, coordAction);
262                    if (EventHandlerService.isEnabled() && coordAction.getStatus() != CoordinatorAction.Status.READY) {
263                        // since event is not to be generated unless action
264                        // RUNNING via StartX
265                        generateEvent(coordAction, coordJob.getUser(), coordJob.getAppName(), null);
266                    }
267                }
268                else {
269                    CoordActionQueryExecutor.getInstance().executeUpdate(
270                            CoordActionQuery.UPDATE_COORD_ACTION_FOR_MODIFIED_DATE, coordAction);
271                }
272            }
273            catch (JPAExecutorException jex) {
274                throw new CommandException(ErrorCode.E1021, jex.getMessage(), jex);
275            }
276        }
277    }
278
279    private void registerForNotification(List<String> missingDeps, Configuration actionConf) {
280        URIHandlerService uriService = Services.get().get(URIHandlerService.class);
281        String user = actionConf.get(OozieClient.USER_NAME, OozieClient.USER_NAME);
282        for (String missingDep : missingDeps) {
283            try {
284                URI missingURI = new URI(missingDep);
285                URIHandler handler = uriService.getURIHandler(missingURI);
286                handler.registerForNotification(missingURI, actionConf, user, actionId);
287                    LOG.debug("Registered uri [{0}] for notifications", missingURI);
288            }
289            catch (Exception e) {
290                LOG.warn("Exception while registering uri [{0}] for notifications", missingDep, e);
291            }
292        }
293    }
294
295    private void unregisterAvailableDependencies(List<String> availableDeps) {
296        URIHandlerService uriService = Services.get().get(URIHandlerService.class);
297        for (String availableDep : availableDeps) {
298            try {
299                URI availableURI = new URI(availableDep);
300                URIHandler handler = uriService.getURIHandler(availableURI);
301                if (handler.unregisterFromNotification(availableURI, actionId)) {
302                    LOG.debug("Successfully unregistered uri [{0}] from notifications", availableURI);
303                }
304                else {
305                    LOG.warn("Unable to unregister uri [{0}] from notifications", availableURI);
306                }
307            }
308            catch (Exception e) {
309                LOG.warn("Exception while unregistering uri [{0}] from notifications", availableDep, e);
310            }
311        }
312    }
313
314    public static void unregisterMissingDependencies(List<String> missingDeps, String actionId) {
315        final XLog LOG = XLog.getLog(CoordPushDependencyCheckXCommand.class);
316        URIHandlerService uriService = Services.get().get(URIHandlerService.class);
317        for (String missingDep : missingDeps) {
318            try {
319                URI missingURI = new URI(missingDep);
320                URIHandler handler = uriService.getURIHandler(missingURI);
321                if (handler.unregisterFromNotification(missingURI, actionId)) {
322                    LOG.debug("Successfully unregistered uri [{0}] from notifications", missingURI);
323                }
324                else {
325                    LOG.warn("Unable to unregister uri [{0}] from notifications", missingURI);
326                }
327            }
328            catch (Exception e) {
329                LOG.warn("Exception while unregistering uri [{0}] from notifications", missingDep, e);
330            }
331        }
332    }
333
334    @Override
335    public String getEntityKey() {
336        return actionId.substring(0, actionId.indexOf("@"));
337    }
338
339    @Override
340    public String getKey(){
341        return getName() + "_" + actionId;
342    }
343
344    @Override
345    protected boolean isLockRequired() {
346        return true;
347    }
348
349    @Override
350    protected void loadState() throws CommandException {
351        jpaService = Services.get().get(JPAService.class);
352        try {
353            coordAction = jpaService.execute(new CoordActionGetForInputCheckJPAExecutor(actionId));
354            if (coordAction != null) {
355                coordJob = jpaService.execute(new CoordJobGetJPAExecutor(coordAction.getJobId()));
356                LogUtils.setLogInfo(coordAction);
357            }
358            else {
359                throw new CommandException(ErrorCode.E0605, actionId);
360            }
361        }
362        catch (JPAExecutorException je) {
363            throw new CommandException(je);
364        }
365    }
366
367    @Override
368    protected void verifyPrecondition() throws CommandException, PreconditionException {
369        if (coordAction.getStatus() != CoordinatorActionBean.Status.WAITING) {
370            throw new PreconditionException(ErrorCode.E1100, "[" + actionId
371                    + "]::CoordPushDependencyCheck:: Ignoring action. Should be in WAITING state, but state="
372                    + coordAction.getStatus());
373        }
374
375        // if eligible to do action input check when running with backward
376        // support is true
377        if (StatusUtils.getStatusForCoordActionInputCheck(coordJob)) {
378            return;
379        }
380
381        if (coordJob.getStatus() != Job.Status.RUNNING && coordJob.getStatus() != Job.Status.RUNNINGWITHERROR
382                && coordJob.getStatus() != Job.Status.PAUSED && coordJob.getStatus() != Job.Status.PAUSEDWITHERROR) {
383            throw new PreconditionException(ErrorCode.E1100, "[" + actionId
384                    + "]::CoordPushDependencyCheck:: Ignoring action."
385                    + " Coordinator job is not in RUNNING/RUNNINGWITHERROR/PAUSED/PAUSEDWITHERROR state, but state="
386                    + coordJob.getStatus());
387        }
388    }
389
390}