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.coord;
020
021import com.google.common.collect.Lists;
022import org.apache.commons.lang.StringUtils;
023import org.apache.hadoop.conf.Configuration;
024import org.apache.oozie.ErrorCode;
025import org.apache.oozie.client.OozieClient;
026import org.apache.oozie.command.CommandException;
027import org.apache.oozie.dependency.URIHandler;
028import org.apache.oozie.dependency.URIHandler.Context;
029import org.apache.oozie.service.Services;
030import org.apache.oozie.service.URIHandlerService;
031import org.apache.oozie.util.DateUtils;
032import org.apache.oozie.util.ELEvaluator;
033import org.apache.oozie.util.ParamChecker;
034import org.apache.oozie.util.XLog;
035
036import java.net.URI;
037import java.util.ArrayList;
038import java.util.Calendar;
039import java.util.Date;
040import java.util.GregorianCalendar;
041import java.util.List;
042import java.util.TimeZone;
043
044/**
045 * This class implements the EL function related to coordinator
046 */
047
048public class CoordELFunctions {
049    final public static String DATASET = "oozie.coord.el.dataset.bean";
050    final public static String COORD_ACTION = "oozie.coord.el.app.bean";
051    final public static String CONFIGURATION = "oozie.coord.el.conf";
052    final public static String LATEST_EL_USE_CURRENT_TIME = "oozie.service.ELService.latest-el.use-current-time";
053    // INSTANCE_SEPARATOR is used to separate multiple directories into one tag.
054    final public static String INSTANCE_SEPARATOR = "#";
055    final public static String DIR_SEPARATOR = ",";
056    // TODO: in next release, support flexibility
057    private static String END_OF_OPERATION_INDICATOR_FILE = "_SUCCESS";
058
059    public static final long MINUTE_MSEC = 60 * 1000L;
060    public static final long HOUR_MSEC = 60 * MINUTE_MSEC;
061    public static final long DAY_MSEC = 24 * HOUR_MSEC;
062    public static final long MONTH_MSEC = 30 * DAY_MSEC;
063    public static final long YEAR_MSEC = 365 * DAY_MSEC;
064
065    /**
066     * Used in defining the frequency in 'day' unit. <p/> domain: <code> val &gt; 0</code> and should be integer.
067     *
068     * @param val frequency in number of days.
069     * @return number of days and also set the frequency timeunit to "day"
070     */
071    public static int ph1_coord_days(int val) {
072        val = ParamChecker.checkGTZero(val, "n");
073        ELEvaluator eval = ELEvaluator.getCurrent();
074        eval.setVariable("timeunit", TimeUnit.DAY);
075        eval.setVariable("endOfDuration", TimeUnit.NONE);
076        return val;
077    }
078
079    /**
080     * Used in defining the frequency in 'month' unit. <p/> domain: <code> val &gt; 0</code> and should be integer.
081     *
082     * @param val frequency in number of months.
083     * @return number of months and also set the frequency timeunit to "month"
084     */
085    public static int ph1_coord_months(int val) {
086        val = ParamChecker.checkGTZero(val, "n");
087        ELEvaluator eval = ELEvaluator.getCurrent();
088        eval.setVariable("timeunit", TimeUnit.MONTH);
089        eval.setVariable("endOfDuration", TimeUnit.NONE);
090        return val;
091    }
092
093    /**
094     * Used in defining the frequency in 'hour' unit. <p/> parameter value domain: <code> val &gt; 0</code> and should
095     * be integer.
096     *
097     * @param val frequency in number of hours.
098     * @return number of minutes and also set the frequency timeunit to "minute"
099     */
100    public static int ph1_coord_hours(int val) {
101        val = ParamChecker.checkGTZero(val, "n");
102        ELEvaluator eval = ELEvaluator.getCurrent();
103        eval.setVariable("timeunit", TimeUnit.MINUTE);
104        eval.setVariable("endOfDuration", TimeUnit.NONE);
105        return val * 60;
106    }
107
108    /**
109     * Used in defining the frequency in 'minute' unit. <p/> domain: <code> val &gt; 0</code> and should be integer.
110     *
111     * @param val frequency in number of minutes.
112     * @return number of minutes and also set the frequency timeunit to "minute"
113     */
114    public static int ph1_coord_minutes(int val) {
115        val = ParamChecker.checkGTZero(val, "n");
116        ELEvaluator eval = ELEvaluator.getCurrent();
117        eval.setVariable("timeunit", TimeUnit.MINUTE);
118        eval.setVariable("endOfDuration", TimeUnit.NONE);
119        return val;
120    }
121
122    /**
123     * Used in defining the frequency in 'day' unit and specify the "end of day" property. <p/> Every instance will
124     * start at 00:00 hour of each day. <p/> domain: <code> val &gt; 0</code> and should be integer.
125     *
126     * @param val frequency in number of days.
127     * @return number of days and also set the frequency timeunit to "day" and end_of_duration flag to "day"
128     */
129    public static int ph1_coord_endOfDays(int val) {
130        val = ParamChecker.checkGTZero(val, "n");
131        ELEvaluator eval = ELEvaluator.getCurrent();
132        eval.setVariable("timeunit", TimeUnit.DAY);
133        eval.setVariable("endOfDuration", TimeUnit.END_OF_DAY);
134        return val;
135    }
136
137    /**
138     * Used in defining the frequency in 'month' unit and specify the "end of month" property. <p/> Every instance will
139     * start at first day of each month at 00:00 hour. <p/> domain: <code> val &gt; 0</code> and should be integer.
140     *
141     * @param val: frequency in number of months.
142     * @return number of months and also set the frequency timeunit to "month" and end_of_duration flag to "month"
143     */
144    public static int ph1_coord_endOfMonths(int val) {
145        val = ParamChecker.checkGTZero(val, "n");
146        ELEvaluator eval = ELEvaluator.getCurrent();
147        eval.setVariable("timeunit", TimeUnit.MONTH);
148        eval.setVariable("endOfDuration", TimeUnit.END_OF_MONTH);
149        return val;
150    }
151
152    /**
153     * Calculate the difference of timezone offset in minutes between dataset and coordinator job. <p/> Depends on: <p/>
154     * 1. Timezone of both dataset and job <p/> 2. Action creation Time
155     *
156     * @return difference in minutes (DataSet TZ Offset - Application TZ offset)
157     */
158    public static int ph2_coord_tzOffset() {
159        long actionCreationTime = getActionCreationtime().getTime();
160        TimeZone dsTZ = ParamChecker.notNull(getDatasetTZ(), "DatasetTZ");
161        TimeZone jobTZ = ParamChecker.notNull(getJobTZ(), "JobTZ");
162        return (dsTZ.getOffset(actionCreationTime) - jobTZ.getOffset(actionCreationTime)) / (1000 * 60);
163    }
164
165    public static int ph3_coord_tzOffset() {
166        return ph2_coord_tzOffset();
167    }
168
169    /**
170     * Returns a date string that is offset from 'strBaseDate' by the amount specified.  The unit can be one of
171     * DAY, MONTH, HOUR, MINUTE, MONTH.
172     *
173     * @param strBaseDate The base date
174     * @param offset any number
175     * @param unit one of DAY, MONTH, HOUR, MINUTE, MONTH
176     * @return the offset date string
177     * @throws Exception
178     */
179    public static String ph2_coord_dateOffset(String strBaseDate, int offset, String unit) throws Exception {
180        Calendar baseCalDate = DateUtils.getCalendar(strBaseDate);
181        StringBuilder buffer = new StringBuilder();
182        baseCalDate.add(TimeUnit.valueOf(unit).getCalendarUnit(), offset);
183        buffer.append(DateUtils.formatDateOozieTZ(baseCalDate));
184        return buffer.toString();
185    }
186
187    public static String ph3_coord_dateOffset(String strBaseDate, int offset, String unit) throws Exception {
188        return ph2_coord_dateOffset(strBaseDate, offset, unit);
189    }
190
191    /**
192     * Returns a date string that is offset from 'strBaseDate' by the difference from Oozie processing timezone to the given
193     * timezone. It will account for daylight saving time based on the given 'strBaseDate' and 'timezone'.
194     *
195     * @param strBaseDate The base date
196     * @param timezone
197     * @return the offset date string
198     * @throws Exception
199     */
200    public static String ph2_coord_dateTzOffset(String strBaseDate, String timezone) throws Exception {
201        Calendar baseCalDate = DateUtils.getCalendar(strBaseDate);
202        StringBuilder buffer = new StringBuilder();
203        baseCalDate.setTimeZone(DateUtils.getTimeZone(timezone));
204        buffer.append(DateUtils.formatDate(baseCalDate));
205        return buffer.toString();
206    }
207
208    public static String ph3_coord_dateTzOffset(String strBaseDate, String timezone) throws Exception{
209        return ph2_coord_dateTzOffset(strBaseDate, timezone);
210    }
211
212    /**
213     * Determine the date-time in Oozie processing timezone of n-th future available dataset instance
214     * from nominal Time but not beyond the instance specified as 'instance.
215     * <p/>
216     * It depends on:
217     * <p/>
218     * 1. Data set frequency
219     * <p/>
220     * 2. Data set Time unit (day, month, minute)
221     * <p/>
222     * 3. Data set Time zone/DST
223     * <p/>
224     * 4. End Day/Month flag
225     * <p/>
226     * 5. Data set initial instance
227     * <p/>
228     * 6. Action Creation Time
229     * <p/>
230     * 7. Existence of dataset's directory
231     *
232     * @param n :instance count
233     *        <p/>
234     *        domain: n >= 0, n is integer
235     * @param instance: How many future instance it should check? value should
236     *        be >=0
237     * @return date-time in Oozie processing timezone of the n-th instance
238     *         <p/>
239     * @throws Exception
240     */
241    public static String ph3_coord_future(int n, int instance) throws Exception {
242        ParamChecker.checkGEZero(n, "future:n");
243        ParamChecker.checkGTZero(instance, "future:instance");
244        if (isSyncDataSet()) {// For Sync Dataset
245            return coord_future_sync(n, instance);
246        }
247        else {
248            throw new UnsupportedOperationException("Asynchronous Dataset is not supported yet");
249        }
250    }
251
252    /**
253     * Determine the date-time in Oozie processing timezone of the future available dataset instances
254     * from start to end offsets from nominal Time but not beyond the instance specified as 'instance'.
255     * <p/>
256     * It depends on:
257     * <p/>
258     * 1. Data set frequency
259     * <p/>
260     * 2. Data set Time unit (day, month, minute)
261     * <p/>
262     * 3. Data set Time zone/DST
263     * <p/>
264     * 4. End Day/Month flag
265     * <p/>
266     * 5. Data set initial instance
267     * <p/>
268     * 6. Action Creation Time
269     * <p/>
270     * 7. Existence of dataset's directory
271     *
272     * @param start : start instance offset
273     *        <p/>
274     *        domain: start >= 0, start is integer
275     * @param end : end instance offset
276     *        <p/>
277     *        domain: end >= 0, end is integer
278     * @param instance: How many future instance it should check? value should
279     *        be >=0
280     * @return date-time in Oozie processing timezone of the instances from start to end offsets
281     *        delimited by comma.
282     *         <p/>
283     * @throws Exception
284     */
285    public static String ph3_coord_futureRange(int start, int end, int instance) throws Exception {
286        ParamChecker.checkGEZero(start, "future:n");
287        ParamChecker.checkGEZero(end, "future:n");
288        ParamChecker.checkGTZero(instance, "future:instance");
289        if (isSyncDataSet()) {// For Sync Dataset
290            return coord_futureRange_sync(start, end, instance);
291        }
292        else {
293            throw new UnsupportedOperationException("Asynchronous Dataset is not supported yet");
294        }
295    }
296
297    private static String coord_future_sync(int n, int instance) throws Exception {
298        return coord_futureRange_sync(n, n, instance);
299    }
300
301    private static String coord_futureRange_sync(int startOffset, int endOffset, int instance) throws Exception {
302        final XLog LOG = XLog.getLog(CoordELFunctions.class);
303        final Thread currentThread = Thread.currentThread();
304        ELEvaluator eval = ELEvaluator.getCurrent();
305        String retVal = "";
306        int datasetFrequency = (int) getDSFrequency();// in minutes
307        TimeUnit dsTimeUnit = getDSTimeUnit();
308        int[] instCount = new int[1];
309        Calendar nominalInstanceCal = getCurrentInstance(getActionCreationtime(), instCount);
310        StringBuilder resolvedInstances = new StringBuilder();
311        StringBuilder resolvedURIPaths = new StringBuilder();
312        if (nominalInstanceCal != null) {
313            Calendar initInstance = getInitialInstanceCal();
314            nominalInstanceCal = (Calendar) initInstance.clone();
315            nominalInstanceCal.add(dsTimeUnit.getCalendarUnit(), instCount[0] * datasetFrequency);
316
317            SyncCoordDataset ds = (SyncCoordDataset) eval.getVariable(DATASET);
318            if (ds == null) {
319                throw new RuntimeException("Associated Dataset should be defined with key " + DATASET);
320            }
321            String uriTemplate = ds.getUriTemplate();
322            Configuration conf = (Configuration) eval.getVariable(CONFIGURATION);
323            if (conf == null) {
324                throw new RuntimeException("Associated Configuration should be defined with key " + CONFIGURATION);
325            }
326            int available = 0, checkedInstance = 0;
327            boolean resolved = false;
328            String user = ParamChecker
329                    .notEmpty((String) eval.getVariable(OozieClient.USER_NAME), OozieClient.USER_NAME);
330            String doneFlag = ds.getDoneFlag();
331            URIHandlerService uriService = Services.get().get(URIHandlerService.class);
332            URIHandler uriHandler = null;
333            Context uriContext = null;
334            try {
335                while (instance >= checkedInstance && !currentThread.isInterrupted()) {
336                    ELEvaluator uriEval = getUriEvaluator(nominalInstanceCal);
337                    String uriPath = uriEval.evaluate(uriTemplate, String.class);
338                    if (uriHandler == null) {
339                        URI uri = new URI(uriPath);
340                        uriHandler = uriService.getURIHandler(uri);
341                        uriContext = uriHandler.getContext(uri, conf, user);
342                    }
343                    String uriWithDoneFlag = uriHandler.getURIWithDoneFlag(uriPath, doneFlag);
344                    if (uriHandler.exists(new URI(uriWithDoneFlag), uriContext)) {
345                        if (available == endOffset) {
346                            LOG.debug("Matched future(" + available + "): " + uriWithDoneFlag);
347                            resolved = true;
348                            resolvedInstances.append(DateUtils.formatDateOozieTZ(nominalInstanceCal));
349                            resolvedURIPaths.append(uriPath);
350                            retVal = resolvedInstances.toString();
351                            eval.setVariable("resolved_path", resolvedURIPaths.toString());
352                            break;
353                        }
354                        else if (available >= startOffset) {
355                            LOG.debug("Matched future(" + available + "): " + uriWithDoneFlag);
356                            resolvedInstances.append(DateUtils.formatDateOozieTZ(nominalInstanceCal)).append(
357                                    INSTANCE_SEPARATOR);
358                            resolvedURIPaths.append(uriPath).append(INSTANCE_SEPARATOR);
359                        }
360                        available++;
361                    }
362                    // nominalInstanceCal.add(dsTimeUnit.getCalendarUnit(), datasetFrequency);
363                    nominalInstanceCal = (Calendar) initInstance.clone();
364                    instCount[0]++;
365                    nominalInstanceCal.add(dsTimeUnit.getCalendarUnit(), instCount[0] * datasetFrequency);
366                    checkedInstance++;
367                    // DateUtils.moveToEnd(nominalInstanceCal, getDSEndOfFlag());
368                }
369            }
370            finally {
371                if (uriContext != null) {
372                    uriContext.destroy();
373                }
374            }
375            if (!resolved) {
376                // return unchanged future function with variable 'is_resolved'
377                // to 'false'
378                eval.setVariable("is_resolved", Boolean.FALSE);
379                if (startOffset == endOffset) {
380                    retVal = "${coord:future(" + startOffset + ", " + instance + ")}";
381                }
382                else {
383                    retVal = "${coord:futureRange(" + startOffset + ", " + endOffset + ", " + instance + ")}";
384                }
385            }
386            else {
387                eval.setVariable("is_resolved", Boolean.TRUE);
388            }
389        }
390        else {// No feasible nominal time
391            eval.setVariable("is_resolved", Boolean.TRUE);
392            retVal = "";
393        }
394        return retVal;
395    }
396
397    /**
398     * Return nominal time or Action Creation Time.
399     * <p/>
400     *
401     * @return coordinator action creation or materialization date time
402     * @throws Exception if unable to format the Date object to String
403     */
404    public static String ph2_coord_nominalTime() throws Exception {
405        ELEvaluator eval = ELEvaluator.getCurrent();
406        SyncCoordAction action = ParamChecker.notNull((SyncCoordAction) eval.getVariable(COORD_ACTION),
407                "Coordinator Action");
408        return DateUtils.formatDateOozieTZ(action.getNominalTime());
409    }
410
411    public static String ph3_coord_nominalTime() throws Exception {
412        return ph2_coord_nominalTime();
413    }
414
415    /**
416     * Convert from standard date-time formatting to a desired format.
417     * <p/>
418     * @param dateTimeStr - A timestamp in standard (ISO8601) format.
419     * @param format - A string representing the desired format.
420     * @return coordinator action creation or materialization date time
421     * @throws Exception if unable to format the Date object to String
422     */
423    public static String ph2_coord_formatTime(String dateTimeStr, String format)
424            throws Exception {
425        Date dateTime = DateUtils.parseDateOozieTZ(dateTimeStr);
426        return DateUtils.formatDateCustom(dateTime, format);
427    }
428
429    public static String ph3_coord_formatTime(String dateTimeStr, String format)
430            throws Exception {
431        return ph2_coord_formatTime(dateTimeStr, format);
432    }
433
434    /**
435     * Return Action Id. <p/>
436     *
437     * @return coordinator action Id
438     */
439    public static String ph2_coord_actionId() throws Exception {
440        ELEvaluator eval = ELEvaluator.getCurrent();
441        SyncCoordAction action = ParamChecker.notNull((SyncCoordAction) eval.getVariable(COORD_ACTION),
442                "Coordinator Action");
443        return action.getActionId();
444    }
445
446    public static String ph3_coord_actionId() throws Exception {
447        return ph2_coord_actionId();
448    }
449
450    /**
451     * Return Job Name. <p/>
452     *
453     * @return coordinator name
454     */
455    public static String ph2_coord_name() throws Exception {
456        ELEvaluator eval = ELEvaluator.getCurrent();
457        SyncCoordAction action = ParamChecker.notNull((SyncCoordAction) eval.getVariable(COORD_ACTION),
458                "Coordinator Action");
459        return action.getName();
460    }
461
462    public static String ph3_coord_name() throws Exception {
463        return ph2_coord_name();
464    }
465
466    /**
467     * Return Action Start time. <p/>
468     *
469     * @return coordinator action start time
470     * @throws Exception if unable to format the Date object to String
471     */
472    public static String ph2_coord_actualTime() throws Exception {
473        ELEvaluator eval = ELEvaluator.getCurrent();
474        SyncCoordAction coordAction = (SyncCoordAction) eval.getVariable(COORD_ACTION);
475        if (coordAction == null) {
476            throw new RuntimeException("Associated Application instance should be defined with key " + COORD_ACTION);
477        }
478        return DateUtils.formatDateOozieTZ(coordAction.getActualTime());
479    }
480
481    public static String ph3_coord_actualTime() throws Exception {
482        return ph2_coord_actualTime();
483    }
484
485    /**
486     * Used to specify a list of URI's that are used as input dir to the workflow job. <p/> Look for two evaluator-level
487     * variables <p/> A) .datain.<DATAIN_NAME> B) .datain.<DATAIN_NAME>.unresolved <p/> A defines the current list of
488     * URI. <p/> B defines whether there are any unresolved EL-function (i.e latest) <p/> If there are something
489     * unresolved, this function will echo back the original function <p/> otherwise it sends the uris.
490     *
491     * @param dataInName : Datain name
492     * @return the list of URI's separated by INSTANCE_SEPARATOR <p/> if there are unresolved EL function (i.e. latest)
493     *         , echo back <p/> the function without resolving the function.
494     */
495    public static String ph3_coord_dataIn(String dataInName) {
496        String uris = "";
497        ELEvaluator eval = ELEvaluator.getCurrent();
498        uris = (String) eval.getVariable(".datain." + dataInName);
499        Boolean unresolved = (Boolean) eval.getVariable(".datain." + dataInName + ".unresolved");
500        if (unresolved != null && unresolved.booleanValue() == true) {
501            return "${coord:dataIn('" + dataInName + "')}";
502        }
503        return uris;
504    }
505
506    /**
507     * Used to specify a list of URI's that are output dir of the workflow job. <p/> Look for one evaluator-level
508     * variable <p/> dataout.<DATAOUT_NAME> <p/> It defines the current list of URI. <p/> otherwise it sends the uris.
509     *
510     * @param dataOutName : Dataout name
511     * @return the list of URI's separated by INSTANCE_SEPARATOR
512     */
513    public static String ph3_coord_dataOut(String dataOutName) {
514        String uris = "";
515        ELEvaluator eval = ELEvaluator.getCurrent();
516        uris = (String) eval.getVariable(".dataout." + dataOutName);
517        return uris;
518    }
519
520    /**
521     * Determine the date-time in Oozie processing timezone of n-th dataset instance. <p/> It depends on: <p/> 1.
522     * Data set frequency <p/> 2.
523     * Data set Time unit (day, month, minute) <p/> 3. Data set Time zone/DST <p/> 4. End Day/Month flag <p/> 5. Data
524     * set initial instance <p/> 6. Action Creation Time
525     *
526     * @param n instance count domain: n is integer
527     * @return date-time in Oozie processing timezone of the n-th instance returns 'null' means n-th instance is
528     * earlier than Initial-Instance of DS
529     * @throws Exception
530     */
531    public static String ph2_coord_current(int n) throws Exception {
532        if (isSyncDataSet()) { // For Sync Dataset
533            return coord_current_sync(n);
534        }
535        else {
536            throw new UnsupportedOperationException("Asynchronous Dataset is not supported yet");
537        }
538    }
539
540    /**
541     * Determine the date-time in Oozie processing timezone of current dataset instances
542     * from start to end offsets from the nominal time. <p/> It depends
543     * on: <p/> 1. Data set frequency <p/> 2. Data set Time unit (day, month, minute) <p/> 3. Data set Time zone/DST
544     * <p/> 4. End Day/Month flag <p/> 5. Data set initial instance <p/> 6. Action Creation Time
545     *
546     * @param start :start instance offset <p/> domain: start <= 0, start is integer
547     * @param end :end instance offset <p/> domain: end <= 0, end is integer
548     * @return date-time in Oozie processing timezone of the instances from start to end offsets
549     *        delimited by comma. <p/> If the current instance time of the dataset based on the Action Creation Time
550     *        is earlier than the Initial-Instance of DS an empty string is returned.
551     *        If an instance within the range is earlier than Initial-Instance of DS that instance is ignored
552     * @throws Exception
553     */
554    public static String ph2_coord_currentRange(int start, int end) throws Exception {
555        if (isSyncDataSet()) { // For Sync Dataset
556            return coord_currentRange_sync(start, end);
557        }
558        else {
559            throw new UnsupportedOperationException("Asynchronous Dataset is not supported yet");
560        }
561    }
562    /**
563     * Determine the date-time in Oozie processing timezone of the given offset from the dataset effective nominal time. <p/> It
564     * depends on: <p> 1. Data set frequency <p/> 2. Data set Time Unit <p/> 3. Data set Time zone/DST
565     * <p/> 4. Data set initial instance <p/> 5. Action Creation Time
566     *
567     * @param n offset amount (integer)
568     * @param timeUnit TimeUnit for offset n ("MINUTE", "HOUR", "DAY", "MONTH", "YEAR")
569     * @return date-time in Oozie processing timezone of the given offset from the dataset effective nominal time
570     * @throws Exception if there was a problem formatting
571     */
572    public static String ph2_coord_offset(int n, String timeUnit) throws Exception {
573        if (isSyncDataSet()) { // For Sync Dataset
574            return coord_offset_sync(n, timeUnit);
575        }
576        else {
577            throw new UnsupportedOperationException("Asynchronous Dataset is not supported yet");
578        }
579    }
580
581    /**
582     * Determine how many hours is on the date of n-th dataset instance. <p/> It depends on: <p/> 1. Data set frequency
583     * <p/> 2. Data set Time unit (day, month, minute) <p/> 3. Data set Time zone/DST <p/> 4. End Day/Month flag <p/> 5.
584     * Data set initial instance <p/> 6. Action Creation Time
585     *
586     * @param n instance count <p/> domain: n is integer
587     * @return number of hours on that day <p/> returns -1 means n-th instance is earlier than Initial-Instance of DS
588     * @throws Exception
589     */
590    public static int ph2_coord_hoursInDay(int n) throws Exception {
591        int datasetFrequency = (int) getDSFrequency();
592        // /Calendar nominalInstanceCal =
593        // getCurrentInstance(getActionCreationtime());
594        Calendar nominalInstanceCal = getEffectiveNominalTime();
595        if (nominalInstanceCal == null) {
596            return -1;
597        }
598        nominalInstanceCal.add(getDSTimeUnit().getCalendarUnit(), datasetFrequency * n);
599        /*
600         * if (nominalInstanceCal.getTime().compareTo(getInitialInstance()) < 0)
601         * { return -1; }
602         */
603        nominalInstanceCal.setTimeZone(getDatasetTZ());// Use Dataset TZ
604        // DateUtils.moveToEnd(nominalInstanceCal, getDSEndOfFlag());
605        return DateUtils.hoursInDay(nominalInstanceCal);
606    }
607
608    public static int ph3_coord_hoursInDay(int n) throws Exception {
609        return ph2_coord_hoursInDay(n);
610    }
611
612    /**
613     * Calculate number of days in one month for n-th dataset instance. <p/> It depends on: <p/> 1. Data set frequency .
614     * <p/> 2. Data set Time unit (day, month, minute) <p/> 3. Data set Time zone/DST <p/> 4. End Day/Month flag <p/> 5.
615     * Data set initial instance <p/> 6. Action Creation Time
616     *
617     * @param n instance count. domain: n is integer
618     * @return number of days in that month <p/> returns -1 means n-th instance is earlier than Initial-Instance of DS
619     * @throws Exception
620     */
621    public static int ph2_coord_daysInMonth(int n) throws Exception {
622        int datasetFrequency = (int) getDSFrequency();// in minutes
623        // Calendar nominalInstanceCal =
624        // getCurrentInstance(getActionCreationtime());
625        Calendar nominalInstanceCal = getEffectiveNominalTime();
626        if (nominalInstanceCal == null) {
627            return -1;
628        }
629        nominalInstanceCal.add(getDSTimeUnit().getCalendarUnit(), datasetFrequency * n);
630        /*
631         * if (nominalInstanceCal.getTime().compareTo(getInitialInstance()) < 0)
632         * { return -1; }
633         */
634        nominalInstanceCal.setTimeZone(getDatasetTZ());// Use Dataset TZ
635        // DateUtils.moveToEnd(nominalInstanceCal, getDSEndOfFlag());
636        return nominalInstanceCal.getActualMaximum(Calendar.DAY_OF_MONTH);
637    }
638
639    public static int ph3_coord_daysInMonth(int n) throws Exception {
640        return ph2_coord_daysInMonth(n);
641    }
642
643    /**
644     * Determine the date-time in Oozie processing timezone of n-th latest available dataset instance. <p/> It depends
645     * on: <p/> 1. Data set frequency <p/> 2. Data set Time unit (day, month, minute) <p/> 3. Data set Time zone/DST
646     * <p/> 4. End Day/Month flag <p/> 5. Data set initial instance <p/> 6. Action Creation Time <p/> 7. Existence of
647     * dataset's directory
648     *
649     * @param n :instance count <p/> domain: n <= 0, n is integer
650     * @return date-time in Oozie processing timezone of the n-th instance <p/> returns 'null' means n-th instance is
651     * earlier than Initial-Instance of DS
652     * @throws Exception
653     */
654    public static String ph3_coord_latest(int n) throws Exception {
655        ParamChecker.checkLEZero(n, "latest:n");
656        if (isSyncDataSet()) {// For Sync Dataset
657            return coord_latest_sync(n);
658        }
659        else {
660            throw new UnsupportedOperationException("Asynchronous Dataset is not supported yet");
661        }
662    }
663
664    /**
665     * Determine the date-time in Oozie processing timezone of latest available dataset instances
666     * from start to end offsets from the nominal time. <p/> It depends
667     * on: <p/> 1. Data set frequency <p/> 2. Data set Time unit (day, month, minute) <p/> 3. Data set Time zone/DST
668     * <p/> 4. End Day/Month flag <p/> 5. Data set initial instance <p/> 6. Action Creation Time <p/> 7. Existence of
669     * dataset's directory
670     *
671     * @param start :start instance offset <p/> domain: start <= 0, start is integer
672     * @param end :end instance offset <p/> domain: end <= 0, end is integer
673     * @return date-time in Oozie processing timezone of the instances from start to end offsets
674     *        delimited by comma. <p/> returns 'null' means start offset instance is
675     *        earlier than Initial-Instance of DS
676     * @throws Exception
677     */
678    public static String ph3_coord_latestRange(int start, int end) throws Exception {
679        ParamChecker.checkLEZero(start, "latest:n");
680        ParamChecker.checkLEZero(end, "latest:n");
681        if (isSyncDataSet()) {// For Sync Dataset
682            return coord_latestRange_sync(start, end);
683        }
684        else {
685            throw new UnsupportedOperationException("Asynchronous Dataset is not supported yet");
686        }
687    }
688
689    /**
690     * Configure an evaluator with data set and application specific information. <p/> Helper method of associating
691     * dataset and application object
692     *
693     * @param evaluator : to set variables
694     * @param ds : Data Set object
695     * @param coordAction : Application instance
696     */
697    public static void configureEvaluator(ELEvaluator evaluator, SyncCoordDataset ds, SyncCoordAction coordAction) {
698        evaluator.setVariable(COORD_ACTION, coordAction);
699        evaluator.setVariable(DATASET, ds);
700    }
701
702    /**
703     * Helper method to wrap around with "${..}". <p/>
704     *
705     *
706     * @param eval :EL evaluator
707     * @param expr : expression to evaluate
708     * @return Resolved expression or echo back the same expression
709     * @throws Exception
710     */
711    public static String evalAndWrap(ELEvaluator eval, String expr) throws Exception {
712        try {
713            eval.setVariable(".wrap", null);
714            String result = eval.evaluate(expr, String.class);
715            if (eval.getVariable(".wrap") != null) {
716                return "${" + result + "}";
717            }
718            else {
719                return result;
720            }
721        }
722        catch (Exception e) {
723            throw new Exception("Unable to evaluate :" + expr + ":\n", e);
724        }
725    }
726
727    // Set of echo functions
728
729    public static String ph1_coord_current_echo(String n) {
730        return echoUnResolved("current", n);
731    }
732
733    public static String ph1_coord_absolute_echo(String date) {
734        return echoUnResolved("absolute", date);
735    }
736
737    public static String ph1_coord_currentRange_echo(String start, String end) {
738        return echoUnResolved("currentRange", start + ", " + end);
739    }
740
741    public static String ph1_coord_offset_echo(String n, String timeUnit) {
742        return echoUnResolved("offset", n + " , " + timeUnit);
743    }
744
745    public static String ph2_coord_current_echo(String n) {
746        return echoUnResolved("current", n);
747    }
748
749    public static String ph2_coord_currentRange_echo(String start, String end) {
750        return echoUnResolved("currentRange", start + ", " + end);
751    }
752
753    public static String ph2_coord_offset_echo(String n, String timeUnit) {
754        return echoUnResolved("offset", n + " , " + timeUnit);
755    }
756
757    public static String ph2_coord_absolute_echo(String date) {
758        return echoUnResolved("absolute", date);
759    }
760
761    public static String ph2_coord_absolute_range(String startInstance, int end) throws Exception {
762        int[] instanceCount = new int[1];
763
764        // getCurrentInstance() returns null, which means startInstance is less
765        // than initial instance
766        if (getCurrentInstance(DateUtils.getCalendar(startInstance).getTime(), instanceCount) == null) {
767            throw new CommandException(ErrorCode.E1010,
768                    "intial-instance should be equal or earlier than the start-instance. intial-instance is "
769                            + getInitialInstance() + " and start-instance is " + startInstance);
770        }
771        int[] nominalCount = new int[1];
772        if (getCurrentInstance(getActionCreationtime(), nominalCount) == null) {
773            throw new CommandException(ErrorCode.E1010,
774                    "intial-instance should be equal or earlier than the nominal time. intial-instance is "
775                            + getInitialInstance() + " and nominal time is " + getActionCreationtime());
776        }
777        // getCurrentInstance return offset relative to initial instance.
778        // start instance offset - nominal offset = start offset relative to
779        // nominal time-stamp.
780        int start = instanceCount[0] - nominalCount[0];
781        if (start > end) {
782            throw new CommandException(ErrorCode.E1010,
783                    "start-instance should be equal or earlier than the end-instance. startInstance is "
784                            + startInstance + " which is equivalent to current (" + instanceCount[0]
785                            + ") but end is specified as current (" + end + ")");
786        }
787        return ph2_coord_currentRange(start, end);
788    }
789
790    public static String ph1_coord_dateOffset_echo(String n, String offset, String unit) {
791        return echoUnResolved("dateOffset", n + " , " + offset + " , " + unit);
792    }
793
794    public static String ph1_coord_dateTzOffset_echo(String n, String timezone) {
795        return echoUnResolved("dateTzOffset", n + " , " + timezone);
796    }
797
798    public static String ph1_coord_formatTime_echo(String dateTime, String format) {
799        // Quote the dateTime value since it would contain a ':'.
800        return echoUnResolved("formatTime", "'"+dateTime+"'" + " , " + format);
801    }
802
803    public static String ph1_coord_latest_echo(String n) {
804        return echoUnResolved("latest", n);
805    }
806
807    public static String ph2_coord_latest_echo(String n) {
808        return ph1_coord_latest_echo(n);
809    }
810
811    public static String ph1_coord_future_echo(String n, String instance) {
812        return echoUnResolved("future", n + ", " + instance + "");
813    }
814
815    public static String ph2_coord_future_echo(String n, String instance) {
816        return ph1_coord_future_echo(n, instance);
817    }
818
819    public static String ph1_coord_latestRange_echo(String start, String end) {
820        return echoUnResolved("latestRange", start + ", " + end);
821    }
822
823    public static String ph2_coord_latestRange_echo(String start, String end) {
824        return ph1_coord_latestRange_echo(start, end);
825    }
826
827    public static String ph1_coord_futureRange_echo(String start, String end, String instance) {
828        return echoUnResolved("futureRange", start + ", " + end + ", " + instance);
829    }
830
831    public static String ph2_coord_futureRange_echo(String start, String end, String instance) {
832        return ph1_coord_futureRange_echo(start, end, instance);
833    }
834
835    public static String ph1_coord_dataIn_echo(String n) {
836        ELEvaluator eval = ELEvaluator.getCurrent();
837        String val = (String) eval.getVariable("oozie.dataname." + n);
838        if (val == null || val.equals("data-in") == false) {
839            XLog.getLog(CoordELFunctions.class).error("data_in_name " + n + " is not valid");
840            throw new RuntimeException("data_in_name " + n + " is not valid");
841        }
842        return echoUnResolved("dataIn", "'" + n + "'");
843    }
844
845    public static String ph1_coord_dataOut_echo(String n) {
846        ELEvaluator eval = ELEvaluator.getCurrent();
847        String val = (String) eval.getVariable("oozie.dataname." + n);
848        if (val == null || val.equals("data-out") == false) {
849            XLog.getLog(CoordELFunctions.class).error("data_out_name " + n + " is not valid");
850            throw new RuntimeException("data_out_name " + n + " is not valid");
851        }
852        return echoUnResolved("dataOut", "'" + n + "'");
853    }
854
855    public static String ph1_coord_nominalTime_echo() {
856        return echoUnResolved("nominalTime", "");
857    }
858
859    public static String ph1_coord_nominalTime_echo_wrap() {
860        // return "${coord:nominalTime()}"; // no resolution
861        return echoUnResolved("nominalTime", "");
862    }
863
864    public static String ph1_coord_nominalTime_echo_fixed() {
865        return "2009-03-06T010:00"; // Dummy resolution
866    }
867
868    public static String ph1_coord_actualTime_echo_wrap() {
869        // return "${coord:actualTime()}"; // no resolution
870        return echoUnResolved("actualTime", "");
871    }
872
873    public static String ph1_coord_actionId_echo() {
874        return echoUnResolved("actionId", "");
875    }
876
877    public static String ph1_coord_name_echo() {
878        return echoUnResolved("name", "");
879    }
880
881    // The following echo functions are not used in any phases yet
882    // They are here for future purpose.
883    public static String coord_minutes_echo(String n) {
884        return echoUnResolved("minutes", n);
885    }
886
887    public static String coord_hours_echo(String n) {
888        return echoUnResolved("hours", n);
889    }
890
891    public static String coord_days_echo(String n) {
892        return echoUnResolved("days", n);
893    }
894
895    public static String coord_endOfDay_echo(String n) {
896        return echoUnResolved("endOfDay", n);
897    }
898
899    public static String coord_months_echo(String n) {
900        return echoUnResolved("months", n);
901    }
902
903    public static String coord_endOfMonth_echo(String n) {
904        return echoUnResolved("endOfMonth", n);
905    }
906
907    public static String coord_actualTime_echo() {
908        return echoUnResolved("actualTime", "");
909    }
910
911    // This echo function will always return "24" for validation only.
912    // This evaluation ****should not**** replace the original XML
913    // Create a temporary string and validate the function
914    // This is **required** for evaluating an expression like
915    // coord:HoursInDay(0) + 3
916    // actual evaluation will happen in phase 2 or phase 3.
917    public static String ph1_coord_hoursInDay_echo(String n) {
918        return "24";
919        // return echoUnResolved("hoursInDay", n);
920    }
921
922    // This echo function will always return "30" for validation only.
923    // This evaluation ****should not**** replace the original XML
924    // Create a temporary string and validate the function
925    // This is **required** for evaluating an expression like
926    // coord:daysInMonth(0) + 3
927    // actual evaluation will happen in phase 2 or phase 3.
928    public static String ph1_coord_daysInMonth_echo(String n) {
929        // return echoUnResolved("daysInMonth", n);
930        return "30";
931    }
932
933    // This echo function will always return "3" for validation only.
934    // This evaluation ****should not**** replace the original XML
935    // Create a temporary string and validate the function
936    // This is **required** for evaluating an expression like coord:tzOffset + 2
937    // actual evaluation will happen in phase 2 or phase 3.
938    public static String ph1_coord_tzOffset_echo() {
939        // return echoUnResolved("tzOffset", "");
940        return "3";
941    }
942
943    // Local methods
944    /**
945     * @param n
946     * @return n-th instance Date-Time from current instance for data-set <p/> return empty string ("") if the
947     *         Action_Creation_time or the n-th instance <p/> is earlier than the Initial_Instance of dataset.
948     * @throws Exception
949     */
950    private static String coord_current_sync(int n) throws Exception {
951        return coord_currentRange_sync(n, n);
952    }
953
954    private static String coord_currentRange_sync(int start, int end) throws Exception {
955        final XLog LOG = XLog.getLog(CoordELFunctions.class);
956        int datasetFrequency = getDSFrequency();// in minutes
957        TimeUnit dsTimeUnit = getDSTimeUnit();
958        int[] instCount = new int[1];// used as pass by ref
959        Calendar nominalInstanceCal = getCurrentInstance(getActionCreationtime(), instCount);
960        if (nominalInstanceCal == null) {
961            LOG.warn("If the initial instance of the dataset is later than the nominal time, an empty string is"
962                    + " returned. This means that no data is available at the current-instance specified by the user"
963                    + " and the user could try modifying his initial-instance to an earlier time.");
964            return "";
965        } else {
966            Calendar initInstance = getInitialInstanceCal();
967            // Add in the reverse order - newest instance first.
968            nominalInstanceCal = (Calendar) initInstance.clone();
969            nominalInstanceCal.add(dsTimeUnit.getCalendarUnit(), (instCount[0] + start) * datasetFrequency);
970            List<String> instances = new ArrayList<String>();
971            for (int i = start; i <= end; i++) {
972                if (nominalInstanceCal.compareTo(initInstance) < 0) {
973                    LOG.warn("If the initial instance of the dataset is later than the current-instance specified,"
974                            + " such as coord:current({0}) in this case, an empty string is returned. This means that"
975                            + " no data is available at the current-instance specified by the user and the user could"
976                            + " try modifying his initial-instance to an earlier time.", start);
977                }
978                else {
979                    instances.add(DateUtils.formatDateOozieTZ(nominalInstanceCal));
980                }
981                nominalInstanceCal.add(dsTimeUnit.getCalendarUnit(), datasetFrequency);
982            }
983            instances = Lists.reverse(instances);
984            return StringUtils.join(instances, CoordELFunctions.INSTANCE_SEPARATOR);
985        }
986    }
987
988    /**
989     *
990     * @param n offset amount (integer)
991     * @param timeUnit TimeUnit for offset n ("MINUTE", "HOUR", "DAY", "MONTH", "YEAR")
992     * @return the offset time from the effective nominal time <p/> return empty string ("") if the Action_Creation_time or the
993     *         offset instance <p/> is earlier than the Initial_Instance of dataset.
994     * @throws Exception
995     */
996    private static String coord_offset_sync(int n, String timeUnit) throws Exception {
997        Calendar rawCal = resolveOffsetRawTime(n, TimeUnit.valueOf(timeUnit), null);
998        if (rawCal == null) {
999            // warning already logged by resolveOffsetRawTime()
1000            return "";
1001        }
1002
1003        int freq = getDSFrequency();
1004        TimeUnit freqUnit = getDSTimeUnit();
1005        int freqCount = 0;
1006        // We're going to manually turn back/forward cal by decrements/increments of freq and then check that it gives the same
1007        // time as rawCal; this is to check that the offset time resolves to a frequency offset of the effective nominal time
1008        // In other words, that there exists an integer x, such that coord:offset(n, timeUnit) == coord:current(x) is true
1009        // If not, then we'll "rewind" rawCal to the latest instance earlier than rawCal and use that.
1010        Calendar cal = getInitialInstanceCal();
1011        if (rawCal.before(cal)) {
1012            while (cal.after(rawCal)) {
1013                cal.add(freqUnit.getCalendarUnit(), -freq);
1014                freqCount--;
1015            }
1016        }
1017        else if (rawCal.after(cal)) {
1018            while (cal.before(rawCal)) {
1019                cal.add(freqUnit.getCalendarUnit(), freq);
1020                freqCount++;
1021            }
1022        }
1023        if (cal.before(rawCal)) {
1024            rawCal = cal;
1025        }
1026        else if (cal.after(rawCal)) {
1027            cal.add(freqUnit.getCalendarUnit(), -freq);
1028            rawCal = cal;
1029            freqCount--;
1030        }
1031        String rawCalStr = DateUtils.formatDateOozieTZ(rawCal);
1032
1033        Calendar nominalInstanceCal = getInitialInstanceCal();
1034        nominalInstanceCal.add(freqUnit.getCalendarUnit(), freq * freqCount);
1035        if (nominalInstanceCal.getTime().compareTo(getInitialInstance()) < 0) {
1036            XLog.getLog(CoordELFunctions.class).warn("If the initial instance of the dataset is later than the offset instance"
1037                    + " specified, such as coord:offset({0}, {1}) in this case, an empty string is returned. This means that no"
1038                    + " data is available at the offset instance specified by the user and the user could try modifying his"
1039                    + " initial-instance to an earlier time.", n, timeUnit);
1040            return "";
1041        }
1042        String nominalCalStr = DateUtils.formatDateOozieTZ(nominalInstanceCal);
1043
1044        if (!rawCalStr.equals(nominalCalStr)) {
1045            throw new RuntimeException("Shouldn't happen");
1046        }
1047        return rawCalStr;
1048    }
1049
1050    /**
1051     * @param offset
1052     * @return n-th available latest instance Date-Time for SYNC data-set
1053     * @throws Exception
1054     */
1055    private static String coord_latest_sync(int offset) throws Exception {
1056        return coord_latestRange_sync(offset, offset);
1057    }
1058
1059    private static String coord_latestRange_sync(int startOffset, int endOffset) throws Exception {
1060        final XLog LOG = XLog.getLog(CoordELFunctions.class);
1061        final Thread currentThread = Thread.currentThread();
1062        ELEvaluator eval = ELEvaluator.getCurrent();
1063        String retVal = "";
1064        int datasetFrequency = (int) getDSFrequency();// in minutes
1065        TimeUnit dsTimeUnit = getDSTimeUnit();
1066        int[] instCount = new int[1];
1067        boolean useCurrentTime = Services.get().getConf().getBoolean(LATEST_EL_USE_CURRENT_TIME, false);
1068        Calendar nominalInstanceCal;
1069        if (useCurrentTime) {
1070            nominalInstanceCal = getCurrentInstance(new Date(), instCount);
1071        }
1072        else {
1073            nominalInstanceCal = getCurrentInstance(getActualTime(), instCount);
1074        }
1075        StringBuilder resolvedInstances = new StringBuilder();
1076        StringBuilder resolvedURIPaths = new StringBuilder();
1077        if (nominalInstanceCal != null) {
1078            Calendar initInstance = getInitialInstanceCal();
1079            SyncCoordDataset ds = (SyncCoordDataset) eval.getVariable(DATASET);
1080            if (ds == null) {
1081                throw new RuntimeException("Associated Dataset should be defined with key " + DATASET);
1082            }
1083            String uriTemplate = ds.getUriTemplate();
1084            Configuration conf = (Configuration) eval.getVariable(CONFIGURATION);
1085            if (conf == null) {
1086                throw new RuntimeException("Associated Configuration should be defined with key " + CONFIGURATION);
1087            }
1088            int available = 0;
1089            boolean resolved = false;
1090            String user = ParamChecker
1091                    .notEmpty((String) eval.getVariable(OozieClient.USER_NAME), OozieClient.USER_NAME);
1092            String doneFlag = ds.getDoneFlag();
1093            URIHandlerService uriService = Services.get().get(URIHandlerService.class);
1094            URIHandler uriHandler = null;
1095            Context uriContext = null;
1096            try {
1097                while (nominalInstanceCal.compareTo(initInstance) >= 0 && !currentThread.isInterrupted()) {
1098                    ELEvaluator uriEval = getUriEvaluator(nominalInstanceCal);
1099                    String uriPath = uriEval.evaluate(uriTemplate, String.class);
1100                    if (uriHandler == null) {
1101                        URI uri = new URI(uriPath);
1102                        uriHandler = uriService.getURIHandler(uri);
1103                        uriContext = uriHandler.getContext(uri, conf, user);
1104                    }
1105                    String uriWithDoneFlag = uriHandler.getURIWithDoneFlag(uriPath, doneFlag);
1106                    if (uriHandler.exists(new URI(uriWithDoneFlag), uriContext)) {
1107                        XLog.getLog(CoordELFunctions.class)
1108                        .debug("Found latest(" + available + "): " + uriWithDoneFlag);
1109                        if (available == startOffset) {
1110                            LOG.debug("Matched latest(" + available + "): " + uriWithDoneFlag);
1111                            resolved = true;
1112                            resolvedInstances.append(DateUtils.formatDateOozieTZ(nominalInstanceCal));
1113                            resolvedURIPaths.append(uriPath);
1114                            retVal = resolvedInstances.toString();
1115                            eval.setVariable("resolved_path", resolvedURIPaths.toString());
1116                            break;
1117                        }
1118                        else if (available <= endOffset) {
1119                            LOG.debug("Matched latest(" + available + "): " + uriWithDoneFlag);
1120                            resolvedInstances.append(DateUtils.formatDateOozieTZ(nominalInstanceCal)).append(
1121                                    INSTANCE_SEPARATOR);
1122                            resolvedURIPaths.append(uriPath).append(INSTANCE_SEPARATOR);
1123                        }
1124
1125                        available--;
1126                    }
1127                    // nominalInstanceCal.add(dsTimeUnit.getCalendarUnit(), -datasetFrequency);
1128                    nominalInstanceCal = (Calendar) initInstance.clone();
1129                    instCount[0]--;
1130                    nominalInstanceCal.add(dsTimeUnit.getCalendarUnit(), instCount[0] * datasetFrequency);
1131                    // DateUtils.moveToEnd(nominalInstanceCal, getDSEndOfFlag());
1132                }
1133            }
1134            finally {
1135                if (uriContext != null) {
1136                    uriContext.destroy();
1137                }
1138            }
1139            if (!resolved) {
1140                // return unchanged latest function with variable 'is_resolved'
1141                // to 'false'
1142                eval.setVariable("is_resolved", Boolean.FALSE);
1143                if (startOffset == endOffset) {
1144                    retVal = "${coord:latest(" + startOffset + ")}";
1145                }
1146                else {
1147                    retVal = "${coord:latestRange(" + startOffset + "," + endOffset + ")}";
1148                }
1149            }
1150            else {
1151                eval.setVariable("is_resolved", Boolean.TRUE);
1152            }
1153        }
1154        else {// No feasible nominal time
1155            eval.setVariable("is_resolved", Boolean.FALSE);
1156        }
1157        return retVal;
1158    }
1159
1160    /**
1161     * @param tm
1162     * @return a new Evaluator to be used for URI-template evaluation
1163     */
1164    private static ELEvaluator getUriEvaluator(Calendar tm) {
1165        tm.setTimeZone(DateUtils.getOozieProcessingTimeZone());
1166        ELEvaluator retEval = new ELEvaluator();
1167        retEval.setVariable("YEAR", tm.get(Calendar.YEAR));
1168        retEval.setVariable("MONTH", (tm.get(Calendar.MONTH) + 1) < 10 ? "0" + (tm.get(Calendar.MONTH) + 1) : (tm
1169                .get(Calendar.MONTH) + 1));
1170        retEval.setVariable("DAY", tm.get(Calendar.DAY_OF_MONTH) < 10 ? "0" + tm.get(Calendar.DAY_OF_MONTH) : tm
1171                .get(Calendar.DAY_OF_MONTH));
1172        retEval.setVariable("HOUR", tm.get(Calendar.HOUR_OF_DAY) < 10 ? "0" + tm.get(Calendar.HOUR_OF_DAY) : tm
1173                .get(Calendar.HOUR_OF_DAY));
1174        retEval.setVariable("MINUTE", tm.get(Calendar.MINUTE) < 10 ? "0" + tm.get(Calendar.MINUTE) : tm
1175                .get(Calendar.MINUTE));
1176        return retEval;
1177    }
1178
1179    /**
1180     * @return whether a data set is SYNCH or ASYNC
1181     */
1182    private static boolean isSyncDataSet() {
1183        ELEvaluator eval = ELEvaluator.getCurrent();
1184        SyncCoordDataset ds = (SyncCoordDataset) eval.getVariable(DATASET);
1185        if (ds == null) {
1186            throw new RuntimeException("Associated Dataset should be defined with key " + DATASET);
1187        }
1188        return ds.getType().equalsIgnoreCase("SYNC");
1189    }
1190
1191    /**
1192     * Check whether a function should be resolved.
1193     *
1194     * @param functionName
1195     * @param n
1196     * @return null if the functionName needs to be resolved otherwise return the calling function unresolved.
1197     */
1198    private static String checkIfResolved(String functionName, String n) {
1199        ELEvaluator eval = ELEvaluator.getCurrent();
1200        String replace = (String) eval.getVariable("resolve_" + functionName);
1201        if (replace == null || (replace != null && replace.equalsIgnoreCase("false"))) { // Don't
1202            // resolve
1203            // return "${coord:" + functionName + "(" + n +")}"; //Unresolved
1204            eval.setVariable(".wrap", "true");
1205            return "coord:" + functionName + "(" + n + ")"; // Unresolved
1206        }
1207        return null; // Resolved it
1208    }
1209
1210    private static String echoUnResolved(String functionName, String n) {
1211        return echoUnResolvedPre(functionName, n, "coord:");
1212    }
1213
1214    private static String echoUnResolvedPre(String functionName, String n, String prefix) {
1215        ELEvaluator eval = ELEvaluator.getCurrent();
1216        eval.setVariable(".wrap", "true");
1217        return prefix + functionName + "(" + n + ")"; // Unresolved
1218    }
1219
1220    /**
1221     * @return the initial instance of a DataSet in DATE
1222     */
1223    private static Date getInitialInstance() {
1224        ELEvaluator eval = ELEvaluator.getCurrent();
1225        return getInitialInstance(eval);
1226    }
1227
1228    /**
1229     * @return the initial instance of a DataSet in DATE
1230     */
1231    private static Date getInitialInstance(ELEvaluator eval) {
1232        return getInitialInstanceCal(eval).getTime();
1233        // return ds.getInitInstance();
1234    }
1235
1236    /**
1237     * @return the initial instance of a DataSet in Calendar
1238     */
1239    private static Calendar getInitialInstanceCal() {
1240        ELEvaluator eval = ELEvaluator.getCurrent();
1241        return getInitialInstanceCal(eval);
1242    }
1243
1244    /**
1245     * @return the initial instance of a DataSet in Calendar
1246     */
1247    private static Calendar getInitialInstanceCal(ELEvaluator eval) {
1248        SyncCoordDataset ds = (SyncCoordDataset) eval.getVariable(DATASET);
1249        if (ds == null) {
1250            throw new RuntimeException("Associated Dataset should be defined with key " + DATASET);
1251        }
1252        Calendar effInitTS = new GregorianCalendar(ds.getTimeZone());
1253        effInitTS.setTime(ds.getInitInstance());
1254        // To adjust EOD/EOM
1255        DateUtils.moveToEnd(effInitTS, getDSEndOfFlag(eval));
1256        return effInitTS;
1257        // return ds.getInitInstance();
1258    }
1259
1260    /**
1261     * @return Nominal or action creation Time when all the dependencies of an application instance are met.
1262     */
1263    private static Date getActionCreationtime() {
1264        ELEvaluator eval = ELEvaluator.getCurrent();
1265        return getActionCreationtime(eval);
1266    }
1267
1268    /**
1269     * @return Nominal or action creation Time when all the dependencies of an application instance are met.
1270     */
1271    private static Date getActionCreationtime(ELEvaluator eval) {
1272        SyncCoordAction coordAction = (SyncCoordAction) eval.getVariable(COORD_ACTION);
1273        if (coordAction == null) {
1274            throw new RuntimeException("Associated Application instance should be defined with key " + COORD_ACTION);
1275        }
1276        return coordAction.getNominalTime();
1277    }
1278
1279    /**
1280     * @return Actual Time when all the dependencies of an application instance are met.
1281     */
1282    private static Date getActualTime() {
1283        ELEvaluator eval = ELEvaluator.getCurrent();
1284        SyncCoordAction coordAction = (SyncCoordAction) eval.getVariable(COORD_ACTION);
1285        if (coordAction == null) {
1286            throw new RuntimeException("Associated Application instance should be defined with key " + COORD_ACTION);
1287        }
1288        return coordAction.getActualTime();
1289    }
1290
1291    /**
1292     * @return TimeZone for the application or job.
1293     */
1294    private static TimeZone getJobTZ() {
1295        ELEvaluator eval = ELEvaluator.getCurrent();
1296        SyncCoordAction coordAction = (SyncCoordAction) eval.getVariable(COORD_ACTION);
1297        if (coordAction == null) {
1298            throw new RuntimeException("Associated Application instance should be defined with key " + COORD_ACTION);
1299        }
1300        return coordAction.getTimeZone();
1301    }
1302
1303    /**
1304     * Find the current instance based on effectiveTime (i.e Action_Creation_Time or Action_Start_Time)
1305     *
1306     * @return current instance i.e. current(0) returns null if effectiveTime is earlier than Initial Instance time of
1307     *         the dataset.
1308     */
1309    public static Calendar getCurrentInstance(Date effectiveTime, int instanceCount[]) {
1310        ELEvaluator eval = ELEvaluator.getCurrent();
1311        return getCurrentInstance(effectiveTime, instanceCount, eval);
1312    }
1313
1314    /**
1315     * Find the current instance based on effectiveTime (i.e Action_Creation_Time or Action_Start_Time)
1316     *
1317     * @return current instance i.e. current(0) returns null if effectiveTime is earlier than Initial Instance time of
1318     *         the dataset.
1319     */
1320    private static Calendar getCurrentInstance(Date effectiveTime, int instanceCount[], ELEvaluator eval) {
1321        Date datasetInitialInstance = getInitialInstance(eval);
1322        TimeUnit dsTimeUnit = getDSTimeUnit(eval);
1323        TimeZone dsTZ = getDatasetTZ(eval);
1324        int dsFreq = getDSFrequency(eval);
1325        // Convert Date to Calendar for corresponding TZ
1326        Calendar current = Calendar.getInstance(dsTZ);
1327        current.setTime(datasetInitialInstance);
1328
1329        Calendar calEffectiveTime = new GregorianCalendar(dsTZ);
1330        calEffectiveTime.setTime(effectiveTime);
1331        if (instanceCount == null) {    // caller doesn't care about this value
1332            instanceCount = new int[1];
1333        }
1334        instanceCount[0] = 0;
1335        if (current.compareTo(calEffectiveTime) > 0) {
1336            return null;
1337        }
1338
1339        switch(dsTimeUnit) {
1340            case MINUTE:
1341                instanceCount[0] = (int) ((effectiveTime.getTime() - datasetInitialInstance.getTime()) / MINUTE_MSEC);
1342                break;
1343            case HOUR:
1344                instanceCount[0] = (int) ((effectiveTime.getTime() - datasetInitialInstance.getTime()) / HOUR_MSEC);
1345                break;
1346            case DAY:
1347            case END_OF_DAY:
1348                instanceCount[0] = (int) ((effectiveTime.getTime() - datasetInitialInstance.getTime()) / DAY_MSEC);
1349                break;
1350            case MONTH:
1351            case END_OF_MONTH:
1352                instanceCount[0] = (int) ((effectiveTime.getTime() - datasetInitialInstance.getTime()) / MONTH_MSEC);
1353                break;
1354            case YEAR:
1355                instanceCount[0] = (int) ((effectiveTime.getTime() - datasetInitialInstance.getTime()) / YEAR_MSEC);
1356                break;
1357            default:
1358                throw new IllegalArgumentException("Unhandled dataset time unit " + dsTimeUnit);
1359        }
1360
1361        if (instanceCount[0] > 2) {
1362            instanceCount[0] = (instanceCount[0] / dsFreq);
1363            current.add(dsTimeUnit.getCalendarUnit(), instanceCount[0] * dsFreq);
1364        } else {
1365            instanceCount[0] = 0;
1366        }
1367        while (!current.getTime().after(effectiveTime)) {
1368            current.add(dsTimeUnit.getCalendarUnit(), dsFreq);
1369            instanceCount[0]++;
1370        }
1371        current.add(dsTimeUnit.getCalendarUnit(), -dsFreq);
1372        instanceCount[0]--;
1373        return current;
1374    }
1375
1376    /**
1377     * Find the current instance based on effectiveTime (i.e Action_Creation_Time or Action_Start_Time)
1378     *
1379     * @return current instance i.e. current(0) returns null if effectiveTime is earlier than Initial Instance time of
1380     *         the dataset.
1381     */
1382    private static Calendar getCurrentInstance_old(Date effectiveTime, int instanceCount[], ELEvaluator eval) {
1383        Date datasetInitialInstance = getInitialInstance(eval);
1384        TimeUnit dsTimeUnit = getDSTimeUnit(eval);
1385        TimeZone dsTZ = getDatasetTZ(eval);
1386        int dsFreq = getDSFrequency(eval);
1387        // Convert Date to Calendar for corresponding TZ
1388        Calendar current = Calendar.getInstance();
1389        current.setTime(datasetInitialInstance);
1390        current.setTimeZone(dsTZ);
1391
1392        Calendar calEffectiveTime = Calendar.getInstance();
1393        calEffectiveTime.setTime(effectiveTime);
1394        calEffectiveTime.setTimeZone(dsTZ);
1395        if (instanceCount == null) {    // caller doesn't care about this value
1396            instanceCount = new int[1];
1397        }
1398        instanceCount[0] = 0;
1399        if (current.compareTo(calEffectiveTime) > 0) {
1400            return null;
1401        }
1402        Calendar origCurrent = (Calendar) current.clone();
1403        while (current.compareTo(calEffectiveTime) <= 0) {
1404            current = (Calendar) origCurrent.clone();
1405            instanceCount[0]++;
1406            current.add(dsTimeUnit.getCalendarUnit(), instanceCount[0] * dsFreq);
1407        }
1408        instanceCount[0]--;
1409
1410        current = (Calendar) origCurrent.clone();
1411        current.add(dsTimeUnit.getCalendarUnit(), instanceCount[0] * dsFreq);
1412        return current;
1413    }
1414
1415    public static Calendar getEffectiveNominalTime() {
1416        Date datasetInitialInstance = getInitialInstance();
1417        TimeZone dsTZ = getDatasetTZ();
1418        // Convert Date to Calendar for corresponding TZ
1419        Calendar current = Calendar.getInstance();
1420        current.setTime(datasetInitialInstance);
1421        current.setTimeZone(dsTZ);
1422
1423        Calendar calEffectiveTime = Calendar.getInstance();
1424        calEffectiveTime.setTime(getActionCreationtime());
1425        calEffectiveTime.setTimeZone(dsTZ);
1426        if (current.compareTo(calEffectiveTime) > 0) {
1427            // Nominal Time < initial Instance
1428            // TODO: getClass() call doesn't work from static method.
1429            // XLog.getLog("CoordELFunction.class").warn("ACTION CREATED BEFORE INITIAL INSTACE "+
1430            // current.getTime());
1431            return null;
1432        }
1433        return calEffectiveTime;
1434    }
1435
1436    /**
1437     * @return dataset frequency in minutes
1438     */
1439    private static int getDSFrequency() {
1440        ELEvaluator eval = ELEvaluator.getCurrent();
1441        return getDSFrequency(eval);
1442    }
1443
1444    /**
1445     * @return dataset frequency in minutes
1446     */
1447    private static int getDSFrequency(ELEvaluator eval) {
1448        SyncCoordDataset ds = (SyncCoordDataset) eval.getVariable(DATASET);
1449        if (ds == null) {
1450            throw new RuntimeException("Associated Dataset should be defined with key " + DATASET);
1451        }
1452        return ds.getFrequency();
1453    }
1454
1455    /**
1456     * @return dataset TimeUnit
1457     */
1458    private static TimeUnit getDSTimeUnit() {
1459        ELEvaluator eval = ELEvaluator.getCurrent();
1460        return getDSTimeUnit(eval);
1461    }
1462
1463    /**
1464     * @return dataset TimeUnit
1465     */
1466    public static TimeUnit getDSTimeUnit(ELEvaluator eval) {
1467        SyncCoordDataset ds = (SyncCoordDataset) eval.getVariable(DATASET);
1468        if (ds == null) {
1469            throw new RuntimeException("Associated Dataset should be defined with key " + DATASET);
1470        }
1471        return ds.getTimeUnit();
1472    }
1473
1474    /**
1475     * @return dataset TimeZone
1476     */
1477    public static TimeZone getDatasetTZ() {
1478        ELEvaluator eval = ELEvaluator.getCurrent();
1479        return getDatasetTZ(eval);
1480    }
1481
1482    /**
1483     * @return dataset TimeZone
1484     */
1485    private static TimeZone getDatasetTZ(ELEvaluator eval) {
1486        SyncCoordDataset ds = (SyncCoordDataset) eval.getVariable(DATASET);
1487        if (ds == null) {
1488            throw new RuntimeException("Associated Dataset should be defined with key " + DATASET);
1489        }
1490        return ds.getTimeZone();
1491    }
1492
1493    /**
1494     * @return dataset TimeUnit
1495     */
1496    private static TimeUnit getDSEndOfFlag() {
1497        ELEvaluator eval = ELEvaluator.getCurrent();
1498        return getDSEndOfFlag(eval);
1499    }
1500
1501    /**
1502     * @return dataset TimeUnit
1503     */
1504    private static TimeUnit getDSEndOfFlag(ELEvaluator eval) {
1505        SyncCoordDataset ds = (SyncCoordDataset) eval.getVariable(DATASET);
1506        if (ds == null) {
1507            throw new RuntimeException("Associated Dataset should be defined with key " + DATASET);
1508        }
1509        return ds.getEndOfDuration();// == null ? "": ds.getEndOfDuration();
1510    }
1511
1512    /**
1513     * Return a job configuration property for the coordinator.
1514     *
1515     * @param property property name.
1516     * @return the value of the property, <code>null</code> if the property is undefined.
1517     */
1518    public static String coord_conf(String property) {
1519        ELEvaluator eval = ELEvaluator.getCurrent();
1520        return (String) eval.getVariable(property);
1521    }
1522
1523    /**
1524     * Return the user that submitted the coordinator job.
1525     *
1526     * @return the user that submitted the coordinator job.
1527     */
1528    public static String coord_user() {
1529        ELEvaluator eval = ELEvaluator.getCurrent();
1530        return (String) eval.getVariable(OozieClient.USER_NAME);
1531    }
1532
1533    /**
1534     * Takes two offset times and returns a list of multiples of the frequency offset from the effective nominal time that occur
1535     * between them.  The caller should make sure that startCal is earlier than endCal.
1536     * <p>
1537     * As a simple example, assume its the same day: startCal is 1:00, endCal is 2:00, frequency is 20min, and effective nominal
1538     * time is 1:20 -- then this method would return a list containing: -20, 0, 20, 40, 60
1539     *
1540     * @param startCal The earlier offset time
1541     * @param endCal The later offset time
1542     * @param eval The ELEvaluator to use; cannot be null
1543     * @return A list of multiple of the frequency offset from the effective nominal time that occur between the startCal and endCal
1544     */
1545    public static List<Integer> expandOffsetTimes(Calendar startCal, Calendar endCal, ELEvaluator eval) {
1546        List<Integer> expandedFreqs = new ArrayList<Integer>();
1547        // Use eval because the "current" eval isn't set
1548        int freq = getDSFrequency(eval);
1549        TimeUnit freqUnit = getDSTimeUnit(eval);
1550        Calendar cal = getCurrentInstance(getActionCreationtime(eval), null, eval);
1551        int totalFreq = 0;
1552        if (startCal.before(cal)) {
1553            while (cal.after(startCal)) {
1554                cal.add(freqUnit.getCalendarUnit(), -freq);
1555                totalFreq += -freq;
1556            }
1557            if (cal.before(startCal)) {
1558                cal.add(freqUnit.getCalendarUnit(), freq);
1559                totalFreq += freq;
1560            }
1561        }
1562        else if (startCal.after(cal)) {
1563            while (cal.before(startCal)) {
1564                cal.add(freqUnit.getCalendarUnit(), freq);
1565                totalFreq += freq;
1566            }
1567        }
1568        // At this point, cal is the smallest multiple of the dataset frequency that is >= to the startCal and offset from the
1569        // effective nominal time.  Now we can find all of the instances that occur between startCal and endCal, inclusive.
1570        while (cal.before(endCal) || cal.equals(endCal)) {
1571            expandedFreqs.add(totalFreq);
1572            cal.add(freqUnit.getCalendarUnit(), freq);
1573            totalFreq += freq;
1574        }
1575        return expandedFreqs;
1576    }
1577
1578    /**
1579     * Resolve the offset time from the effective nominal time
1580     *
1581     * @param n offset amount (integer)
1582     * @param timeUnit TimeUnit for offset n ("MINUTE", "HOUR", "DAY", "MONTH", "YEAR")
1583     * @param eval The ELEvaluator to use; or null to use the "current" eval
1584     * @return A Calendar of the offset time
1585     */
1586    public static Calendar resolveOffsetRawTime(int n, TimeUnit timeUnit, ELEvaluator eval) {
1587        // Use eval if given (for when the "current" eval isn't set)
1588        Calendar cal;
1589        if (eval == null) {
1590            cal = getCurrentInstance(getActionCreationtime(), null);
1591        }
1592        else {
1593            cal = getCurrentInstance(getActionCreationtime(eval), null, eval);
1594        }
1595        if (cal == null) {
1596            XLog.getLog(CoordELFunctions.class).warn("If the initial instance of the dataset is later than the nominal time, an"
1597                    + " empty string is returned. This means that no data is available at the offset instance specified by the user"
1598                    + " and the user could try modifying his or her initial-instance to an earlier time.");
1599            return null;
1600        }
1601        cal.add(timeUnit.getCalendarUnit(), n);
1602        return cal;
1603    }
1604}