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.net.URISyntaxException;
025import java.text.ParseException;
026import java.util.Calendar;
027import java.util.Date;
028import java.util.List;
029
030import org.apache.hadoop.conf.Configuration;
031import org.apache.hadoop.security.AccessControlException;
032import org.apache.oozie.CoordinatorActionBean;
033import org.apache.oozie.CoordinatorJobBean;
034import org.apache.oozie.ErrorCode;
035import org.apache.oozie.client.CoordinatorAction;
036import org.apache.oozie.client.Job;
037import org.apache.oozie.client.OozieClient;
038import org.apache.oozie.command.CommandException;
039import org.apache.oozie.command.PreconditionException;
040import org.apache.oozie.coord.CoordELEvaluator;
041import org.apache.oozie.coord.CoordELFunctions;
042import org.apache.oozie.coord.TimeUnit;
043import org.apache.oozie.dependency.URIHandler;
044import org.apache.oozie.dependency.URIHandlerException;
045import org.apache.oozie.executor.jpa.CoordActionGetForInputCheckJPAExecutor;
046import org.apache.oozie.executor.jpa.CoordActionQueryExecutor;
047import org.apache.oozie.executor.jpa.CoordActionQueryExecutor.CoordActionQuery;
048import org.apache.oozie.executor.jpa.CoordJobQueryExecutor;
049import org.apache.oozie.executor.jpa.CoordJobQueryExecutor.CoordJobQuery;
050import org.apache.oozie.executor.jpa.JPAExecutorException;
051import org.apache.oozie.service.CallableQueueService;
052import org.apache.oozie.service.ConfigurationService;
053import org.apache.oozie.service.EventHandlerService;
054import org.apache.oozie.service.JPAService;
055import org.apache.oozie.service.Service;
056import org.apache.oozie.service.Services;
057import org.apache.oozie.service.URIHandlerService;
058import org.apache.oozie.util.DateUtils;
059import org.apache.oozie.util.ELEvaluator;
060import org.apache.oozie.util.LogUtils;
061import org.apache.oozie.util.ParamChecker;
062import org.apache.oozie.util.StatusUtils;
063import org.apache.oozie.util.XConfiguration;
064import org.apache.oozie.util.XLog;
065import org.apache.oozie.util.XmlUtils;
066import org.jdom.Element;
067
068/**
069 * The command to check if an action's data input paths exist in the file system.
070 */
071public class CoordActionInputCheckXCommand extends CoordinatorXCommand<Void> {
072
073    public static final String COORD_EXECUTION_NONE_TOLERANCE = "oozie.coord.execution.none.tolerance";
074
075    private final String actionId;
076    /**
077     * Property name of command re-queue interval for coordinator action input check in
078     * milliseconds.
079     */
080    public static final String CONF_COORD_INPUT_CHECK_REQUEUE_INTERVAL = Service.CONF_PREFIX
081            + "coord.input.check.requeue.interval";
082    private CoordinatorActionBean coordAction = null;
083    private CoordinatorJobBean coordJob = null;
084    private JPAService jpaService = null;
085    private String jobId = null;
086
087    public CoordActionInputCheckXCommand(String actionId, String jobId) {
088        super("coord_action_input", "coord_action_input", 1);
089        this.actionId = ParamChecker.notEmpty(actionId, "actionId");
090        this.jobId = jobId;
091    }
092
093    @Override
094    protected void setLogInfo() {
095        LogUtils.setLogInfo(actionId);
096    }
097
098    @Override
099    protected Void execute() throws CommandException {
100        LOG.debug("[" + actionId + "]::ActionInputCheck:: Action is in WAITING state.");
101
102        // this action should only get processed if current time > nominal time;
103        // otherwise, requeue this action for delay execution;
104        Date nominalTime = coordAction.getNominalTime();
105        Date currentTime = new Date();
106        if (nominalTime.compareTo(currentTime) > 0) {
107            queue(new CoordActionInputCheckXCommand(coordAction.getId(), coordAction.getJobId()), Math.max((nominalTime.getTime() - currentTime
108                    .getTime()), getCoordInputCheckRequeueInterval()));
109            updateCoordAction(coordAction, false);
110            LOG.info("[" + actionId
111                    + "]::ActionInputCheck:: nominal Time is newer than current time, so requeue and wait. Current="
112                    + DateUtils.formatDateOozieTZ(currentTime) + ", nominal=" + DateUtils.formatDateOozieTZ(nominalTime));
113
114            return null;
115        }
116
117        StringBuilder actionXml = new StringBuilder(coordAction.getActionXml());
118        boolean isChangeInDependency = false;
119        try {
120            Configuration actionConf = new XConfiguration(new StringReader(coordAction.getRunConf()));
121            Date now = new Date();
122            if (coordJob.getExecutionOrder().equals(CoordinatorJobBean.Execution.LAST_ONLY)) {
123                Date nextNominalTime = CoordCommandUtils.computeNextNominalTime(coordJob, coordAction);
124                if (nextNominalTime != null) {
125                    // If the current time is after the next action's nominal time, then we've passed the window where this action
126                    // should be started; so set it to SKIPPED
127                    if (now.after(nextNominalTime)) {
128                        LOG.info("LAST_ONLY execution: Preparing to skip action [{0}] because the current time [{1}] is later than "
129                                + "the nominal time [{2}] of the next action]", coordAction.getId(),
130                                DateUtils.formatDateOozieTZ(now), DateUtils.formatDateOozieTZ(nextNominalTime));
131                        queue(new CoordActionSkipXCommand(coordAction, coordJob.getUser(), coordJob.getAppName()));
132                        return null;
133                    } else {
134                        LOG.debug("LAST_ONLY execution: Not skipping action [{0}] because the current time [{1}] is earlier than "
135                                + "the nominal time [{2}] of the next action]", coordAction.getId(),
136                                DateUtils.formatDateOozieTZ(now), DateUtils.formatDateOozieTZ(nextNominalTime));
137                    }
138                }
139            }
140            else if (coordJob.getExecutionOrder().equals(CoordinatorJobBean.Execution.NONE)) {
141                // If the current time is after the nominal time of this action plus some tolerance,
142                // then we've passed the window where this action should be started; so set it to SKIPPED
143                Calendar cal = Calendar.getInstance(DateUtils.getTimeZone(coordJob.getTimeZone()));
144                cal.setTime(nominalTime);
145                int tolerance = ConfigurationService.getInt(COORD_EXECUTION_NONE_TOLERANCE);
146                cal.add(Calendar.MINUTE, tolerance);
147                if (now.after(cal.getTime())) {
148                    LOG.info("NONE execution: Preparing to skip action [{0}] because the current time [{1}] is more than [{2}]"
149                            + " minutes later than the nominal time [{3}] of the current action]", coordAction.getId(),
150                            DateUtils.formatDateOozieTZ(now), tolerance, DateUtils.formatDateOozieTZ(nominalTime));
151                    queue(new CoordActionSkipXCommand(coordAction, coordJob.getUser(), coordJob.getAppName()));
152                    return null;
153                } else {
154                    LOG.debug("NONE execution: Not skipping action [{0}] because the current time [{1}] is earlier than [{2}]"
155                            + " minutes later than the nominal time [{3}] of the current action]", coordAction.getId(),
156                            DateUtils.formatDateOozieTZ(now), tolerance, DateUtils.formatDateOozieTZ(coordAction.getNominalTime()));
157                }
158            }
159
160            StringBuilder existList = new StringBuilder();
161            StringBuilder nonExistList = new StringBuilder();
162            StringBuilder nonResolvedList = new StringBuilder();
163            String firstMissingDependency = "";
164            String missingDeps = coordAction.getMissingDependencies();
165            CoordCommandUtils.getResolvedList(missingDeps, nonExistList, nonResolvedList);
166
167            // For clarity regarding which is the missing dependency in synchronous order
168            // instead of printing entire list, some of which, may be available
169            if(nonExistList.length() > 0) {
170                firstMissingDependency = nonExistList.toString().split(CoordELFunctions.INSTANCE_SEPARATOR)[0];
171            }
172            LOG.info("[" + actionId + "]::CoordActionInputCheck:: Missing deps:" + firstMissingDependency + " "
173                    + nonResolvedList.toString());
174            // Updating the list of data dependencies that are available and those that are yet not
175            boolean status = checkInput(actionXml, existList, nonExistList, actionConf);
176            String pushDeps = coordAction.getPushMissingDependencies();
177            // Resolve latest/future only when all current missingDependencies and
178            // pushMissingDependencies are met
179            if (status && nonResolvedList.length() > 0) {
180                status = (pushDeps == null || pushDeps.length() == 0) ? checkUnResolvedInput(actionXml, actionConf)
181                        : false;
182            }
183            coordAction.setLastModifiedTime(currentTime);
184            coordAction.setActionXml(actionXml.toString());
185            if (nonResolvedList.length() > 0 && status == false) {
186                nonExistList.append(CoordCommandUtils.RESOLVED_UNRESOLVED_SEPARATOR).append(nonResolvedList);
187            }
188            String nonExistListStr = nonExistList.toString();
189            if (!nonExistListStr.equals(missingDeps) || missingDeps.isEmpty()) {
190                // missingDeps null or empty means action should become READY
191                isChangeInDependency = true;
192                coordAction.setMissingDependencies(nonExistListStr);
193            }
194            if (status && (pushDeps == null || pushDeps.length() == 0)) {
195                String newActionXml = resolveCoordConfiguration(actionXml, actionConf, actionId);
196                actionXml.replace(0, actionXml.length(), newActionXml);
197                coordAction.setActionXml(actionXml.toString());
198                coordAction.setStatus(CoordinatorAction.Status.READY);
199                updateCoordAction(coordAction, true);
200                new CoordActionReadyXCommand(coordAction.getJobId()).call(getEntityKey());
201            }
202            else if (!isTimeout(currentTime)) {
203                if (status == false) {
204                    queue(new CoordActionInputCheckXCommand(coordAction.getId(), coordAction.getJobId()),
205                            getCoordInputCheckRequeueInterval());
206                }
207                updateCoordAction(coordAction, isChangeInDependency);
208            }
209            else {
210                if (!nonExistListStr.isEmpty() && pushDeps == null || pushDeps.length() == 0) {
211                    queue(new CoordActionTimeOutXCommand(coordAction, coordJob.getUser(), coordJob.getAppName()));
212                }
213                else {
214                    // Let CoordPushDependencyCheckXCommand queue the timeout
215                    queue(new CoordPushDependencyCheckXCommand(coordAction.getId()));
216                }
217                updateCoordAction(coordAction, isChangeInDependency);
218            }
219        }
220        catch (AccessControlException e) {
221            LOG.error("Permission error in ActionInputCheck", e);
222            if (isTimeout(currentTime)) {
223                LOG.debug("Queueing timeout command");
224                Services.get().get(CallableQueueService.class)
225                        .queue(new CoordActionTimeOutXCommand(coordAction, coordJob.getUser(), coordJob.getAppName()));
226            }
227            else {
228                // Requeue InputCheckCommand for permission denied error with longer interval
229                Services.get()
230                        .get(CallableQueueService.class)
231                        .queue(new CoordActionInputCheckXCommand(coordAction.getId(), coordAction.getJobId()),
232                                2 * getCoordInputCheckRequeueInterval());
233            }
234            updateCoordAction(coordAction, isChangeInDependency);
235        }
236        catch (Exception e) {
237            if (isTimeout(currentTime)) {
238                LOG.debug("Queueing timeout command");
239                // XCommand.queue() will not work when there is a Exception
240                Services.get().get(CallableQueueService.class)
241                        .queue(new CoordActionTimeOutXCommand(coordAction, coordJob.getUser(), coordJob.getAppName()));
242            }
243            updateCoordAction(coordAction, isChangeInDependency);
244            throw new CommandException(ErrorCode.E1021, e.getMessage(), e);
245        }
246        return null;
247    }
248
249
250    static String resolveCoordConfiguration(StringBuilder actionXml, Configuration actionConf, String actionId) throws Exception {
251        Element eAction = XmlUtils.parseXml(actionXml.toString());
252        ELEvaluator eval = CoordELEvaluator.createDataEvaluator(eAction, actionConf, actionId);
253        materializeDataProperties(eAction, actionConf, eval);
254        return XmlUtils.prettyPrint(eAction).toString();
255    }
256
257    private boolean isTimeout(Date currentTime) {
258        long waitingTime = (currentTime.getTime() - Math.max(coordAction.getNominalTime().getTime(), coordAction
259                .getCreatedTime().getTime()))
260                / (60 * 1000);
261        int timeOut = coordAction.getTimeOut();
262        return (timeOut >= 0) && (waitingTime > timeOut);
263    }
264
265    private void updateCoordAction(CoordinatorActionBean coordAction, boolean isChangeInDependency)
266            throws CommandException {
267        coordAction.setLastModifiedTime(new Date());
268        if (jpaService != null) {
269            try {
270                if (isChangeInDependency) {
271                    CoordActionQueryExecutor.getInstance().executeUpdate(
272                            CoordActionQuery.UPDATE_COORD_ACTION_FOR_INPUTCHECK, coordAction);
273                    if (EventHandlerService.isEnabled() && coordAction.getStatus() != CoordinatorAction.Status.READY) {
274                        // since event is not to be generated unless action
275                        // RUNNING via StartX
276                        generateEvent(coordAction, coordJob.getUser(), coordJob.getAppName(), null);
277                    }
278                }
279                else {
280                    CoordActionQueryExecutor.getInstance().executeUpdate(
281                            CoordActionQuery.UPDATE_COORD_ACTION_FOR_MODIFIED_DATE, coordAction);
282                }
283            }
284            catch (JPAExecutorException jex) {
285                throw new CommandException(ErrorCode.E1021, jex.getMessage(), jex);
286            }
287        }
288    }
289
290    /**
291     * This function reads the value of re-queue interval for coordinator input
292     * check command from the Oozie configuration provided by Configuration
293     * Service. If nothing defined in the configuration, it uses the code
294     * specified default value.
295     *
296     * @return re-queue interval in ms
297     */
298    public long getCoordInputCheckRequeueInterval() {
299        long requeueInterval = ConfigurationService.getLong(CONF_COORD_INPUT_CHECK_REQUEUE_INTERVAL);
300        return requeueInterval;
301    }
302
303    /**
304     * To check the list of input paths if all of them exist
305     *
306     * @param actionXml action xml
307     * @param existList the list of existed paths
308     * @param nonExistList the list of non existed paths
309     * @param conf action configuration
310     * @return true if all input paths are existed
311     * @throws Exception thrown of unable to check input path
312     */
313    protected boolean checkInput(StringBuilder actionXml, StringBuilder existList, StringBuilder nonExistList,
314            Configuration conf) throws Exception {
315        Element eAction = XmlUtils.parseXml(actionXml.toString());
316        return checkResolvedUris(eAction, existList, nonExistList, conf);
317    }
318
319    protected boolean checkUnResolvedInput(StringBuilder actionXml, Configuration conf) throws Exception {
320        Element eAction = XmlUtils.parseXml(actionXml.toString());
321        LOG.debug("[" + actionId + "]::ActionInputCheck:: Checking Latest/future");
322        boolean allExist = checkUnresolvedInstances(eAction, conf);
323        if (allExist) {
324            actionXml.replace(0, actionXml.length(), XmlUtils.prettyPrint(eAction).toString());
325        }
326        return allExist;
327    }
328
329    /**
330     * Materialize data properties defined in <action> tag. it includes dataIn(<DS>) and dataOut(<DS>) it creates a list
331     * of files that will be needed.
332     *
333     * @param eAction action element
334     * @param conf action configuration
335     * @throws Exception thrown if failed to resolve data properties
336     * @update modify 'Action' element with appropriate list of files.
337     */
338    @SuppressWarnings("unchecked")
339    static void materializeDataProperties(Element eAction, Configuration conf, ELEvaluator eval) throws Exception {
340        Element configElem = eAction.getChild("action", eAction.getNamespace()).getChild("workflow",
341                eAction.getNamespace()).getChild("configuration", eAction.getNamespace());
342        if (configElem != null) {
343            for (Element propElem : (List<Element>) configElem.getChildren("property", configElem.getNamespace())) {
344                resolveTagContents("value", propElem, eval);
345            }
346        }
347    }
348
349    /**
350     * To resolve property value which contains el functions
351     *
352     * @param tagName tag name
353     * @param elem the child element of "property" element
354     * @param eval el functions evaluator
355     * @throws Exception thrown if unable to resolve tag value
356     */
357    private static void resolveTagContents(String tagName, Element elem, ELEvaluator eval) throws Exception {
358        if (elem == null) {
359            return;
360        }
361        Element tagElem = elem.getChild(tagName, elem.getNamespace());
362        if (tagElem != null) {
363            String updated = CoordELFunctions.evalAndWrap(eval, tagElem.getText());
364            tagElem.removeContent();
365            tagElem.addContent(updated);
366        }
367        else {
368            XLog.getLog(CoordActionInputCheckXCommand.class).warn(" Value NOT FOUND " + tagName);
369        }
370    }
371
372    /**
373     * Check if any unsolved paths under data output. Resolve the unresolved data input paths.
374     *
375     * @param eAction action element
376     * @param actionConf action configuration
377     * @return true if successful to resolve input and output paths
378     * @throws Exception thrown if failed to resolve data input and output paths
379     */
380    @SuppressWarnings("unchecked")
381    private boolean checkUnresolvedInstances(Element eAction, Configuration actionConf) throws Exception {
382        String strAction = XmlUtils.prettyPrint(eAction).toString();
383        Date nominalTime = DateUtils.parseDateOozieTZ(eAction.getAttributeValue("action-nominal-time"));
384        String actualTimeStr = eAction.getAttributeValue("action-actual-time");
385        Date actualTime = null;
386        if (actualTimeStr == null) {
387            LOG.debug("Unable to get action-actual-time from action xml, this job is submitted " +
388            "from previous version. Assign current date to actual time, action = " + actionId);
389            actualTime = new Date();
390        } else {
391            actualTime = DateUtils.parseDateOozieTZ(actualTimeStr);
392        }
393
394        StringBuffer resultedXml = new StringBuffer();
395
396        boolean ret;
397        Element inputList = eAction.getChild("input-events", eAction.getNamespace());
398        if (inputList != null) {
399            ret = materializeUnresolvedEvent(inputList.getChildren("data-in", eAction.getNamespace()), nominalTime,
400                    actualTime, actionConf);
401            if (ret == false) {
402                resultedXml.append(strAction);
403                return false;
404            }
405        }
406
407        // Using latest() or future() in output-event is not intuitive.
408        // We need to make sure, this assumption is correct.
409        Element outputList = eAction.getChild("output-events", eAction.getNamespace());
410        if (outputList != null) {
411            for (Element dEvent : (List<Element>) outputList.getChildren("data-out", eAction.getNamespace())) {
412                if (dEvent.getChild(CoordCommandUtils.UNRESOLVED_INST_TAG, dEvent.getNamespace()) != null) {
413                    throw new CommandException(ErrorCode.E1006, "coord:latest()/future()",
414                            " not permitted in output-event ");
415                }
416            }
417        }
418        return true;
419    }
420
421    /**
422     * Resolve the list of data input paths
423     *
424     * @param eDataEvents the list of data input elements
425     * @param nominalTime action nominal time
426     * @param actualTime current time
427     * @param conf action configuration
428     * @return true if all unresolved URIs can be resolved
429     * @throws Exception thrown if failed to resolve data input paths
430     */
431    @SuppressWarnings("unchecked")
432    private boolean materializeUnresolvedEvent(List<Element> eDataEvents, Date nominalTime, Date actualTime,
433            Configuration conf) throws Exception {
434        for (Element dEvent : eDataEvents) {
435            if (dEvent.getChild(CoordCommandUtils.UNRESOLVED_INST_TAG, dEvent.getNamespace()) == null) {
436                continue;
437            }
438            ELEvaluator eval = CoordELEvaluator.createLazyEvaluator(actualTime, nominalTime, dEvent, conf);
439            String uresolvedInstance = dEvent.getChild(CoordCommandUtils.UNRESOLVED_INST_TAG, dEvent.getNamespace()).getTextTrim();
440            String unresolvedList[] = uresolvedInstance.split(CoordELFunctions.INSTANCE_SEPARATOR);
441            StringBuffer resolvedTmp = new StringBuffer();
442            for (int i = 0; i < unresolvedList.length; i++) {
443                String ret = CoordELFunctions.evalAndWrap(eval, unresolvedList[i]);
444                Boolean isResolved = (Boolean) eval.getVariable("is_resolved");
445                if (isResolved == false) {
446                    LOG.info("[" + actionId + "]::Cannot resolve: " + ret);
447                    return false;
448                }
449                if (resolvedTmp.length() > 0) {
450                    resolvedTmp.append(CoordELFunctions.INSTANCE_SEPARATOR);
451                }
452                resolvedTmp.append((String) eval.getVariable("resolved_path"));
453            }
454            if (resolvedTmp.length() > 0) {
455                if (dEvent.getChild("uris", dEvent.getNamespace()) != null) {
456                    resolvedTmp.append(CoordELFunctions.INSTANCE_SEPARATOR).append(
457                            dEvent.getChild("uris", dEvent.getNamespace()).getTextTrim());
458                    dEvent.removeChild("uris", dEvent.getNamespace());
459                }
460                Element uriInstance = new Element("uris", dEvent.getNamespace());
461                uriInstance.addContent(resolvedTmp.toString());
462                dEvent.getContent().add(1, uriInstance);
463            }
464            dEvent.removeChild(CoordCommandUtils.UNRESOLVED_INST_TAG, dEvent.getNamespace());
465        }
466
467        return true;
468    }
469
470    /**
471     * Check all resolved URIs existence
472     *
473     * @param eAction action element
474     * @param existList the list of existed paths
475     * @param nonExistList the list of paths to check existence
476     * @param conf action configuration
477     * @return true if all nonExistList paths exist
478     * @throws IOException thrown if unable to access the path
479     */
480    private boolean checkResolvedUris(Element eAction, StringBuilder existList, StringBuilder nonExistList,
481            Configuration conf) throws IOException {
482        Element inputList = eAction.getChild("input-events", eAction.getNamespace());
483        if (inputList != null) {
484            if (nonExistList.length() > 0) {
485                checkListOfPaths(existList, nonExistList, conf);
486            }
487            return nonExistList.length() == 0;
488        }
489        return true;
490    }
491
492    /**
493     * Check a list of non existed paths and add to exist list if it exists
494     *
495     * @param existList the list of existed paths
496     * @param nonExistList the list of paths to check existence
497     * @param conf action configuration
498     * @return true if all nonExistList paths exist
499     * @throws IOException thrown if unable to access the path
500     */
501    private boolean checkListOfPaths(StringBuilder existList, StringBuilder nonExistList, Configuration conf)
502            throws IOException {
503
504        String[] uriList = nonExistList.toString().split(CoordELFunctions.INSTANCE_SEPARATOR);
505        if (uriList[0] != null) {
506            LOG.info("[" + actionId + "]::ActionInputCheck:: In checkListOfPaths: " + uriList[0] + " is Missing.");
507        }
508
509        nonExistList.delete(0, nonExistList.length());
510        boolean allExists = true;
511        String existSeparator = "", nonExistSeparator = "";
512        String user = ParamChecker.notEmpty(conf.get(OozieClient.USER_NAME), OozieClient.USER_NAME);
513        for (int i = 0; i < uriList.length; i++) {
514            if (allExists) {
515                allExists = pathExists(uriList[i], conf, user);
516                LOG.info("[" + actionId + "]::ActionInputCheck:: File:" + uriList[i] + ", Exists? :" + allExists);
517            }
518            if (allExists) {
519                existList.append(existSeparator).append(uriList[i]);
520                existSeparator = CoordELFunctions.INSTANCE_SEPARATOR;
521            }
522            else {
523                nonExistList.append(nonExistSeparator).append(uriList[i]);
524                nonExistSeparator = CoordELFunctions.INSTANCE_SEPARATOR;
525            }
526        }
527        return allExists;
528    }
529
530    /**
531     * Check if given path exists
532     *
533     * @param sPath uri path
534     * @param actionConf action configuration
535     * @return true if path exists
536     * @throws IOException thrown if unable to access the path
537     */
538    protected boolean pathExists(String sPath, Configuration actionConf, String user) throws IOException {
539        LOG.debug("checking for the file " + sPath);
540        try {
541            URI uri = new URI(sPath);
542            URIHandlerService service = Services.get().get(URIHandlerService.class);
543            URIHandler handler = service.getURIHandler(uri);
544            return handler.exists(uri, actionConf, user);
545        }
546        catch (URIHandlerException e) {
547            coordAction.setErrorCode(e.getErrorCode().toString());
548            coordAction.setErrorMessage(e.getMessage());
549            if (e.getCause() != null && e.getCause() instanceof AccessControlException) {
550                throw (AccessControlException) e.getCause();
551            }
552            else {
553                throw new IOException(e);
554            }
555        }
556        catch (URISyntaxException e) {
557            coordAction.setErrorCode(ErrorCode.E0906.toString());
558            coordAction.setErrorMessage(e.getMessage());
559            throw new IOException(e);
560        }
561    }
562
563    /**
564     * The function create a list of URIs separated by "," using the instances time stamp and URI-template
565     *
566     * @param event : <data-in> event
567     * @param instances : List of time stamp seprated by ","
568     * @param unresolvedInstances : list of instance with latest/future function
569     * @return : list of URIs separated by ",".
570     * @throws Exception thrown if failed to create URIs from unresolvedInstances
571     */
572    @SuppressWarnings("unused")
573    private String createURIs(Element event, String instances, StringBuilder unresolvedInstances) throws Exception {
574        if (instances == null || instances.length() == 0) {
575            return "";
576        }
577        String[] instanceList = instances.split(CoordELFunctions.INSTANCE_SEPARATOR);
578        StringBuilder uris = new StringBuilder();
579
580        for (int i = 0; i < instanceList.length; i++) {
581            int funcType = CoordCommandUtils.getFuncType(instanceList[i]);
582            if (funcType == CoordCommandUtils.LATEST || funcType == CoordCommandUtils.FUTURE) {
583                if (unresolvedInstances.length() > 0) {
584                    unresolvedInstances.append(CoordELFunctions.INSTANCE_SEPARATOR);
585                }
586                unresolvedInstances.append(instanceList[i]);
587                continue;
588            }
589            ELEvaluator eval = CoordELEvaluator.createURIELEvaluator(instanceList[i]);
590            if (uris.length() > 0) {
591                uris.append(CoordELFunctions.INSTANCE_SEPARATOR);
592            }
593            uris.append(CoordELFunctions.evalAndWrap(eval, event.getChild("dataset", event.getNamespace()).getChild(
594                    "uri-template", event.getNamespace()).getTextTrim()));
595        }
596        return uris.toString();
597    }
598
599    /**
600     * getting the error code of the coord action. (used mainly for unit testing)
601     */
602    protected String getCoordActionErrorCode() {
603        if (coordAction != null) {
604            return coordAction.getErrorCode();
605        }
606        return null;
607    }
608
609    /**
610     * getting the error message of the coord action. (used mainly for unit testing)
611     */
612    protected String getCoordActionErrorMsg() {
613        if (coordAction != null) {
614            return coordAction.getErrorMessage();
615        }
616        return null;
617    }
618
619    @Override
620    public String getEntityKey() {
621        return this.jobId;
622    }
623
624    @Override
625    protected boolean isLockRequired() {
626        return true;
627    }
628
629    @Override
630    protected void loadState() throws CommandException {
631        if (jpaService == null) {
632            jpaService = Services.get().get(JPAService.class);
633        }
634        try {
635            coordAction = jpaService.execute(new CoordActionGetForInputCheckJPAExecutor(actionId));
636            coordJob = CoordJobQueryExecutor.getInstance().get(CoordJobQuery.GET_COORD_JOB_INPUT_CHECK,
637                    coordAction.getJobId());
638        }
639        catch (JPAExecutorException je) {
640            throw new CommandException(je);
641        }
642        LogUtils.setLogInfo(coordAction);
643    }
644
645    @Override
646    protected void verifyPrecondition() throws CommandException, PreconditionException {
647        if (coordAction.getStatus() != CoordinatorActionBean.Status.WAITING) {
648            throw new PreconditionException(ErrorCode.E1100, "[" + actionId
649                    + "]::CoordActionInputCheck:: Ignoring action. Should be in WAITING state, but state="
650                    + coordAction.getStatus());
651        }
652
653        // if eligible to do action input check when running with backward support is true
654        if (StatusUtils.getStatusForCoordActionInputCheck(coordJob)) {
655            return;
656        }
657
658        if (coordJob.getStatus() != Job.Status.RUNNING && coordJob.getStatus() != Job.Status.RUNNINGWITHERROR && coordJob.getStatus() != Job.Status.PAUSED
659                && coordJob.getStatus() != Job.Status.PAUSEDWITHERROR) {
660            throw new PreconditionException(
661                    ErrorCode.E1100, "["+ actionId + "]::CoordActionInputCheck:: Ignoring action." +
662                                " Coordinator job is not in RUNNING/RUNNINGWITHERROR/PAUSED/PAUSEDWITHERROR state, but state="
663                            + coordJob.getStatus());
664        }
665    }
666
667    @Override
668    public String getKey(){
669        return getName() + "_" + actionId;
670    }
671
672}