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.InputStreamReader;
023import java.io.Reader;
024import java.io.StringReader;
025import java.io.StringWriter;
026import java.net.URI;
027import java.net.URISyntaxException;
028import java.util.ArrayList;
029import java.util.Calendar;
030import java.util.Date;
031import java.util.HashMap;
032import java.util.HashSet;
033import java.util.Iterator;
034import java.util.List;
035import java.util.Map;
036import java.util.Set;
037import java.util.TreeSet;
038
039import javax.xml.transform.stream.StreamSource;
040import javax.xml.validation.Validator;
041
042import org.apache.hadoop.conf.Configuration;
043import org.apache.hadoop.fs.FileSystem;
044import org.apache.hadoop.fs.Path;
045import org.apache.oozie.CoordinatorJobBean;
046import org.apache.oozie.ErrorCode;
047import org.apache.oozie.client.CoordinatorJob;
048import org.apache.oozie.client.Job;
049import org.apache.oozie.client.OozieClient;
050import org.apache.oozie.client.CoordinatorJob.Execution;
051import org.apache.oozie.command.CommandException;
052import org.apache.oozie.command.SubmitTransitionXCommand;
053import org.apache.oozie.command.bundle.BundleStatusUpdateXCommand;
054import org.apache.oozie.coord.CoordELEvaluator;
055import org.apache.oozie.coord.CoordELFunctions;
056import org.apache.oozie.coord.CoordinatorJobException;
057import org.apache.oozie.coord.TimeUnit;
058import org.apache.oozie.executor.jpa.CoordJobQueryExecutor;
059import org.apache.oozie.executor.jpa.JPAExecutorException;
060import org.apache.oozie.service.CoordMaterializeTriggerService;
061import org.apache.oozie.service.ConfigurationService;
062import org.apache.oozie.service.HadoopAccessorException;
063import org.apache.oozie.service.HadoopAccessorService;
064import org.apache.oozie.service.JPAService;
065import org.apache.oozie.service.SchemaService;
066import org.apache.oozie.service.Service;
067import org.apache.oozie.service.Services;
068import org.apache.oozie.service.UUIDService;
069import org.apache.oozie.service.SchemaService.SchemaName;
070import org.apache.oozie.service.UUIDService.ApplicationType;
071import org.apache.oozie.util.ConfigUtils;
072import org.apache.oozie.util.DateUtils;
073import org.apache.oozie.util.ELEvaluator;
074import org.apache.oozie.util.ELUtils;
075import org.apache.oozie.util.IOUtils;
076import org.apache.oozie.util.InstrumentUtils;
077import org.apache.oozie.util.LogUtils;
078import org.apache.oozie.util.ParamChecker;
079import org.apache.oozie.util.ParameterVerifier;
080import org.apache.oozie.util.ParameterVerifierException;
081import org.apache.oozie.util.PropertiesUtils;
082import org.apache.oozie.util.XConfiguration;
083import org.apache.oozie.util.XmlUtils;
084import org.jdom.Attribute;
085import org.jdom.Element;
086import org.jdom.JDOMException;
087import org.jdom.Namespace;
088import org.xml.sax.SAXException;
089
090/**
091 * This class provides the functionalities to resolve a coordinator job XML and write the job information into a DB
092 * table.
093 * <p/>
094 * Specifically it performs the following functions: 1. Resolve all the variables or properties using job
095 * configurations. 2. Insert all datasets definition as part of the <data-in> and <data-out> tags. 3. Validate the XML
096 * at runtime.
097 */
098public class CoordSubmitXCommand extends SubmitTransitionXCommand {
099
100    protected Configuration conf;
101    private final String bundleId;
102    private final String coordName;
103    protected boolean dryrun;
104    protected JPAService jpaService = null;
105    private CoordinatorJob.Status prevStatus = CoordinatorJob.Status.PREP;
106
107    public static final String CONFIG_DEFAULT = "coord-config-default.xml";
108    public static final String COORDINATOR_XML_FILE = "coordinator.xml";
109    public final String COORD_INPUT_EVENTS ="input-events";
110    public final String COORD_OUTPUT_EVENTS = "output-events";
111    public final String COORD_INPUT_EVENTS_DATA_IN ="data-in";
112    public final String COORD_OUTPUT_EVENTS_DATA_OUT = "data-out";
113
114    private static final Set<String> DISALLOWED_USER_PROPERTIES = new HashSet<String>();
115    private static final Set<String> DISALLOWED_DEFAULT_PROPERTIES = new HashSet<String>();
116
117    protected CoordinatorJobBean coordJob = null;
118    /**
119     * Default timeout for normal jobs, in minutes, after which coordinator input check will timeout
120     */
121    public static final String CONF_DEFAULT_TIMEOUT_NORMAL = Service.CONF_PREFIX + "coord.normal.default.timeout";
122
123    public static final String CONF_DEFAULT_CONCURRENCY = Service.CONF_PREFIX + "coord.default.concurrency";
124
125    public static final String CONF_DEFAULT_THROTTLE = Service.CONF_PREFIX + "coord.default.throttle";
126
127    public static final String CONF_MAT_THROTTLING_FACTOR = Service.CONF_PREFIX
128            + "coord.materialization.throttling.factor";
129
130    /**
131     * Default MAX timeout in minutes, after which coordinator input check will timeout
132     */
133    public static final String CONF_DEFAULT_MAX_TIMEOUT = Service.CONF_PREFIX + "coord.default.max.timeout";
134
135    public static final String CONF_QUEUE_SIZE = Service.CONF_PREFIX + "CallableQueueService.queue.size";
136
137    public static final String CONF_CHECK_MAX_FREQUENCY = Service.CONF_PREFIX + "coord.check.maximum.frequency";
138
139    private ELEvaluator evalFreq = null;
140    private ELEvaluator evalNofuncs = null;
141    private ELEvaluator evalData = null;
142    private ELEvaluator evalInst = null;
143    private ELEvaluator evalAction = null;
144    private ELEvaluator evalSla = null;
145    private ELEvaluator evalTimeout = null;
146    private ELEvaluator evalInitialInstance = null;
147
148    static {
149        String[] badUserProps = { PropertiesUtils.YEAR, PropertiesUtils.MONTH, PropertiesUtils.DAY,
150                PropertiesUtils.HOUR, PropertiesUtils.MINUTE, PropertiesUtils.DAYS, PropertiesUtils.HOURS,
151                PropertiesUtils.MINUTES, PropertiesUtils.KB, PropertiesUtils.MB, PropertiesUtils.GB,
152                PropertiesUtils.TB, PropertiesUtils.PB, PropertiesUtils.RECORDS, PropertiesUtils.MAP_IN,
153                PropertiesUtils.MAP_OUT, PropertiesUtils.REDUCE_IN, PropertiesUtils.REDUCE_OUT, PropertiesUtils.GROUPS };
154        PropertiesUtils.createPropertySet(badUserProps, DISALLOWED_USER_PROPERTIES);
155
156        String[] badDefaultProps = { PropertiesUtils.HADOOP_USER};
157        PropertiesUtils.createPropertySet(badUserProps, DISALLOWED_DEFAULT_PROPERTIES);
158        PropertiesUtils.createPropertySet(badDefaultProps, DISALLOWED_DEFAULT_PROPERTIES);
159    }
160
161    /**
162     * Constructor to create the Coordinator Submit Command.
163     *
164     * @param conf : Configuration for Coordinator job
165     */
166    public CoordSubmitXCommand(Configuration conf) {
167        super("coord_submit", "coord_submit", 1);
168        this.conf = ParamChecker.notNull(conf, "conf");
169        this.bundleId = null;
170        this.coordName = null;
171    }
172
173    /**
174     * Constructor to create the Coordinator Submit Command by bundle job.
175     *
176     * @param conf : Configuration for Coordinator job
177     * @param bundleId : bundle id
178     * @param coordName : coord name
179     */
180    public CoordSubmitXCommand(Configuration conf, String bundleId, String coordName) {
181        super("coord_submit", "coord_submit", 1);
182        this.conf = ParamChecker.notNull(conf, "conf");
183        this.bundleId = ParamChecker.notEmpty(bundleId, "bundleId");
184        this.coordName = ParamChecker.notEmpty(coordName, "coordName");
185    }
186
187    /**
188     * Constructor to create the Coordinator Submit Command.
189     *
190     * @param dryrun : if dryrun
191     * @param conf : Configuration for Coordinator job
192     */
193    public CoordSubmitXCommand(boolean dryrun, Configuration conf) {
194        this(conf);
195        this.dryrun = dryrun;
196    }
197
198    /* (non-Javadoc)
199     * @see org.apache.oozie.command.XCommand#execute()
200     */
201    @Override
202    protected String submit() throws CommandException {
203        LOG.info("STARTED Coordinator Submit");
204        String jobId = submitJob();
205        LOG.info("ENDED Coordinator Submit jobId=" + jobId);
206        return jobId;
207    }
208
209    protected String submitJob() throws CommandException {
210        String jobId = null;
211        InstrumentUtils.incrJobCounter(getName(), 1, getInstrumentation());
212
213        boolean exceptionOccured = false;
214        try {
215            mergeDefaultConfig();
216
217            String appXml = readAndValidateXml();
218            coordJob.setOrigJobXml(appXml);
219            LOG.debug("jobXml after initial validation " + XmlUtils.prettyPrint(appXml).toString());
220
221            Element eXml = XmlUtils.parseXml(appXml);
222
223            String appNamespace = readAppNamespace(eXml);
224            coordJob.setAppNamespace(appNamespace);
225
226            ParameterVerifier.verifyParameters(conf, eXml);
227
228            appXml = XmlUtils.removeComments(appXml);
229            initEvaluators();
230            Element eJob = basicResolveAndIncludeDS(appXml, conf, coordJob);
231
232            validateCoordinatorJob();
233
234            // checking if the coordinator application data input/output events
235            // specify multiple data instance values in erroneous manner
236            checkMultipleTimeInstances(eJob, COORD_INPUT_EVENTS, COORD_INPUT_EVENTS_DATA_IN);
237            checkMultipleTimeInstances(eJob, COORD_OUTPUT_EVENTS, COORD_OUTPUT_EVENTS_DATA_OUT);
238
239            LOG.debug("jobXml after all validation " + XmlUtils.prettyPrint(eJob).toString());
240
241            jobId = storeToDB(appXml, eJob, coordJob);
242            // log job info for coordinator job
243            LogUtils.setLogInfo(coordJob);
244
245            if (!dryrun) {
246                queueMaterializeTransitionXCommand(jobId);
247            }
248            else {
249                return getDryRun(coordJob);
250            }
251        }
252        catch (JDOMException jex) {
253            exceptionOccured = true;
254            LOG.warn("ERROR: ", jex);
255            throw new CommandException(ErrorCode.E0700, jex.getMessage(), jex);
256        }
257        catch (CoordinatorJobException cex) {
258            exceptionOccured = true;
259            LOG.warn("ERROR:  ", cex);
260            throw new CommandException(cex);
261        }
262        catch (ParameterVerifierException pex) {
263            exceptionOccured = true;
264            LOG.warn("ERROR: ", pex);
265            throw new CommandException(pex);
266        }
267        catch (IllegalArgumentException iex) {
268            exceptionOccured = true;
269            LOG.warn("ERROR:  ", iex);
270            throw new CommandException(ErrorCode.E1003, iex.getMessage(), iex);
271        }
272        catch (Exception ex) {
273            exceptionOccured = true;
274            LOG.warn("ERROR:  ", ex);
275            throw new CommandException(ErrorCode.E0803, ex.getMessage(), ex);
276        }
277        finally {
278            if (exceptionOccured) {
279                if (coordJob.getId() == null || coordJob.getId().equalsIgnoreCase("")) {
280                    coordJob.setStatus(CoordinatorJob.Status.FAILED);
281                    coordJob.resetPending();
282                }
283            }
284        }
285        return jobId;
286    }
287
288    /**
289     * Gets the dryrun output.
290     *
291     * @param coordJob the coordinatorJobBean
292     * @return the dry run
293     * @throws Exception the exception
294     */
295    protected String getDryRun(CoordinatorJobBean coordJob) throws Exception{
296        int materializationWindow = ConfigurationService
297                .getInt(CoordMaterializeTriggerService.CONF_MATERIALIZATION_WINDOW);
298        Date startTime = coordJob.getStartTime();
299        long startTimeMilli = startTime.getTime();
300        long endTimeMilli = startTimeMilli + (materializationWindow * 1000);
301        Date jobEndTime = coordJob.getEndTime();
302        Date endTime = new Date(endTimeMilli);
303        if (endTime.compareTo(jobEndTime) > 0) {
304            endTime = jobEndTime;
305        }
306        String jobId = coordJob.getId();
307        LOG.info("[" + jobId + "]: Update status to RUNNING");
308        coordJob.setStatus(Job.Status.RUNNING);
309        coordJob.setPending();
310        Configuration jobConf = null;
311        try {
312            jobConf = new XConfiguration(new StringReader(coordJob.getConf()));
313        }
314        catch (IOException e1) {
315            LOG.warn("Configuration parse error. read from DB :" + coordJob.getConf(), e1);
316        }
317        String action = new CoordMaterializeTransitionXCommand(coordJob, materializationWindow, startTime,
318                endTime).materializeActions(true);
319        String output = coordJob.getJobXml() + System.getProperty("line.separator")
320        + "***actions for instance***" + action;
321        return output;
322    }
323
324    /**
325     * Queue MaterializeTransitionXCommand
326     */
327    protected void queueMaterializeTransitionXCommand(String jobId) {
328        int materializationWindow = ConfigurationService
329                .getInt(CoordMaterializeTriggerService.CONF_MATERIALIZATION_WINDOW);
330        queue(new CoordMaterializeTransitionXCommand(jobId, materializationWindow), 100);
331    }
332
333    /**
334     * Method that validates values in the definition for correctness. Placeholder to add more.
335     */
336    private void validateCoordinatorJob() throws Exception {
337        // check if startTime < endTime
338        if (!coordJob.getStartTime().before(coordJob.getEndTime())) {
339            throw new IllegalArgumentException("Coordinator Start Time must be earlier than End Time.");
340        }
341
342        try {
343            // Check if a coord job with cron frequency will materialize actions
344            int freq = Integer.parseInt(coordJob.getFrequency());
345
346            // Check if the frequency is faster than 5 min if enabled
347            if (ConfigurationService.getBoolean(CONF_CHECK_MAX_FREQUENCY)) {
348                CoordinatorJob.Timeunit unit = coordJob.getTimeUnit();
349                if (freq == 0 || (freq < 5 && unit == CoordinatorJob.Timeunit.MINUTE)) {
350                    throw new IllegalArgumentException("Coordinator job with frequency [" + freq +
351                            "] minutes is faster than allowed maximum of 5 minutes ("
352                            + CONF_CHECK_MAX_FREQUENCY + " is set to true)");
353                }
354            }
355        } catch (NumberFormatException e) {
356            Date start = coordJob.getStartTime();
357            Calendar cal = Calendar.getInstance();
358            cal.setTime(start);
359            cal.add(Calendar.MINUTE, -1);
360            start = cal.getTime();
361
362            Date nextTime = CoordCommandUtils.getNextValidActionTimeForCronFrequency(start, coordJob);
363            if (nextTime == null) {
364                throw new IllegalArgumentException("Invalid coordinator cron frequency: " + coordJob.getFrequency());
365            }
366            if (!nextTime.before(coordJob.getEndTime())) {
367                throw new IllegalArgumentException("Coordinator job with frequency '" +
368                        coordJob.getFrequency() + "' materializes no actions between start and end time.");
369            }
370        }
371    }
372
373  /*
374  * Check against multiple data instance values inside a single <instance> <start-instance> or <end-instance> tag
375  * If found, the job is not submitted and user is informed to correct the error, instead of defaulting to the first instance value in the list
376  */
377    private void checkMultipleTimeInstances(Element eCoordJob, String eventType, String dataType) throws CoordinatorJobException {
378        Element eventsSpec, dataSpec, instance;
379        List<Element> instanceSpecList;
380        Namespace ns = eCoordJob.getNamespace();
381        String instanceValue;
382        eventsSpec = eCoordJob.getChild(eventType, ns);
383        if (eventsSpec != null) {
384            dataSpec = eventsSpec.getChild(dataType, ns);
385            if (dataSpec != null) {
386                // In case of input-events, there can be multiple child <instance> datasets. Iterating to ensure none of them have errors
387                instanceSpecList = dataSpec.getChildren("instance", ns);
388                Iterator instanceIter = instanceSpecList.iterator();
389                while(instanceIter.hasNext()) {
390                    instance = ((Element) instanceIter.next());
391                    if(instance.getContentSize() == 0) { //empty string or whitespace
392                        throw new CoordinatorJobException(ErrorCode.E1021, "<instance> tag within " + eventType + " is empty!");
393                    }
394                    instanceValue = instance.getContent(0).toString();
395                    boolean isInvalid = false;
396                    try {
397                        isInvalid = evalAction.checkForExistence(instanceValue, ",");
398                    } catch (Exception e) {
399                        handleELParseException(eventType, dataType, instanceValue);
400                    }
401                    if (isInvalid) { // reaching this block implies instance is not empty i.e. length > 0
402                        handleExpresionWithMultipleInstances(eventType, dataType, instanceValue);
403                    }
404                }
405
406                // In case of input-events, there can be multiple child <start-instance> datasets. Iterating to ensure none of them have errors
407                instanceSpecList = dataSpec.getChildren("start-instance", ns);
408                instanceIter = instanceSpecList.iterator();
409                while(instanceIter.hasNext()) {
410                    instance = ((Element) instanceIter.next());
411                    if(instance.getContentSize() == 0) { //empty string or whitespace
412                        throw new CoordinatorJobException(ErrorCode.E1021, "<start-instance> tag within " + eventType + " is empty!");
413                    }
414                    instanceValue = instance.getContent(0).toString();
415                    boolean isInvalid = false;
416                    try {
417                        isInvalid = evalAction.checkForExistence(instanceValue, ",");
418                    } catch (Exception e) {
419                        handleELParseException(eventType, dataType, instanceValue);
420                    }
421                    if (isInvalid) { // reaching this block implies start instance is not empty i.e. length > 0
422                        handleExpresionWithStartMultipleInstances(eventType, dataType, instanceValue);
423                    }
424                }
425
426                // In case of input-events, there can be multiple child <end-instance> datasets. Iterating to ensure none of them have errors
427                instanceSpecList = dataSpec.getChildren("end-instance", ns);
428                instanceIter = instanceSpecList.iterator();
429                while(instanceIter.hasNext()) {
430                    instance = ((Element) instanceIter.next());
431                    if(instance.getContentSize() == 0) { //empty string or whitespace
432                        throw new CoordinatorJobException(ErrorCode.E1021, "<end-instance> tag within " + eventType + " is empty!");
433                    }
434                    instanceValue = instance.getContent(0).toString();
435                    boolean isInvalid = false;
436                    try {
437                        isInvalid = evalAction.checkForExistence(instanceValue, ",");
438                    } catch (Exception e) {
439                        handleELParseException(eventType, dataType, instanceValue);
440                    }
441                    if (isInvalid) { // reaching this block implies instance is not empty i.e. length > 0
442                        handleExpresionWithMultipleEndInstances(eventType, dataType, instanceValue);
443                    }
444                }
445
446            }
447        }
448    }
449
450    private void handleELParseException(String eventType, String dataType, String instanceValue)
451            throws CoordinatorJobException {
452        String correctAction = null;
453        if(dataType.equals(COORD_INPUT_EVENTS_DATA_IN)) {
454            correctAction = "Coordinator app definition should have valid <instance> tag for data-in";
455        } else if(dataType.equals(COORD_OUTPUT_EVENTS_DATA_OUT)) {
456            correctAction = "Coordinator app definition should have valid <instance> tag for data-out";
457        }
458        throw new CoordinatorJobException(ErrorCode.E1021, eventType + " instance '" + instanceValue
459                + "' is not valid. Coordinator job NOT SUBMITTED. " + correctAction);
460    }
461
462    private void handleExpresionWithMultipleInstances(String eventType, String dataType, String instanceValue)
463            throws CoordinatorJobException {
464        String correctAction = null;
465        if(dataType.equals(COORD_INPUT_EVENTS_DATA_IN)) {
466            correctAction = "Coordinator app definition should have separate <instance> tag per data-in instance";
467        } else if(dataType.equals(COORD_OUTPUT_EVENTS_DATA_OUT)) {
468            correctAction = "Coordinator app definition can have only one <instance> tag per data-out instance";
469        }
470        throw new CoordinatorJobException(ErrorCode.E1021, eventType + " instance '" + instanceValue
471                + "' contains more than one date instance. Coordinator job NOT SUBMITTED. " + correctAction);
472    }
473
474    private void handleExpresionWithStartMultipleInstances(String eventType, String dataType, String instanceValue)
475            throws CoordinatorJobException {
476        String correctAction = "Coordinator app definition should not have multiple start-instances";
477        throw new CoordinatorJobException(ErrorCode.E1021, eventType + " start-instance '" + instanceValue
478                + "' contains more than one date start-instance. Coordinator job NOT SUBMITTED. " + correctAction);
479    }
480
481    private void handleExpresionWithMultipleEndInstances(String eventType, String dataType, String instanceValue)
482            throws CoordinatorJobException {
483        String correctAction = "Coordinator app definition should not have multiple end-instances";
484        throw new CoordinatorJobException(ErrorCode.E1021, eventType + " end-instance '" + instanceValue
485                + "' contains more than one date end-instance. Coordinator job NOT SUBMITTED. " + correctAction);
486    }
487    /**
488     * Read the application XML and validate against coordinator Schema
489     *
490     * @return validated coordinator XML
491     * @throws CoordinatorJobException thrown if unable to read or validate coordinator xml
492     */
493    protected String readAndValidateXml() throws CoordinatorJobException {
494        String appPath = ParamChecker.notEmpty(conf.get(OozieClient.COORDINATOR_APP_PATH),
495                OozieClient.COORDINATOR_APP_PATH);
496        String coordXml = readDefinition(appPath);
497        validateXml(coordXml);
498        return coordXml;
499    }
500
501    /**
502     * Validate against Coordinator XSD file
503     *
504     * @param xmlContent : Input coordinator xml
505     * @throws CoordinatorJobException thrown if unable to validate coordinator xml
506     */
507    private void validateXml(String xmlContent) throws CoordinatorJobException {
508        javax.xml.validation.Schema schema = Services.get().get(SchemaService.class).getSchema(SchemaName.COORDINATOR);
509        Validator validator = schema.newValidator();
510        try {
511            validator.validate(new StreamSource(new StringReader(xmlContent)));
512        }
513        catch (SAXException ex) {
514            LOG.warn("SAXException :", ex);
515            throw new CoordinatorJobException(ErrorCode.E0701, ex.getMessage(), ex);
516        }
517        catch (IOException ex) {
518            LOG.warn("IOException :", ex);
519            throw new CoordinatorJobException(ErrorCode.E0702, ex.getMessage(), ex);
520        }
521    }
522
523    /**
524     * Read the application XML schema namespace
525     *
526     * @param coordXmlElement input coordinator xml Element
527     * @return app xml namespace
528     * @throws CoordinatorJobException
529     */
530    private String readAppNamespace(Element coordXmlElement) throws CoordinatorJobException {
531        Namespace ns = coordXmlElement.getNamespace();
532        if (ns != null && bundleId != null && ns.getURI().equals(SchemaService.COORDINATOR_NAMESPACE_URI_1)) {
533            throw new CoordinatorJobException(ErrorCode.E1319, "bundle app can not submit coordinator namespace "
534                    + SchemaService.COORDINATOR_NAMESPACE_URI_1 + ", please use 0.2 or later");
535        }
536        if (ns != null) {
537            return ns.getURI();
538        }
539        else {
540            throw new CoordinatorJobException(ErrorCode.E0700, "the application xml namespace is not given");
541        }
542    }
543
544    /**
545     * Merge default configuration with user-defined configuration.
546     *
547     * @throws CommandException thrown if failed to read or merge configurations
548     */
549    protected void mergeDefaultConfig() throws CommandException {
550        Path configDefault = null;
551        try {
552            String coordAppPathStr = conf.get(OozieClient.COORDINATOR_APP_PATH);
553            Path coordAppPath = new Path(coordAppPathStr);
554            String user = ParamChecker.notEmpty(conf.get(OozieClient.USER_NAME), OozieClient.USER_NAME);
555            HadoopAccessorService has = Services.get().get(HadoopAccessorService.class);
556            Configuration fsConf = has.createJobConf(coordAppPath.toUri().getAuthority());
557            FileSystem fs = has.createFileSystem(user, coordAppPath.toUri(), fsConf);
558
559            // app path could be a directory
560            if (!fs.isFile(coordAppPath)) {
561                configDefault = new Path(coordAppPath, CONFIG_DEFAULT);
562            } else {
563                configDefault = new Path(coordAppPath.getParent(), CONFIG_DEFAULT);
564            }
565
566            if (fs.exists(configDefault)) {
567                Configuration defaultConf = new XConfiguration(fs.open(configDefault));
568                PropertiesUtils.checkDisallowedProperties(defaultConf, DISALLOWED_DEFAULT_PROPERTIES);
569                XConfiguration.injectDefaults(defaultConf, conf);
570            }
571            else {
572                LOG.info("configDefault Doesn't exist " + configDefault);
573            }
574            PropertiesUtils.checkDisallowedProperties(conf, DISALLOWED_USER_PROPERTIES);
575
576            // Resolving all variables in the job properties.
577            // This ensures the Hadoop Configuration semantics is preserved.
578            XConfiguration resolvedVarsConf = new XConfiguration();
579            for (Map.Entry<String, String> entry : conf) {
580                resolvedVarsConf.set(entry.getKey(), conf.get(entry.getKey()));
581            }
582            conf = resolvedVarsConf;
583        }
584        catch (IOException e) {
585            throw new CommandException(ErrorCode.E0702, e.getMessage() + " : Problem reading default config "
586                    + configDefault, e);
587        }
588        catch (HadoopAccessorException e) {
589            throw new CommandException(e);
590        }
591        LOG.debug("Merged CONF :" + XmlUtils.prettyPrint(conf).toString());
592    }
593
594    /**
595     * The method resolve all the variables that are defined in configuration. It also include the data set definition
596     * from dataset file into XML.
597     *
598     * @param appXml : Original job XML
599     * @param conf : Configuration of the job
600     * @param coordJob : Coordinator job bean to be populated.
601     * @return Resolved and modified job XML element.
602     * @throws CoordinatorJobException thrown if failed to resolve basic entities or include referred datasets
603     * @throws Exception thrown if failed to resolve basic entities or include referred datasets
604     */
605    public Element basicResolveAndIncludeDS(String appXml, Configuration conf, CoordinatorJobBean coordJob)
606    throws CoordinatorJobException, Exception {
607        Element basicResolvedApp = resolveInitial(conf, appXml, coordJob);
608        includeDataSets(basicResolvedApp, conf);
609        return basicResolvedApp;
610    }
611
612    /**
613     * Insert data set into data-in and data-out tags.
614     *
615     * @param eAppXml : coordinator application XML
616     * @param eDatasets : DataSet XML
617     */
618    @SuppressWarnings("unchecked")
619    private void insertDataSet(Element eAppXml, Element eDatasets) {
620        // Adding DS definition in the coordinator XML
621        Element inputList = eAppXml.getChild("input-events", eAppXml.getNamespace());
622        if (inputList != null) {
623            for (Element dataIn : (List<Element>) inputList.getChildren("data-in", eAppXml.getNamespace())) {
624                Element eDataset = findDataSet(eDatasets, dataIn.getAttributeValue("dataset"));
625                dataIn.getContent().add(0, eDataset);
626            }
627        }
628        Element outputList = eAppXml.getChild("output-events", eAppXml.getNamespace());
629        if (outputList != null) {
630            for (Element dataOut : (List<Element>) outputList.getChildren("data-out", eAppXml.getNamespace())) {
631                Element eDataset = findDataSet(eDatasets, dataOut.getAttributeValue("dataset"));
632                dataOut.getContent().add(0, eDataset);
633            }
634        }
635    }
636
637    /**
638     * Find a specific dataset from a list of Datasets.
639     *
640     * @param eDatasets : List of data sets
641     * @param name : queried data set name
642     * @return one Dataset element. otherwise throw Exception
643     */
644    @SuppressWarnings("unchecked")
645    private static Element findDataSet(Element eDatasets, String name) {
646        for (Element eDataset : (List<Element>) eDatasets.getChildren("dataset", eDatasets.getNamespace())) {
647            if (eDataset.getAttributeValue("name").equals(name)) {
648                eDataset = (Element) eDataset.clone();
649                eDataset.detach();
650                return eDataset;
651            }
652        }
653        throw new RuntimeException("undefined dataset: " + name);
654    }
655
656    /**
657     * Initialize all the required EL Evaluators.
658     */
659    protected void initEvaluators() {
660        evalFreq = CoordELEvaluator.createELEvaluatorForGroup(conf, "coord-job-submit-freq");
661        evalNofuncs = CoordELEvaluator.createELEvaluatorForGroup(conf, "coord-job-submit-nofuncs");
662        evalInst = CoordELEvaluator.createELEvaluatorForGroup(conf, "coord-job-submit-instances");
663        evalAction = CoordELEvaluator.createELEvaluatorForGroup(conf, "coord-action-start");
664        evalTimeout = CoordELEvaluator.createELEvaluatorForGroup(conf, "coord-job-wait-timeout");
665        evalInitialInstance = CoordELEvaluator.createELEvaluatorForGroup(conf, "coord-job-submit-initial-instance");
666
667    }
668
669    /**
670     * Resolve basic entities using job Configuration.
671     *
672     * @param conf :Job configuration
673     * @param appXml : Original job XML
674     * @param coordJob : Coordinator job bean to be populated.
675     * @return Resolved job XML element.
676     * @throws CoordinatorJobException thrown if failed to resolve basic entities
677     * @throws Exception thrown if failed to resolve basic entities
678     */
679    @SuppressWarnings("unchecked")
680    protected Element resolveInitial(Configuration conf, String appXml, CoordinatorJobBean coordJob)
681    throws CoordinatorJobException, Exception {
682        Element eAppXml = XmlUtils.parseXml(appXml);
683        // job's main attributes
684        // frequency
685        String val = resolveAttribute("frequency", eAppXml, evalFreq);
686        int ival = 0;
687
688        val = ParamChecker.checkFrequency(val);
689        coordJob.setFrequency(val);
690        TimeUnit tmp = (evalFreq.getVariable("timeunit") == null) ? TimeUnit.MINUTE : ((TimeUnit) evalFreq
691                .getVariable("timeunit"));
692        try {
693            Integer.parseInt(val);
694        }
695        catch (NumberFormatException ex) {
696            tmp=TimeUnit.CRON;
697        }
698
699        addAnAttribute("freq_timeunit", eAppXml, tmp.toString());
700        // TimeUnit
701        coordJob.setTimeUnit(CoordinatorJob.Timeunit.valueOf(tmp.toString()));
702        // End Of Duration
703        tmp = evalFreq.getVariable("endOfDuration") == null ? TimeUnit.NONE : ((TimeUnit) evalFreq
704                .getVariable("endOfDuration"));
705        addAnAttribute("end_of_duration", eAppXml, tmp.toString());
706        // coordJob.setEndOfDuration(tmp) // TODO: Add new attribute in Job bean
707
708        // Application name
709        if (this.coordName == null) {
710            String name = ELUtils.resolveAppName(eAppXml.getAttribute("name").getValue(), conf);
711            coordJob.setAppName(name);
712        }
713        else {
714            // this coord job is created from bundle
715            coordJob.setAppName(this.coordName);
716        }
717
718        // start time
719        val = resolveAttribute("start", eAppXml, evalNofuncs);
720        ParamChecker.checkDateOozieTZ(val, "start");
721        coordJob.setStartTime(DateUtils.parseDateOozieTZ(val));
722        // end time
723        val = resolveAttribute("end", eAppXml, evalNofuncs);
724        ParamChecker.checkDateOozieTZ(val, "end");
725        coordJob.setEndTime(DateUtils.parseDateOozieTZ(val));
726        // Time zone
727        val = resolveAttribute("timezone", eAppXml, evalNofuncs);
728        ParamChecker.checkTimeZone(val, "timezone");
729        coordJob.setTimeZone(val);
730
731        // controls
732        val = resolveTagContents("timeout", eAppXml.getChild("controls", eAppXml.getNamespace()), evalTimeout);
733        if (val != null && val != "") {
734            int t = Integer.parseInt(val);
735            tmp = (evalTimeout.getVariable("timeunit") == null) ? TimeUnit.MINUTE : ((TimeUnit) evalTimeout
736                    .getVariable("timeunit"));
737            switch (tmp) {
738                case HOUR:
739                    val = String.valueOf(t * 60);
740                    break;
741                case DAY:
742                    val = String.valueOf(t * 60 * 24);
743                    break;
744                case MONTH:
745                    val = String.valueOf(t * 60 * 24 * 30);
746                    break;
747                default:
748                    break;
749            }
750        }
751        else {
752            val = ConfigurationService.get(CONF_DEFAULT_TIMEOUT_NORMAL);
753        }
754
755        ival = ParamChecker.checkInteger(val, "timeout");
756        if (ival < 0 || ival > ConfigurationService.getInt(CONF_DEFAULT_MAX_TIMEOUT)) {
757            ival = ConfigurationService.getInt(CONF_DEFAULT_MAX_TIMEOUT);
758        }
759        coordJob.setTimeout(ival);
760
761        val = resolveTagContents("concurrency", eAppXml.getChild("controls", eAppXml.getNamespace()), evalNofuncs);
762        if (val == null || val.isEmpty()) {
763            val = ConfigurationService.get(CONF_DEFAULT_CONCURRENCY);
764        }
765        ival = ParamChecker.checkInteger(val, "concurrency");
766        coordJob.setConcurrency(ival);
767
768        val = resolveTagContents("throttle", eAppXml.getChild("controls", eAppXml.getNamespace()), evalNofuncs);
769        if (val == null || val.isEmpty()) {
770            int defaultThrottle = ConfigurationService.getInt(CONF_DEFAULT_THROTTLE);
771            ival = defaultThrottle;
772        }
773        else {
774            ival = ParamChecker.checkInteger(val, "throttle");
775        }
776        int maxQueue = ConfigurationService.getInt(CONF_QUEUE_SIZE);
777        float factor = ConfigurationService.getFloat(CONF_MAT_THROTTLING_FACTOR);
778        int maxThrottle = (int) (maxQueue * factor);
779        if (ival > maxThrottle || ival < 1) {
780            ival = maxThrottle;
781        }
782        LOG.debug("max throttle " + ival);
783        coordJob.setMatThrottling(ival);
784
785        val = resolveTagContents("execution", eAppXml.getChild("controls", eAppXml.getNamespace()), evalNofuncs);
786        if (val == "") {
787            val = Execution.FIFO.toString();
788        }
789        coordJob.setExecutionOrder(Execution.valueOf(val));
790        String[] acceptedVals = { Execution.LIFO.toString(), Execution.FIFO.toString(), Execution.LAST_ONLY.toString(),
791            Execution.NONE.toString()};
792        ParamChecker.isMember(val, acceptedVals, "execution");
793
794        // datasets
795        resolveTagContents("include", eAppXml.getChild("datasets", eAppXml.getNamespace()), evalNofuncs);
796        // for each data set
797        resolveDataSets(eAppXml);
798        HashMap<String, String> dataNameList = new HashMap<String, String>();
799        resolveIODataset(eAppXml);
800        resolveIOEvents(eAppXml, dataNameList);
801
802        resolveTagContents("app-path", eAppXml.getChild("action", eAppXml.getNamespace()).getChild("workflow",
803                eAppXml.getNamespace()), evalNofuncs);
804        // TODO: If action or workflow tag is missing, NullPointerException will
805        // occur
806        Element configElem = eAppXml.getChild("action", eAppXml.getNamespace()).getChild("workflow",
807                eAppXml.getNamespace()).getChild("configuration", eAppXml.getNamespace());
808        evalData = CoordELEvaluator.createELEvaluatorForDataEcho(conf, "coord-job-submit-data", dataNameList);
809        if (configElem != null) {
810            for (Element propElem : (List<Element>) configElem.getChildren("property", configElem.getNamespace())) {
811                resolveTagContents("name", propElem, evalData);
812                // Want to check the data-integrity but don't want to modify the
813                // XML
814                // for properties only
815                Element tmpProp = (Element) propElem.clone();
816                resolveTagContents("value", tmpProp, evalData);
817            }
818        }
819        evalSla = CoordELEvaluator.createELEvaluatorForDataAndConf(conf, "coord-sla-submit", dataNameList);
820        resolveSLA(eAppXml, coordJob);
821        return eAppXml;
822    }
823
824    /**
825     * Resolve SLA events
826     *
827     * @param eAppXml job XML
828     * @param coordJob coordinator job bean
829     * @throws CommandException thrown if failed to resolve sla events
830     */
831    private void resolveSLA(Element eAppXml, CoordinatorJobBean coordJob) throws CommandException {
832        Element eSla = XmlUtils.getSLAElement(eAppXml.getChild("action", eAppXml.getNamespace()));
833
834        if (eSla != null) {
835            String slaXml = XmlUtils.prettyPrint(eSla).toString();
836            try {
837                // EL evaluation
838                slaXml = evalSla.evaluate(slaXml, String.class);
839                // Validate against semantic SXD
840                XmlUtils.validateData(slaXml, SchemaName.SLA_ORIGINAL);
841            }
842            catch (Exception e) {
843                throw new CommandException(ErrorCode.E1004, "Validation ERROR :" + e.getMessage(), e);
844            }
845        }
846    }
847
848    /**
849     * Resolve input-events/data-in and output-events/data-out tags.
850     *
851     * @param eJobOrg : Job element
852     * @throws CoordinatorJobException thrown if failed to resolve input and output events
853     */
854    @SuppressWarnings("unchecked")
855    private void resolveIOEvents(Element eJobOrg, HashMap<String, String> dataNameList) throws CoordinatorJobException {
856        // Resolving input-events/data-in
857        // Clone the job and don't update anything in the original
858        Element eJob = (Element) eJobOrg.clone();
859        Element inputList = eJob.getChild("input-events", eJob.getNamespace());
860        if (inputList != null) {
861            TreeSet<String> eventNameSet = new TreeSet<String>();
862            for (Element dataIn : (List<Element>) inputList.getChildren("data-in", eJob.getNamespace())) {
863                String dataInName = dataIn.getAttributeValue("name");
864                dataNameList.put(dataInName, "data-in");
865                // check whether there is any duplicate data-in name
866                if (eventNameSet.contains(dataInName)) {
867                    throw new RuntimeException("Duplicate dataIn name " + dataInName);
868                }
869                else {
870                    eventNameSet.add(dataInName);
871                }
872                resolveTagContents("instance", dataIn, evalInst);
873                resolveTagContents("start-instance", dataIn, evalInst);
874                resolveTagContents("end-instance", dataIn, evalInst);
875
876            }
877        }
878        // Resolving output-events/data-out
879        Element outputList = eJob.getChild("output-events", eJob.getNamespace());
880        if (outputList != null) {
881            TreeSet<String> eventNameSet = new TreeSet<String>();
882            for (Element dataOut : (List<Element>) outputList.getChildren("data-out", eJob.getNamespace())) {
883                String dataOutName = dataOut.getAttributeValue("name");
884                dataNameList.put(dataOutName, "data-out");
885                // check whether there is any duplicate data-out name
886                if (eventNameSet.contains(dataOutName)) {
887                    throw new RuntimeException("Duplicate dataIn name " + dataOutName);
888                }
889                else {
890                    eventNameSet.add(dataOutName);
891                }
892                resolveTagContents("instance", dataOut, evalInst);
893
894            }
895        }
896
897    }
898
899    /**
900     * Resolve input-events/dataset and output-events/dataset tags.
901     *
902     * @param eJob : Job element
903     * @throws CoordinatorJobException thrown if failed to resolve input and output events
904     */
905    @SuppressWarnings("unchecked")
906    private void resolveIODataset(Element eAppXml) throws CoordinatorJobException {
907        // Resolving input-events/data-in
908        Element inputList = eAppXml.getChild("input-events", eAppXml.getNamespace());
909        if (inputList != null) {
910            for (Element dataIn : (List<Element>) inputList.getChildren("data-in", eAppXml.getNamespace())) {
911                resolveAttribute("dataset", dataIn, evalInst);
912
913            }
914        }
915        // Resolving output-events/data-out
916        Element outputList = eAppXml.getChild("output-events", eAppXml.getNamespace());
917        if (outputList != null) {
918            for (Element dataOut : (List<Element>) outputList.getChildren("data-out", eAppXml.getNamespace())) {
919                resolveAttribute("dataset", dataOut, evalInst);
920
921            }
922        }
923
924    }
925
926
927    /**
928     * Add an attribute into XML element.
929     *
930     * @param attrName :attribute name
931     * @param elem : Element to add attribute
932     * @param value :Value of attribute
933     */
934    private void addAnAttribute(String attrName, Element elem, String value) {
935        elem.setAttribute(attrName, value);
936    }
937
938    /**
939     * Resolve datasets using job configuration.
940     *
941     * @param eAppXml : Job Element XML
942     * @throws Exception thrown if failed to resolve datasets
943     */
944    @SuppressWarnings("unchecked")
945    private void resolveDataSets(Element eAppXml) throws Exception {
946        Element datasetList = eAppXml.getChild("datasets", eAppXml.getNamespace());
947        if (datasetList != null) {
948
949            List<Element> dsElems = datasetList.getChildren("dataset", eAppXml.getNamespace());
950            resolveDataSets(dsElems);
951            resolveTagContents("app-path", eAppXml.getChild("action", eAppXml.getNamespace()).getChild("workflow",
952                    eAppXml.getNamespace()), evalNofuncs);
953        }
954    }
955
956    /**
957     * Resolve datasets using job configuration.
958     *
959     * @param dsElems : Data set XML element.
960     * @throws CoordinatorJobException thrown if failed to resolve datasets
961     */
962    private void resolveDataSets(List<Element> dsElems) throws CoordinatorJobException {
963        for (Element dsElem : dsElems) {
964            // Setting up default TimeUnit and EndOFDuraion
965            evalFreq.setVariable("timeunit", TimeUnit.MINUTE);
966            evalFreq.setVariable("endOfDuration", TimeUnit.NONE);
967
968            String val = resolveAttribute("frequency", dsElem, evalFreq);
969            int ival = ParamChecker.checkInteger(val, "frequency");
970            ParamChecker.checkGTZero(ival, "frequency");
971            addAnAttribute("freq_timeunit", dsElem, evalFreq.getVariable("timeunit") == null ? TimeUnit.MINUTE
972                    .toString() : ((TimeUnit) evalFreq.getVariable("timeunit")).toString());
973            addAnAttribute("end_of_duration", dsElem, evalFreq.getVariable("endOfDuration") == null ? TimeUnit.NONE
974                    .toString() : ((TimeUnit) evalFreq.getVariable("endOfDuration")).toString());
975            val = resolveAttribute("initial-instance", dsElem, evalInitialInstance);
976            ParamChecker.checkDateOozieTZ(val, "initial-instance");
977            checkInitialInstance(val);
978            val = resolveAttribute("timezone", dsElem, evalNofuncs);
979            ParamChecker.checkTimeZone(val, "timezone");
980            resolveTagContents("uri-template", dsElem, evalNofuncs);
981            resolveTagContents("done-flag", dsElem, evalNofuncs);
982        }
983    }
984
985    /**
986     * Resolve the content of a tag.
987     *
988     * @param tagName : Tag name of job XML i.e. <timeout> 10 </timeout>
989     * @param elem : Element where the tag exists.
990     * @param eval : EL evealuator
991     * @return Resolved tag content.
992     * @throws CoordinatorJobException thrown if failed to resolve tag content
993     */
994    @SuppressWarnings("unchecked")
995    private String resolveTagContents(String tagName, Element elem, ELEvaluator eval) throws CoordinatorJobException {
996        String ret = "";
997        if (elem != null) {
998            for (Element tagElem : (List<Element>) elem.getChildren(tagName, elem.getNamespace())) {
999                if (tagElem != null) {
1000                    String updated;
1001                    try {
1002                        updated = CoordELFunctions.evalAndWrap(eval, tagElem.getText().trim());
1003
1004                    }
1005                    catch (Exception e) {
1006                        throw new CoordinatorJobException(ErrorCode.E1004, e.getMessage(), e);
1007                    }
1008                    tagElem.removeContent();
1009                    tagElem.addContent(updated);
1010                    ret += updated;
1011                }
1012            }
1013        }
1014        return ret;
1015    }
1016
1017    /**
1018     * Resolve an attribute value.
1019     *
1020     * @param attrName : Attribute name.
1021     * @param elem : XML Element where attribute is defiend
1022     * @param eval : ELEvaluator used to resolve
1023     * @return Resolved attribute value
1024     * @throws CoordinatorJobException thrown if failed to resolve an attribute value
1025     */
1026    private String resolveAttribute(String attrName, Element elem, ELEvaluator eval) throws CoordinatorJobException {
1027        Attribute attr = elem.getAttribute(attrName);
1028        String val = null;
1029        if (attr != null) {
1030            try {
1031                val = CoordELFunctions.evalAndWrap(eval, attr.getValue().trim());
1032            }
1033            catch (Exception e) {
1034                throw new CoordinatorJobException(ErrorCode.E1004, e.getMessage(), e);
1035            }
1036            attr.setValue(val);
1037        }
1038        return val;
1039    }
1040
1041    /**
1042     * Include referred datasets into XML.
1043     *
1044     * @param resolvedXml : Job XML element.
1045     * @param conf : Job configuration
1046     * @throws CoordinatorJobException thrown if failed to include referred datasets into XML
1047     */
1048    @SuppressWarnings("unchecked")
1049    protected void includeDataSets(Element resolvedXml, Configuration conf) throws CoordinatorJobException {
1050        Element datasets = resolvedXml.getChild("datasets", resolvedXml.getNamespace());
1051        Element allDataSets = new Element("all_datasets", resolvedXml.getNamespace());
1052        List<String> dsList = new ArrayList<String>();
1053        if (datasets != null) {
1054            for (Element includeElem : (List<Element>) datasets.getChildren("include", datasets.getNamespace())) {
1055                String incDSFile = includeElem.getTextTrim();
1056                includeOneDSFile(incDSFile, dsList, allDataSets, datasets.getNamespace());
1057            }
1058            for (Element e : (List<Element>) datasets.getChildren("dataset", datasets.getNamespace())) {
1059                String dsName = e.getAttributeValue("name");
1060                if (dsList.contains(dsName)) {// Override with this DS
1061                    // Remove duplicate
1062                    removeDataSet(allDataSets, dsName);
1063                }
1064                else {
1065                    dsList.add(dsName);
1066                }
1067                allDataSets.addContent((Element) e.clone());
1068            }
1069        }
1070        insertDataSet(resolvedXml, allDataSets);
1071        resolvedXml.removeChild("datasets", resolvedXml.getNamespace());
1072    }
1073
1074    /**
1075     * Include one dataset file.
1076     *
1077     * @param incDSFile : Include data set filename.
1078     * @param dsList :List of dataset names to verify the duplicate.
1079     * @param allDataSets : Element that includes all dataset definitions.
1080     * @param dsNameSpace : Data set name space
1081     * @throws CoordinatorJobException thrown if failed to include one dataset file
1082     */
1083    @SuppressWarnings("unchecked")
1084    private void includeOneDSFile(String incDSFile, List<String> dsList, Element allDataSets, Namespace dsNameSpace)
1085    throws CoordinatorJobException {
1086        Element tmpDataSets = null;
1087        try {
1088            String dsXml = readDefinition(incDSFile);
1089            LOG.debug("DSFILE :" + incDSFile + "\n" + dsXml);
1090            tmpDataSets = XmlUtils.parseXml(dsXml);
1091        }
1092        catch (JDOMException e) {
1093            LOG.warn("Error parsing included dataset [{0}].  Message [{1}]", incDSFile, e.getMessage());
1094            throw new CoordinatorJobException(ErrorCode.E0700, e.getMessage());
1095        }
1096        resolveDataSets(tmpDataSets.getChildren("dataset"));
1097        for (Element e : (List<Element>) tmpDataSets.getChildren("dataset")) {
1098            String dsName = e.getAttributeValue("name");
1099            if (dsList.contains(dsName)) {
1100                throw new RuntimeException("Duplicate Dataset " + dsName);
1101            }
1102            dsList.add(dsName);
1103            Element tmp = (Element) e.clone();
1104            // TODO: Don't like to over-write the external/include DS's namespace
1105            tmp.setNamespace(dsNameSpace);
1106            tmp.getChild("uri-template").setNamespace(dsNameSpace);
1107            if (e.getChild("done-flag") != null) {
1108                tmp.getChild("done-flag").setNamespace(dsNameSpace);
1109            }
1110            allDataSets.addContent(tmp);
1111        }
1112        // nested include
1113        for (Element includeElem : (List<Element>) tmpDataSets.getChildren("include", tmpDataSets.getNamespace())) {
1114            String incFile = includeElem.getTextTrim();
1115            includeOneDSFile(incFile, dsList, allDataSets, dsNameSpace);
1116        }
1117    }
1118
1119    /**
1120     * Remove a dataset from a list of dataset.
1121     *
1122     * @param eDatasets : List of dataset
1123     * @param name : Dataset name to be removed.
1124     */
1125    @SuppressWarnings("unchecked")
1126    private static void removeDataSet(Element eDatasets, String name) {
1127        for (Element eDataset : (List<Element>) eDatasets.getChildren("dataset", eDatasets.getNamespace())) {
1128            if (eDataset.getAttributeValue("name").equals(name)) {
1129                eDataset.detach();
1130                return;
1131            }
1132        }
1133        throw new RuntimeException("undefined dataset: " + name);
1134    }
1135
1136    /**
1137     * Read coordinator definition.
1138     *
1139     * @param appPath application path.
1140     * @return coordinator definition.
1141     * @throws CoordinatorJobException thrown if the definition could not be read.
1142     */
1143    protected String readDefinition(String appPath) throws CoordinatorJobException {
1144        String user = ParamChecker.notEmpty(conf.get(OozieClient.USER_NAME), OozieClient.USER_NAME);
1145        // Configuration confHadoop = CoordUtils.getHadoopConf(conf);
1146        try {
1147            URI uri = new URI(appPath);
1148            LOG.debug("user =" + user);
1149            HadoopAccessorService has = Services.get().get(HadoopAccessorService.class);
1150            Configuration fsConf = has.createJobConf(uri.getAuthority());
1151            FileSystem fs = has.createFileSystem(user, uri, fsConf);
1152            Path appDefPath = null;
1153
1154            // app path could be a directory
1155            Path path = new Path(uri.getPath());
1156            // check file exists for dataset include file, app xml already checked
1157            if (!fs.exists(path)) {
1158                throw new URISyntaxException(path.toString(), "path not existed : " + path.toString());
1159            }
1160            if (!fs.isFile(path)) {
1161                appDefPath = new Path(path, COORDINATOR_XML_FILE);
1162            } else {
1163                appDefPath = path;
1164            }
1165
1166            Reader reader = new InputStreamReader(fs.open(appDefPath));
1167            StringWriter writer = new StringWriter();
1168            IOUtils.copyCharStream(reader, writer);
1169            return writer.toString();
1170        }
1171        catch (IOException ex) {
1172            LOG.warn("IOException :" + XmlUtils.prettyPrint(conf), ex);
1173            throw new CoordinatorJobException(ErrorCode.E1001, ex.getMessage(), ex);
1174        }
1175        catch (URISyntaxException ex) {
1176            LOG.warn("URISyException :" + ex.getMessage());
1177            throw new CoordinatorJobException(ErrorCode.E1002, appPath, ex.getMessage(), ex);
1178        }
1179        catch (HadoopAccessorException ex) {
1180            throw new CoordinatorJobException(ex);
1181        }
1182        catch (Exception ex) {
1183            LOG.warn("Exception :", ex);
1184            throw new CoordinatorJobException(ErrorCode.E1001, ex.getMessage(), ex);
1185        }
1186    }
1187
1188    /**
1189     * Write a coordinator job into database
1190     *
1191     *@param appXML : Coordinator definition xml
1192     * @param eJob : XML element of job
1193     * @param coordJob : Coordinator job bean
1194     * @return Job id
1195     * @throws CommandException thrown if unable to save coordinator job to db
1196     */
1197    protected String storeToDB(String appXML, Element eJob, CoordinatorJobBean coordJob) throws CommandException {
1198        String jobId = Services.get().get(UUIDService.class).generateId(ApplicationType.COORDINATOR);
1199        coordJob.setId(jobId);
1200
1201        coordJob.setAppPath(conf.get(OozieClient.COORDINATOR_APP_PATH));
1202        coordJob.setCreatedTime(new Date());
1203        coordJob.setUser(conf.get(OozieClient.USER_NAME));
1204        String group = ConfigUtils.getWithDeprecatedCheck(conf, OozieClient.JOB_ACL, OozieClient.GROUP_NAME, null);
1205        coordJob.setGroup(group);
1206        coordJob.setConf(XmlUtils.prettyPrint(conf).toString());
1207        coordJob.setJobXml(XmlUtils.prettyPrint(eJob).toString());
1208        coordJob.setLastActionNumber(0);
1209        coordJob.setLastModifiedTime(new Date());
1210
1211        if (!dryrun) {
1212            coordJob.setLastModifiedTime(new Date());
1213            try {
1214                CoordJobQueryExecutor.getInstance().insert(coordJob);
1215            }
1216            catch (JPAExecutorException jpaee) {
1217                coordJob.setId(null);
1218                coordJob.setStatus(CoordinatorJob.Status.FAILED);
1219                throw new CommandException(jpaee);
1220            }
1221        }
1222        return jobId;
1223    }
1224
1225    /*
1226     * this method checks if the initial-instance specified for a particular
1227       is not a date earlier than the oozie server default Jan 01, 1970 00:00Z UTC
1228     */
1229    private void checkInitialInstance(String val) throws CoordinatorJobException, IllegalArgumentException {
1230        Date initialInstance, givenInstance;
1231        try {
1232            initialInstance = DateUtils.parseDateUTC("1970-01-01T00:00Z");
1233            givenInstance = DateUtils.parseDateOozieTZ(val);
1234        }
1235        catch (Exception e) {
1236            throw new IllegalArgumentException("Unable to parse dataset initial-instance string '" + val +
1237                                               "' to Date object. ",e);
1238        }
1239        if(givenInstance.compareTo(initialInstance) < 0) {
1240            throw new CoordinatorJobException(ErrorCode.E1021, "Dataset initial-instance " + val +
1241                    " is earlier than the default initial instance " + DateUtils.formatDateOozieTZ(initialInstance));
1242        }
1243    }
1244
1245    /* (non-Javadoc)
1246     * @see org.apache.oozie.command.XCommand#getEntityKey()
1247     */
1248    @Override
1249    public String getEntityKey() {
1250        return null;
1251    }
1252
1253    /* (non-Javadoc)
1254     * @see org.apache.oozie.command.XCommand#isLockRequired()
1255     */
1256    @Override
1257    protected boolean isLockRequired() {
1258        return false;
1259    }
1260
1261    /* (non-Javadoc)
1262     * @see org.apache.oozie.command.XCommand#loadState()
1263     */
1264    @Override
1265    protected void loadState() throws CommandException {
1266        jpaService = Services.get().get(JPAService.class);
1267        if (jpaService == null) {
1268            throw new CommandException(ErrorCode.E0610);
1269        }
1270        coordJob = new CoordinatorJobBean();
1271        if (this.bundleId != null) {
1272            // this coord job is created from bundle
1273            coordJob.setBundleId(this.bundleId);
1274            // first use bundle id if submit thru bundle
1275            LogUtils.setLogInfo(this.bundleId);
1276        }
1277        if (this.coordName != null) {
1278            // this coord job is created from bundle
1279            coordJob.setAppName(this.coordName);
1280        }
1281        setJob(coordJob);
1282
1283    }
1284
1285    /* (non-Javadoc)
1286     * @see org.apache.oozie.command.XCommand#verifyPrecondition()
1287     */
1288    @Override
1289    protected void verifyPrecondition() throws CommandException {
1290
1291    }
1292
1293    /* (non-Javadoc)
1294     * @see org.apache.oozie.command.TransitionXCommand#notifyParent()
1295     */
1296    @Override
1297    public void notifyParent() throws CommandException {
1298        // update bundle action
1299        if (coordJob.getBundleId() != null) {
1300            LOG.debug("Updating bundle record: " + coordJob.getBundleId() + " for coord id: " + coordJob.getId());
1301            BundleStatusUpdateXCommand bundleStatusUpdate = new BundleStatusUpdateXCommand(coordJob, prevStatus);
1302            bundleStatusUpdate.call();
1303        }
1304    }
1305
1306    /* (non-Javadoc)
1307     * @see org.apache.oozie.command.TransitionXCommand#updateJob()
1308     */
1309    @Override
1310    public void updateJob() throws CommandException {
1311    }
1312
1313    /* (non-Javadoc)
1314     * @see org.apache.oozie.command.TransitionXCommand#getJob()
1315     */
1316    @Override
1317    public Job getJob() {
1318        return coordJob;
1319    }
1320
1321    @Override
1322    public void performWrites() throws CommandException {
1323    }
1324}