Repository: oozie Updated Branches: refs/heads/master 04bd49699 -> 08fa157df
OOZIE-2397 LAST_ONLY and NONE don't properly handle READY actions (rkanter) Project: http://git-wip-us.apache.org/repos/asf/oozie/repo Commit: http://git-wip-us.apache.org/repos/asf/oozie/commit/08fa157d Tree: http://git-wip-us.apache.org/repos/asf/oozie/tree/08fa157d Diff: http://git-wip-us.apache.org/repos/asf/oozie/diff/08fa157d Branch: refs/heads/master Commit: 08fa157df4eac151be32d9aa5cec5cd4c88fe574 Parents: 04bd496 Author: Robert Kanter <[email protected]> Authored: Wed Nov 25 11:22:03 2015 -0800 Committer: Robert Kanter <[email protected]> Committed: Wed Nov 25 11:22:03 2015 -0800 ---------------------------------------------------------------------- .../org/apache/oozie/CoordinatorActionBean.java | 6 +- .../org/apache/oozie/CoordinatorJobBean.java | 4 +- .../coord/CoordActionInputCheckXCommand.java | 57 ++------- .../command/coord/CoordActionReadyXCommand.java | 67 ++++++++++- .../oozie/command/coord/CoordCommandUtils.java | 48 ++++++++ .../CoordActionGetForInputCheckJPAExecutor.java | 23 ++-- .../jpa/CoordJobGetReadyActionsJPAExecutor.java | 24 ++-- .../executor/jpa/CoordJobQueryExecutor.java | 10 +- .../TestCoordActionInputCheckXCommand.java | 119 +++++-------------- .../coord/TestCoordActionReadyXCommand.java | 100 ++++++++++++++++ .../command/coord/TestCoordCommandUtils.java | 55 +++++++++ .../TestCoordMaterializeTransitionXCommand.java | 23 ---- ...tCoordActionGetForInputCheckJPAExecutor.java | 8 +- .../TestCoordJobGetReadyActionsJPAExecutor.java | 81 +++++-------- .../executor/jpa/TestCoordJobQueryExecutor.java | 10 +- .../org/apache/oozie/test/XDataTestCase.java | 39 +++++- .../site/twiki/CoordinatorFunctionalSpec.twiki | 36 +++--- release-log.txt | 1 + 18 files changed, 452 insertions(+), 259 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/oozie/blob/08fa157d/core/src/main/java/org/apache/oozie/CoordinatorActionBean.java ---------------------------------------------------------------------- diff --git a/core/src/main/java/org/apache/oozie/CoordinatorActionBean.java b/core/src/main/java/org/apache/oozie/CoordinatorActionBean.java index c3d4bb4..91bff4d 100644 --- a/core/src/main/java/org/apache/oozie/CoordinatorActionBean.java +++ b/core/src/main/java/org/apache/oozie/CoordinatorActionBean.java @@ -89,7 +89,7 @@ import org.json.simple.JSONObject; // Select Query used by Timeout and skip commands @NamedQuery(name = "GET_COORD_ACTION_FOR_TIMEOUT", query = "select a.id, a.jobId, a.statusStr, a.runConf, a.pending, a.nominalTimestamp, a.createdTimestamp from CoordinatorActionBean a where a.id = :id"), // Select query used by InputCheck command - @NamedQuery(name = "GET_COORD_ACTION_FOR_INPUTCHECK", query = "select a.id, a.jobId, a.statusStr, a.runConf, a.nominalTimestamp, a.createdTimestamp, a.actionXml, a.missingDependencies, a.pushMissingDependencies, a.timeOut, a.externalId from CoordinatorActionBean a where a.id = :id"), + @NamedQuery(name = "GET_COORD_ACTION_FOR_INPUTCHECK", query = "select a.id, a.actionNumber, a.jobId, a.statusStr, a.runConf, a.nominalTimestamp, a.createdTimestamp, a.actionXml, a.missingDependencies, a.pushMissingDependencies, a.timeOut, a.externalId from CoordinatorActionBean a where a.id = :id"), // Select query used by CoordActionUpdate command @NamedQuery(name = "GET_COORD_ACTION_FOR_EXTERNALID", query = "select a.id, a.jobId, a.statusStr, a.pending, a.externalId, a.lastModifiedTimestamp, a.slaXml, a.nominalTimestamp, a.createdTimestamp from CoordinatorActionBean a where a.externalId = :externalId"), // Select query used by Check command @@ -97,9 +97,9 @@ import org.json.simple.JSONObject; // Select query used by Start command @NamedQuery(name = "GET_COORD_ACTION_FOR_START", query = "select a.id, a.jobId, a.statusStr, a.pending, a.createdConf, a.slaXml, a.actionXml, a.externalId, a.errorMessage, a.errorCode, a.nominalTimestamp, a.createdTimestamp from CoordinatorActionBean a where a.id = :id"), - @NamedQuery(name = "GET_COORD_ACTIONS_FOR_JOB_FIFO", query = "select a.id, a.jobId, a.statusStr, a.pending, a.nominalTimestamp, a.createdTimestamp from CoordinatorActionBean a where a.jobId = :jobId AND a.statusStr = 'READY' order by a.nominalTimestamp"), + @NamedQuery(name = "GET_COORD_ACTIONS_FOR_JOB_FIFO", query = "select a.id, a.actionNumber, a.jobId, a.statusStr, a.pending, a.nominalTimestamp, a.createdTimestamp from CoordinatorActionBean a where a.jobId = :jobId AND a.statusStr = 'READY' order by a.nominalTimestamp"), - @NamedQuery(name = "GET_COORD_ACTIONS_FOR_JOB_LIFO", query = "select a.id, a.jobId, a.statusStr, a.pending, a.nominalTimestamp, a.createdTimestamp from CoordinatorActionBean a where a.jobId = :jobId AND a.statusStr = 'READY' order by a.nominalTimestamp desc"), + @NamedQuery(name = "GET_COORD_ACTIONS_FOR_JOB_LIFO", query = "select a.id, a.actionNumber, a.jobId, a.statusStr, a.pending, a.nominalTimestamp, a.createdTimestamp from CoordinatorActionBean a where a.jobId = :jobId AND a.statusStr = 'READY' order by a.nominalTimestamp desc"), @NamedQuery(name = "GET_COORD_RUNNING_ACTIONS_COUNT", query = "select count(a) from CoordinatorActionBean a where a.jobId = :jobId AND (a.statusStr = 'RUNNING' OR a.statusStr='SUBMITTED')"), http://git-wip-us.apache.org/repos/asf/oozie/blob/08fa157d/core/src/main/java/org/apache/oozie/CoordinatorJobBean.java ---------------------------------------------------------------------- diff --git a/core/src/main/java/org/apache/oozie/CoordinatorJobBean.java b/core/src/main/java/org/apache/oozie/CoordinatorJobBean.java index 3ec2eaf..3a3120d 100644 --- a/core/src/main/java/org/apache/oozie/CoordinatorJobBean.java +++ b/core/src/main/java/org/apache/oozie/CoordinatorJobBean.java @@ -92,9 +92,9 @@ import org.json.simple.JSONObject; @NamedQuery(name = "GET_COORD_JOB_USER_APPNAME", query = "select w.user, w.appName from CoordinatorJobBean w where w.id = :id"), - @NamedQuery(name = "GET_COORD_JOB_INPUT_CHECK", query = "select w.user, w.appName, w.statusStr, w.appNamespace, w.execution, w.frequency, w.timeUnitStr, w.timeZone, w.endTimestamp from CoordinatorJobBean w where w.id = :id"), + @NamedQuery(name = "GET_COORD_JOB_INPUT_CHECK", query = "select w.user, w.appName, w.statusStr, w.appNamespace, w.execution, w.frequency, w.timeUnitStr, w.timeZone, w.startTimestamp, w.endTimestamp, w.jobXml from CoordinatorJobBean w where w.id = :id"), - @NamedQuery(name = "GET_COORD_JOB_ACTION_READY", query = "select w.id, w.user, w.group, w.appName, w.statusStr, w.execution, w.concurrency from CoordinatorJobBean w where w.id = :id"), + @NamedQuery(name = "GET_COORD_JOB_ACTION_READY", query = "select w.id, w.user, w.group, w.appName, w.statusStr, w.execution, w.concurrency, w.frequency, w.timeUnitStr, w.timeZone, w.startTimestamp, w.endTimestamp, w.jobXml from CoordinatorJobBean w where w.id = :id"), @NamedQuery(name = "GET_COORD_JOB_ACTION_KILL", query = "select w.id, w.user, w.group, w.appName, w.statusStr from CoordinatorJobBean w where w.id = :id"), http://git-wip-us.apache.org/repos/asf/oozie/blob/08fa157d/core/src/main/java/org/apache/oozie/command/coord/CoordActionInputCheckXCommand.java ---------------------------------------------------------------------- diff --git a/core/src/main/java/org/apache/oozie/command/coord/CoordActionInputCheckXCommand.java b/core/src/main/java/org/apache/oozie/command/coord/CoordActionInputCheckXCommand.java index 179dfbf..742d00d 100644 --- a/core/src/main/java/org/apache/oozie/command/coord/CoordActionInputCheckXCommand.java +++ b/core/src/main/java/org/apache/oozie/command/coord/CoordActionInputCheckXCommand.java @@ -95,40 +95,6 @@ public class CoordActionInputCheckXCommand extends CoordinatorXCommand<Void> { LogUtils.setLogInfo(actionId); } - /** - * Computes the nominal time of the next action. - * Based on CoordMaterializeTransitionXCommand#materializeActions - * - * @return the nominal time of the next action - * @throws ParseException - */ - private Date computeNextNominalTime() throws ParseException { - Date nextNominalTime; - boolean isCronFrequency = false; - int freq = -1; - try { - freq = Integer.parseInt(coordJob.getFrequency()); - } catch (NumberFormatException e) { - isCronFrequency = true; - } - - if (isCronFrequency) { - nextNominalTime = CoordCommandUtils.getNextValidActionTimeForCronFrequency(coordAction.getNominalTime(), coordJob); - } else { - Calendar nextNominalTimeCal = Calendar.getInstance(DateUtils.getTimeZone(coordJob.getTimeZone())); - nextNominalTimeCal.setTime(coordAction.getNominalTime()); - TimeUnit freqTU = TimeUnit.valueOf(coordJob.getTimeUnitStr()); - nextNominalTimeCal.add(freqTU.getCalendarUnit(), freq); - nextNominalTime = nextNominalTimeCal.getTime(); - } - - // If the next nominal time is after the job's end time, then this is the last action, so return null - if (nextNominalTime.after(coordJob.getEndTime())) { - nextNominalTime = null; - } - return nextNominalTime; - } - @Override protected Void execute() throws CommandException { LOG.debug("[" + actionId + "]::ActionInputCheck:: Action is in WAITING state."); @@ -154,7 +120,7 @@ public class CoordActionInputCheckXCommand extends CoordinatorXCommand<Void> { Configuration actionConf = new XConfiguration(new StringReader(coordAction.getRunConf())); Date now = new Date(); if (coordJob.getExecutionOrder().equals(CoordinatorJobBean.Execution.LAST_ONLY)) { - Date nextNominalTime = computeNextNominalTime(); + Date nextNominalTime = CoordCommandUtils.computeNextNominalTime(coordJob, coordAction); if (nextNominalTime != null) { // If the current time is after the next action's nominal time, then we've passed the window where this action // should be started; so set it to SKIPPED @@ -173,22 +139,21 @@ public class CoordActionInputCheckXCommand extends CoordinatorXCommand<Void> { } else if (coordJob.getExecutionOrder().equals(CoordinatorJobBean.Execution.NONE)) { // If the current time is after the nominal time of this action plus some tolerance, - // then we've passed the window where this action - // should be started; so set it to SKIPPED + // then we've passed the window where this action should be started; so set it to SKIPPED Calendar cal = Calendar.getInstance(DateUtils.getTimeZone(coordJob.getTimeZone())); cal.setTime(nominalTime); - cal.add(Calendar.MINUTE, ConfigurationService.getInt(COORD_EXECUTION_NONE_TOLERANCE)); - nominalTime = cal.getTime(); - if (now.after(nominalTime)) { - LOG.info("NONE execution: Preparing to skip action [{0}] because the current time [{1}] is later than " - + "the nominal time [{2}] of the current action]", coordAction.getId(), - DateUtils.formatDateOozieTZ(now), DateUtils.formatDateOozieTZ(nominalTime)); + int tolerance = ConfigurationService.getInt(COORD_EXECUTION_NONE_TOLERANCE); + cal.add(Calendar.MINUTE, tolerance); + if (now.after(cal.getTime())) { + LOG.info("NONE execution: Preparing to skip action [{0}] because the current time [{1}] is more than [{2}]" + + " minutes later than the nominal time [{3}] of the current action]", coordAction.getId(), + DateUtils.formatDateOozieTZ(now), tolerance, DateUtils.formatDateOozieTZ(nominalTime)); queue(new CoordActionSkipXCommand(coordAction, coordJob.getUser(), coordJob.getAppName())); return null; } else { - LOG.debug("NONE execution: Not skipping action [{0}] because the current time [{1}] is earlier than " - + "the nominal time [{2}] of the current action]", coordAction.getId(), - DateUtils.formatDateOozieTZ(now), DateUtils.formatDateOozieTZ(coordAction.getNominalTime())); + LOG.debug("NONE execution: Not skipping action [{0}] because the current time [{1}] is earlier than [{2}]" + + " minutes later than the nominal time [{3}] of the current action]", coordAction.getId(), + DateUtils.formatDateOozieTZ(now), tolerance, DateUtils.formatDateOozieTZ(coordAction.getNominalTime())); } } http://git-wip-us.apache.org/repos/asf/oozie/blob/08fa157d/core/src/main/java/org/apache/oozie/command/coord/CoordActionReadyXCommand.java ---------------------------------------------------------------------- diff --git a/core/src/main/java/org/apache/oozie/command/coord/CoordActionReadyXCommand.java b/core/src/main/java/org/apache/oozie/command/coord/CoordActionReadyXCommand.java index ad9cb12..2e9c5ea 100644 --- a/core/src/main/java/org/apache/oozie/command/coord/CoordActionReadyXCommand.java +++ b/core/src/main/java/org/apache/oozie/command/coord/CoordActionReadyXCommand.java @@ -18,12 +18,17 @@ package org.apache.oozie.command.coord; +import java.text.ParseException; +import java.util.Calendar; +import java.util.Date; +import java.util.Iterator; import java.util.List; import org.apache.oozie.CoordinatorActionBean; import org.apache.oozie.CoordinatorJobBean; import org.apache.oozie.ErrorCode; import org.apache.oozie.client.CoordinatorAction; +import org.apache.oozie.client.CoordinatorJob; import org.apache.oozie.client.Job; import org.apache.oozie.command.CommandException; import org.apache.oozie.command.PreconditionException; @@ -34,10 +39,13 @@ import org.apache.oozie.executor.jpa.CoordJobGetRunningActionsCountJPAExecutor; import org.apache.oozie.executor.jpa.CoordJobQueryExecutor; import org.apache.oozie.executor.jpa.CoordJobQueryExecutor.CoordJobQuery; import org.apache.oozie.executor.jpa.JPAExecutorException; +import org.apache.oozie.service.ConfigurationService; import org.apache.oozie.service.JPAService; import org.apache.oozie.service.Services; +import org.apache.oozie.util.DateUtils; import org.apache.oozie.util.LogUtils; import org.apache.oozie.util.XLog; +import org.jdom.JDOMException; public class CoordActionReadyXCommand extends CoordinatorXCommand<Void> { private final String jobId; @@ -67,7 +75,7 @@ public class CoordActionReadyXCommand extends CoordinatorXCommand<Void> { int numActionsToStart = -1; // get execution setting for this job (FIFO, LIFO, LAST_ONLY) - String jobExecution = coordJob.getExecution(); + CoordinatorJob.Execution jobExecution = coordJob.getExecutionOrder(); // get concurrency setting for this job int jobConcurrency = coordJob.getConcurrency(); // if less than 0, then UNLIMITED concurrency @@ -93,21 +101,70 @@ public class CoordActionReadyXCommand extends CoordinatorXCommand<Void> { // no actions to start if (numActionsToStart == 0) { log.warn("No actions to start for jobId=" + jobId + " as max concurrency reached!"); - return null; } } // get list of actions that are READY and fit in the concurrency and execution List<CoordinatorActionBean> actions; try { - actions = jpaService.execute(new CoordJobGetReadyActionsJPAExecutor(jobId, numActionsToStart, jobExecution)); + actions = jpaService.execute(new CoordJobGetReadyActionsJPAExecutor(jobId, jobExecution.name())); } catch (JPAExecutorException je) { throw new CommandException(je); } log.debug("Number of READY actions = " + actions.size()); - // make sure auth token is not null - // log.denug("user=" + user + ", token=" + authToken); + Date now = new Date(); + // If we're using LAST_ONLY or NONE, we should check if any of these need to be SKIPPED instead of SUBMITTED + if (jobExecution.equals(CoordinatorJobBean.Execution.LAST_ONLY)) { + for (Iterator<CoordinatorActionBean> it = actions.iterator(); it.hasNext(); ) { + CoordinatorActionBean action = it.next(); + try { + Date nextNominalTime = CoordCommandUtils.computeNextNominalTime(coordJob, action); + if (nextNominalTime != null) { + // If the current time is after the next action's nominal time, then we've passed the window where this + // action should be started; so set it to SKIPPED + if (now.after(nextNominalTime)) { + LOG.info("LAST_ONLY execution: Preparing to skip action [{0}] because the current time [{1}] is later " + + "than the nominal time [{2}] of the next action]", action.getId(), + DateUtils.formatDateOozieTZ(now), DateUtils.formatDateOozieTZ(nextNominalTime)); + queue(new CoordActionSkipXCommand(action, coordJob.getUser(), coordJob.getAppName())); + it.remove(); + } else { + LOG.debug("LAST_ONLY execution: Not skipping action [{0}] because the current time [{1}] is earlier " + + "than the nominal time [{2}] of the next action]", action.getId(), + DateUtils.formatDateOozieTZ(now), DateUtils.formatDateOozieTZ(nextNominalTime)); + } + } + } catch (ParseException e) { + LOG.error("Should not happen", e); + } catch (JDOMException e) { + LOG.error("Should not happen", e); + } + } + } + else if (jobExecution.equals(CoordinatorJobBean.Execution.NONE)) { + for (Iterator<CoordinatorActionBean> it = actions.iterator(); it.hasNext(); ) { + CoordinatorActionBean action = it.next(); + // If the current time is after the nominal time of this action plus some tolerance, + // then we've passed the window where this action should be started; so set it to SKIPPED + Calendar cal = Calendar.getInstance(DateUtils.getTimeZone(coordJob.getTimeZone())); + cal.setTime(action.getNominalTime()); + int tolerance = ConfigurationService.getInt(CoordActionInputCheckXCommand.COORD_EXECUTION_NONE_TOLERANCE); + cal.add(Calendar.MINUTE, tolerance); + if (now.after(cal.getTime())) { + LOG.info("NONE execution: Preparing to skip action [{0}] because the current time [{1}] is more than [{2}]" + + " minutes later than the nominal time [{3}] of the current action]", action.getId(), + DateUtils.formatDateOozieTZ(now), tolerance, DateUtils.formatDateOozieTZ(action.getNominalTime())); + queue(new CoordActionSkipXCommand(action, coordJob.getUser(), coordJob.getAppName())); + it.remove(); + } else { + LOG.debug("NONE execution: Not skipping action [{0}] because the current time [{1}] is earlier than [{2}]" + + " minutes later than the nominal time [{3}] of the current action]", action.getId(), + DateUtils.formatDateOozieTZ(now), tolerance, DateUtils.formatDateOozieTZ(action.getNominalTime())); + } + } + } + int counter = 0; for (CoordinatorActionBean action : actions) { // continue if numActionsToStart is negative (no limit on number of http://git-wip-us.apache.org/repos/asf/oozie/blob/08fa157d/core/src/main/java/org/apache/oozie/command/coord/CoordCommandUtils.java ---------------------------------------------------------------------- diff --git a/core/src/main/java/org/apache/oozie/command/coord/CoordCommandUtils.java b/core/src/main/java/org/apache/oozie/command/coord/CoordCommandUtils.java index 131368a..58ef483 100644 --- a/core/src/main/java/org/apache/oozie/command/coord/CoordCommandUtils.java +++ b/core/src/main/java/org/apache/oozie/command/coord/CoordCommandUtils.java @@ -51,6 +51,7 @@ import org.apache.oozie.util.ELEvaluator; import org.apache.oozie.util.XConfiguration; import org.apache.oozie.util.XmlUtils; import org.jdom.Element; +import org.jdom.JDOMException; import org.quartz.CronExpression; import org.apache.commons.lang.StringUtils; import org.apache.oozie.CoordinatorJobBean; @@ -749,4 +750,51 @@ public class CoordCommandUtils { return nextTime; } + /** + * Computes the nominal time of the next action. + * Based on CoordMaterializeTransitionXCommand#materializeActions + * + * The Coordinator Job needs to have the frequency, time unit, time zone, start time, end time, and job xml. + * The Coordinator Action needs to have the nominal time and action number. + * + * @param coordJob The Coordinator Job + * @param coordAction The Coordinator Action + * @return the nominal time of the next action + * @throws ParseException + * @throws JDOMException + */ + public static Date computeNextNominalTime(CoordinatorJobBean coordJob, CoordinatorActionBean coordAction) + throws ParseException, JDOMException { + Date nextNominalTime; + boolean isCronFrequency = false; + int freq = -1; + try { + freq = Integer.parseInt(coordJob.getFrequency()); + } catch (NumberFormatException e) { + isCronFrequency = true; + } + + if (isCronFrequency) { + nextNominalTime = CoordCommandUtils.getNextValidActionTimeForCronFrequency(coordAction.getNominalTime(), coordJob); + } else { + TimeZone appTz = DateUtils.getTimeZone(coordJob.getTimeZone()); + Calendar nextNominalTimeCal = Calendar.getInstance(appTz); + nextNominalTimeCal.setTime(coordJob.getStartTimestamp()); + TimeUnit freqTU = TimeUnit.valueOf(coordJob.getTimeUnitStr()); + // Action Number is indexed by 1, so no need to +1 here + nextNominalTimeCal.add(freqTU.getCalendarUnit(), coordAction.getActionNumber() * freq); + String jobXml = coordJob.getJobXml(); + Element eJob = XmlUtils.parseXml(jobXml); + TimeUnit endOfFlag = TimeUnit.valueOf(eJob.getAttributeValue("end_of_duration")); + // Move to the End of duration, if needed. + DateUtils.moveToEnd(nextNominalTimeCal, endOfFlag); + nextNominalTime = nextNominalTimeCal.getTime(); + } + + // If the next nominal time is after the job's end time, then this is the last action, so return null + if (nextNominalTime.after(coordJob.getEndTime())) { + nextNominalTime = null; + } + return nextNominalTime; + } } http://git-wip-us.apache.org/repos/asf/oozie/blob/08fa157d/core/src/main/java/org/apache/oozie/executor/jpa/CoordActionGetForInputCheckJPAExecutor.java ---------------------------------------------------------------------- diff --git a/core/src/main/java/org/apache/oozie/executor/jpa/CoordActionGetForInputCheckJPAExecutor.java b/core/src/main/java/org/apache/oozie/executor/jpa/CoordActionGetForInputCheckJPAExecutor.java index e58646b..2023678 100644 --- a/core/src/main/java/org/apache/oozie/executor/jpa/CoordActionGetForInputCheckJPAExecutor.java +++ b/core/src/main/java/org/apache/oozie/executor/jpa/CoordActionGetForInputCheckJPAExecutor.java @@ -73,34 +73,37 @@ public class CoordActionGetForInputCheckJPAExecutor implements JPAExecutor<Coord bean.setId((String) arr[0]); } if (arr[1] != null) { - bean.setJobId((String) arr[1]); + bean.setActionNumber((Integer) arr[1]); } if (arr[2] != null) { - bean.setStatus(CoordinatorAction.Status.valueOf((String) arr[2])); + bean.setJobId((String) arr[2]); } if (arr[3] != null) { - bean.setRunConfBlob((StringBlob) arr[3]); + bean.setStatus(CoordinatorAction.Status.valueOf((String) arr[3])); } if (arr[4] != null) { - bean.setNominalTime(DateUtils.toDate((Timestamp) arr[4])); + bean.setRunConfBlob((StringBlob) arr[4]); } if (arr[5] != null) { - bean.setCreatedTime(DateUtils.toDate((Timestamp) arr[5])); + bean.setNominalTime(DateUtils.toDate((Timestamp) arr[5])); } if (arr[6] != null) { - bean.setActionXmlBlob((StringBlob) arr[6]); + bean.setCreatedTime(DateUtils.toDate((Timestamp) arr[6])); } if (arr[7] != null) { - bean.setMissingDependenciesBlob((StringBlob) arr[7]); + bean.setActionXmlBlob((StringBlob) arr[7]); } if (arr[8] != null) { - bean.setPushMissingDependenciesBlob((StringBlob) arr[8]); + bean.setMissingDependenciesBlob((StringBlob) arr[8]); } if (arr[9] != null) { - bean.setTimeOut((Integer) arr[9]); + bean.setPushMissingDependenciesBlob((StringBlob) arr[9]); } if (arr[10] != null) { - bean.setExternalId((String)arr[10]); + bean.setTimeOut((Integer) arr[10]); + } + if (arr[11] != null) { + bean.setExternalId((String)arr[11]); } return bean; } http://git-wip-us.apache.org/repos/asf/oozie/blob/08fa157d/core/src/main/java/org/apache/oozie/executor/jpa/CoordJobGetReadyActionsJPAExecutor.java ---------------------------------------------------------------------- diff --git a/core/src/main/java/org/apache/oozie/executor/jpa/CoordJobGetReadyActionsJPAExecutor.java b/core/src/main/java/org/apache/oozie/executor/jpa/CoordJobGetReadyActionsJPAExecutor.java index af81091..3b1cda2 100644 --- a/core/src/main/java/org/apache/oozie/executor/jpa/CoordJobGetReadyActionsJPAExecutor.java +++ b/core/src/main/java/org/apache/oozie/executor/jpa/CoordJobGetReadyActionsJPAExecutor.java @@ -18,6 +18,7 @@ package org.apache.oozie.executor.jpa; +import java.sql.Timestamp; import java.util.ArrayList; import java.util.List; @@ -27,6 +28,7 @@ import javax.persistence.Query; import org.apache.oozie.CoordinatorActionBean; import org.apache.oozie.ErrorCode; import org.apache.oozie.client.CoordinatorAction; +import org.apache.oozie.util.DateUtils; import org.apache.oozie.util.ParamChecker; /** @@ -35,13 +37,11 @@ import org.apache.oozie.util.ParamChecker; public class CoordJobGetReadyActionsJPAExecutor implements JPAExecutor<List<CoordinatorActionBean>> { private String coordJobId = null; - private int numResults; private String executionOrder = null; - public CoordJobGetReadyActionsJPAExecutor(String coordJobId, int numResults, String executionOrder) { + public CoordJobGetReadyActionsJPAExecutor(String coordJobId, String executionOrder) { ParamChecker.notNull(coordJobId, "coordJobId"); this.coordJobId = coordJobId; - this.numResults = numResults; this.executionOrder = executionOrder; } @@ -65,9 +65,6 @@ public class CoordJobGetReadyActionsJPAExecutor implements JPAExecutor<List<Coor } q.setParameter("jobId", coordJobId); - if (numResults > 0) { - q.setMaxResults(numResults); - } List<Object[]> objectArrList = q.getResultList(); actionBeans = new ArrayList<CoordinatorActionBean>(); for (Object[] arr : objectArrList) { @@ -87,13 +84,22 @@ public class CoordJobGetReadyActionsJPAExecutor implements JPAExecutor<List<Coor bean.setId((String) arr[0]); } if (arr[1] != null) { - bean.setJobId((String) arr[1]); + bean.setActionNumber((Integer) arr[1]); } if (arr[2] != null) { - bean.setStatus(CoordinatorAction.Status.valueOf((String) arr[2])); + bean.setJobId((String) arr[2]); } if (arr[3] != null) { - bean.setPending((Integer) arr[3]); + bean.setStatus(CoordinatorAction.Status.valueOf((String) arr[3])); + } + if (arr[4] != null) { + bean.setPending((Integer) arr[4]); + } + if (arr[5] != null) { + bean.setNominalTime(DateUtils.toDate((Timestamp) arr[5])); + } + if (arr[6] != null) { + bean.setCreatedTime(DateUtils.toDate((Timestamp) arr[6])); } return bean; } http://git-wip-us.apache.org/repos/asf/oozie/blob/08fa157d/core/src/main/java/org/apache/oozie/executor/jpa/CoordJobQueryExecutor.java ---------------------------------------------------------------------- diff --git a/core/src/main/java/org/apache/oozie/executor/jpa/CoordJobQueryExecutor.java b/core/src/main/java/org/apache/oozie/executor/jpa/CoordJobQueryExecutor.java index 45d2fc9..432f075 100644 --- a/core/src/main/java/org/apache/oozie/executor/jpa/CoordJobQueryExecutor.java +++ b/core/src/main/java/org/apache/oozie/executor/jpa/CoordJobQueryExecutor.java @@ -282,7 +282,9 @@ public class CoordJobQueryExecutor extends QueryExecutor<CoordinatorJobBean, Coo bean.setFrequency((String) arr[5]); bean.setTimeUnitStr((String) arr[6]); bean.setTimeZone((String) arr[7]); - bean.setEndTime(DateUtils.toDate((Timestamp) arr[8])); + bean.setStartTime(DateUtils.toDate((Timestamp) arr[8])); + bean.setEndTime(DateUtils.toDate((Timestamp) arr[9])); + bean.setJobXmlBlob((StringBlob) arr[10]); break; case GET_COORD_JOB_ACTION_READY: bean = new CoordinatorJobBean(); @@ -294,6 +296,12 @@ public class CoordJobQueryExecutor extends QueryExecutor<CoordinatorJobBean, Coo bean.setStatusStr((String) arr[4]); bean.setExecution((String) arr[5]); bean.setConcurrency((Integer) arr[6]); + bean.setFrequency((String) arr[7]); + bean.setTimeUnitStr((String) arr[8]); + bean.setTimeZone((String) arr[9]); + bean.setStartTime(DateUtils.toDate((Timestamp) arr[10])); + bean.setEndTime(DateUtils.toDate((Timestamp) arr[11])); + bean.setJobXmlBlob((StringBlob) arr[12]); break; case GET_COORD_JOB_ACTION_KILL: bean = new CoordinatorJobBean(); http://git-wip-us.apache.org/repos/asf/oozie/blob/08fa157d/core/src/test/java/org/apache/oozie/command/coord/TestCoordActionInputCheckXCommand.java ---------------------------------------------------------------------- diff --git a/core/src/test/java/org/apache/oozie/command/coord/TestCoordActionInputCheckXCommand.java b/core/src/test/java/org/apache/oozie/command/coord/TestCoordActionInputCheckXCommand.java index f78dafd..1fe1b3a 100644 --- a/core/src/test/java/org/apache/oozie/command/coord/TestCoordActionInputCheckXCommand.java +++ b/core/src/test/java/org/apache/oozie/command/coord/TestCoordActionInputCheckXCommand.java @@ -781,57 +781,55 @@ public class TestCoordActionInputCheckXCommand extends XDataTestCase { @Test public void testLastOnly() throws Exception { + // Job starts 1 hour in the past + Date start = new Date(new Date().getTime() - 65 * 60 * 1000); + Date end = new Date(start.getTime() + 300 * 60 * 1000); CoordinatorJobBean job = addRecordToCoordJobTableForWaiting("coord-job-for-action-input-check.xml", - CoordinatorJob.Status.RUNNING, false, true); + CoordinatorJob.Status.RUNNING, start, end, false, true, 7); job.setExecutionOrder(CoordinatorJobBean.Execution.LAST_ONLY); job.setFrequency("10"); job.setTimeUnit(Timeunit.MINUTE); CoordJobQueryExecutor.getInstance().executeUpdate(CoordJobQueryExecutor.CoordJobQuery.UPDATE_COORD_JOB, job); String missingDeps = "hdfs:///dirx/filex"; - // 1 hour in the past means next nominal time is 50 mins ago so should become SKIPPED + // @1 is 65 min in the past; means next nominal time is 55 min ago so should become SKIPPED String actionId1 = addInitRecords(missingDeps, null, TZ, job, 1); - Date nomTime = new Date(new Date().getTime() - 60 * 60 * 1000); // 1 hour ago + Date nomTime = new Date(start.getTime()); // 65 min ago setCoordActionNominalTime(actionId1, nomTime.getTime()); new CoordActionInputCheckXCommand(actionId1, job.getId()).call(); checkCoordActionStatus(actionId1, CoordinatorAction.Status.SKIPPED); - // 1 hour in the future means next nominal time is 70 mins from now, so nothing should happen - String actionId2 = addInitRecords(missingDeps, null, TZ, job, 2); - nomTime = new Date(new Date().getTime() + 60 * 60 * 1000); // 1 hour from now - setCoordActionNominalTime(actionId2, nomTime.getTime()); - new CoordActionInputCheckXCommand(actionId2, job.getId()).call(); - checkCoordActionStatus(actionId2, CoordinatorAction.Status.WAITING); - - // 5 mins in the past means next nominal time is 5 mins from now, so nothing should happen - String actionId3 = addInitRecords(missingDeps, null, TZ, job, 3); - nomTime = new Date(new Date().getTime() - 5 * 60 * 1000); // 5 minutes ago - setCoordActionNominalTime(actionId3, nomTime.getTime()); - new CoordActionInputCheckXCommand(actionId3, job.getId()).call(); - checkCoordActionStatus(actionId3, CoordinatorAction.Status.WAITING); - - // 3 mins in the past means the next nominal time is 7 mins from now, but the end time is only 3 mins from now (which means + // @12 is 55 min in the future; means next nominal time is 65 min from now, so nothing should happen + String actionId12 = addInitRecords(missingDeps, null, TZ, job, 12); + nomTime = new Date(start.getTime() + 120 * 60 * 1000); // 55 min from now + setCoordActionNominalTime(actionId12, nomTime.getTime()); + new CoordActionInputCheckXCommand(actionId12, job.getId()).call(); + checkCoordActionStatus(actionId12, CoordinatorAction.Status.WAITING); + + // @7 is 5 min in the past; means next nominal time is 5 min from now, so nothing should happen + String actionId7 = addInitRecords(missingDeps, null, TZ, job, 7); + nomTime = new Date(start.getTime() + 60 * 60 * 1000); // 5 minutes ago + setCoordActionNominalTime(actionId7, nomTime.getTime()); + new CoordActionInputCheckXCommand(actionId7, job.getId()).call(); + checkCoordActionStatus(actionId7, CoordinatorAction.Status.WAITING); + + // @7 is 5 min in the past; means next nominal time is 5 min from now, but the end time is only 3 mins from now (which means // that the next nominal time would be after the end time), so nothing should happen - Date endTime = new Date(new Date().getTime() + 3 * 60 * 1000); // 3 minutes from now + Date endTime = new Date(start.getTime() + 68 * 60 * 1000); // 3 min from now job.setEndTime(endTime); CoordJobQueryExecutor.getInstance().executeUpdate(CoordJobQueryExecutor.CoordJobQuery.UPDATE_COORD_JOB, job); - String actionId4 = addInitRecords(missingDeps, null, TZ, job, 4); - nomTime = new Date(new Date().getTime() - 5 * 60 * 1000); // 5 minutes ago - setCoordActionNominalTime(actionId4, nomTime.getTime()); - new CoordActionInputCheckXCommand(actionId4, job.getId()).call(); - checkCoordActionStatus(actionId4, CoordinatorAction.Status.WAITING); + new CoordActionInputCheckXCommand(actionId7, job.getId()).call(); + checkCoordActionStatus(actionId7, CoordinatorAction.Status.WAITING); - // 1 hour in the past means the next nominal time is 50 mins ago, but the end time is 55 mins ago (which means that the next - // nominal time would be after the end time but still in the past), so nothing should happen even though the action and the - // next nominal time are both in the past - endTime = new Date(new Date().getTime() - 55 * 60 * 1000); // 55 mins ago + // @1 is 65 min in the past; which means next nominal time is 55 min ago, but the end time is 60 min ago (which means that + // the next nominal time would be after the end time but still in the past), so nothing should happen even though the action + // and the next nominal time are both in the past + endTime = new Date(start.getTime() + 5 * 60 * 1000); // 60 min ago job.setEndTime(endTime); CoordJobQueryExecutor.getInstance().executeUpdate(CoordJobQueryExecutor.CoordJobQuery.UPDATE_COORD_JOB, job); - String actionId5 = addInitRecords(missingDeps, null, TZ, job, 5); - nomTime = new Date(new Date().getTime() - 60 * 60 * 1000); // 1 hour ago - setCoordActionNominalTime(actionId5, nomTime.getTime()); - new CoordActionInputCheckXCommand(actionId5, job.getId()).call(); - checkCoordActionStatus(actionId5, CoordinatorAction.Status.WAITING); + setCoordActionStatus(actionId1, CoordinatorAction.Status.WAITING); + new CoordActionInputCheckXCommand(actionId1, job.getId()).call(); + checkCoordActionStatus(actionId1, CoordinatorAction.Status.WAITING); } @Test @@ -874,41 +872,6 @@ public class TestCoordActionInputCheckXCommand extends XDataTestCase { } - protected CoordinatorJobBean addRecordToCoordJobTableForWaiting(String testFileName, CoordinatorJob.Status status, - Date start, Date end, boolean pending, boolean doneMatd, int lastActionNum) throws Exception { - - String testDir = getTestCaseDir(); - CoordinatorJobBean coordJob = createCoordJob(testFileName, status, start, end, pending, doneMatd, lastActionNum); - String appXml = getCoordJobXmlForWaiting(testFileName, testDir); - coordJob.setJobXml(appXml); - - try { - JPAService jpaService = Services.get().get(JPAService.class); - assertNotNull(jpaService); - CoordJobInsertJPAExecutor coordInsertCmd = new CoordJobInsertJPAExecutor(coordJob); - jpaService.execute(coordInsertCmd); - } - catch (JPAExecutorException je) { - je.printStackTrace(); - fail("Unable to insert the test coord job record to table"); - throw je; - } - - return coordJob; - } - - protected String getCoordJobXmlForWaiting(String testFileName, String testDir) { - try { - Reader reader = IOUtils.getResourceAsReader(testFileName, -1); - String appXml = IOUtils.getReaderAsString(reader, -1); - appXml = appXml.replaceAll("#testDir", testDir); - return appXml; - } - catch (IOException ioe) { - throw new RuntimeException(XLog.format("Could not get " + testFileName, ioe)); - } - } - protected CoordinatorActionBean addRecordToCoordActionTableForWaiting(String jobId, int actionNum, CoordinatorAction.Status status, String resourceXmlName) throws Exception { CoordinatorActionBean action = createCoordAction(jobId, actionNum, status, resourceXmlName, 0, TZ, null); @@ -1053,24 +1016,4 @@ public class TestCoordActionInputCheckXCommand extends XDataTestCase { throw new Exception("Action ID " + actionId + " was not stored properly in db"); } } - - private CoordinatorActionBean checkCoordActionStatus(final String actionId, final CoordinatorAction.Status stat) - throws Exception { - final JPAService jpaService = Services.get().get(JPAService.class); - waitFor(5 * 1000, new Predicate() { - @Override - public boolean evaluate() throws Exception { - try { - CoordinatorActionBean action = jpaService.execute(new CoordActionGetJPAExecutor(actionId)); - return stat.equals(action.getStatus()); - } - catch (JPAExecutorException se) { - throw new Exception("Action ID " + actionId + " was not stored properly in db"); - } - } - }); - CoordinatorActionBean action = jpaService.execute(new CoordActionGetJPAExecutor(actionId)); - assertEquals(stat, action.getStatus()); - return action; - } } http://git-wip-us.apache.org/repos/asf/oozie/blob/08fa157d/core/src/test/java/org/apache/oozie/command/coord/TestCoordActionReadyXCommand.java ---------------------------------------------------------------------- diff --git a/core/src/test/java/org/apache/oozie/command/coord/TestCoordActionReadyXCommand.java b/core/src/test/java/org/apache/oozie/command/coord/TestCoordActionReadyXCommand.java new file mode 100644 index 0000000..425ab52 --- /dev/null +++ b/core/src/test/java/org/apache/oozie/command/coord/TestCoordActionReadyXCommand.java @@ -0,0 +1,100 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.oozie.command.coord; + +import org.apache.oozie.CoordinatorActionBean; +import org.apache.oozie.CoordinatorJobBean; +import org.apache.oozie.client.CoordinatorAction; +import org.apache.oozie.client.CoordinatorJob.Execution; +import org.apache.oozie.client.CoordinatorJob.Timeunit; +import org.apache.oozie.client.Job; +import org.apache.oozie.service.ConfigurationService; +import org.apache.oozie.service.Services; +import org.apache.oozie.test.XDataTestCase; +import org.junit.Test; + +import java.util.Date; + +public class TestCoordActionReadyXCommand extends XDataTestCase { + protected Services services; + + @Override + protected void setUp() throws Exception { + super.setUp(); + services = new Services(); + services.init(); + } + + @Override + protected void tearDown() throws Exception { + services.destroy(); + super.tearDown(); + } + + @Test + public void testActionsInREADYLastOnly() throws Exception { + _testActionsInREADY(Execution.LAST_ONLY); + } + + @Test + public void testActionsInREADYNone() throws Exception { + ConfigurationService.set("oozie.coord.execution.none.tolerance", "5"); + _testActionsInREADY(Execution.NONE); + } + + private void _testActionsInREADY(Execution execPolicy) throws Exception { + Date now = new Date(); + Date start = new Date(now.getTime() - 16 * 60 * 1000); + Date end = new Date(now.getTime() + 20 * 60 * 1000); + CoordinatorJobBean job = createCoordJob(Job.Status.RUNNING, start, end, true, true, 5); + job.setStartTime(start); + job.setEndTime(end); + job.setExecutionOrder(execPolicy); + job.setFrequency("5"); + job.setTimeUnit(Timeunit.MINUTE); + // Concurrency=1 ensures that with the 1 RUNNING action, CoordActionReadyXCommand won't actually start another action + // (CoordActionStartXCommand), which would throw an Exception because the DB only has partial info + job.setConcurrency(1); + addRecordToCoordJobTable(job); + + final CoordinatorActionBean action1 = addRecordToCoordActionTable(job.getId(), 1, CoordinatorAction.Status.RUNNING, + "wf-no-op.xml", 1, start); + final CoordinatorActionBean action2 = addRecordToCoordActionTable(job.getId(), 2, CoordinatorAction.Status.READY, + "wf-no-op.xml", 1, new Date(start.getTime() + 5 * 60 * 1000)); + final CoordinatorActionBean action3 = addRecordToCoordActionTable(job.getId(), 3, CoordinatorAction.Status.READY, + "wf-no-op.xml", 1, new Date(start.getTime() + 10 * 60 * 1000)); + final CoordinatorActionBean action4 = addRecordToCoordActionTable(job.getId(), 4, CoordinatorAction.Status.READY, + "wf-no-op.xml", 1, new Date(start.getTime() + 15 * 60 * 1000)); + final CoordinatorActionBean action5 = addRecordToCoordActionTable(job.getId(), 5, CoordinatorAction.Status.WAITING, + "wf-no-op.xml", 1, new Date(start.getTime() + 20 * 60 * 1000)); + + checkCoordActionStatus(action1.getId(), CoordinatorAction.Status.RUNNING); + checkCoordActionStatus(action2.getId(), CoordinatorAction.Status.READY); + checkCoordActionStatus(action3.getId(), CoordinatorAction.Status.READY); + checkCoordActionStatus(action4.getId(), CoordinatorAction.Status.READY); + checkCoordActionStatus(action5.getId(), CoordinatorAction.Status.WAITING); + + new CoordActionReadyXCommand(job.getId()).call(); + checkCoordActionStatus(action1.getId(), CoordinatorAction.Status.RUNNING); + checkCoordActionStatus(action2.getId(), CoordinatorAction.Status.SKIPPED); + checkCoordActionStatus(action3.getId(), CoordinatorAction.Status.SKIPPED); + checkCoordActionStatus(action4.getId(), CoordinatorAction.Status.READY); + checkCoordActionStatus(action5.getId(), CoordinatorAction.Status.WAITING); + } +} http://git-wip-us.apache.org/repos/asf/oozie/blob/08fa157d/core/src/test/java/org/apache/oozie/command/coord/TestCoordCommandUtils.java ---------------------------------------------------------------------- diff --git a/core/src/test/java/org/apache/oozie/command/coord/TestCoordCommandUtils.java b/core/src/test/java/org/apache/oozie/command/coord/TestCoordCommandUtils.java index 652d792..7062e69 100644 --- a/core/src/test/java/org/apache/oozie/command/coord/TestCoordCommandUtils.java +++ b/core/src/test/java/org/apache/oozie/command/coord/TestCoordCommandUtils.java @@ -459,6 +459,61 @@ public class TestCoordCommandUtils extends XDataTestCase { } } + @Test + public void testComputeNextNominalTime() throws Exception { + // The 10th minute of every hour + CoordinatorJobBean job = new CoordinatorJobBean(); + CoordinatorActionBean action = new CoordinatorActionBean(); + job.setFrequency("10 * * * *"); + job.setTimeUnit(CoordinatorJob.Timeunit.CRON); + job.setTimeZone("America/Los_Angeles"); + job.setStartTime(DateUtils.parseDateOozieTZ("2015-07-21T08:00Z")); + job.setEndTime(DateUtils.parseDateOozieTZ("2015-07-21T13:50Z")); + job.setJobXml("<coordinator-app name=\"test\" end_of_duration='NONE'></coordinator-app>"); + // @5 occurs at the 10th minute of the 5th hour after the starting time (12:10) + action.setNominalTime(DateUtils.parseDateOozieTZ("2015-07-21T12:10Z")); + action.setActionNumber(5); + // The next nominal time should be at the 10th minute of the next hour (13:10) + assertEquals(DateUtils.parseDateOozieTZ("2015-07-21T13:10Z"), CoordCommandUtils.computeNextNominalTime(job, action)); + job.setEndTime(DateUtils.parseDateOozieTZ("2015-07-21T13:00Z")); + assertNull(CoordCommandUtils.computeNextNominalTime(job, action)); + + // Every 10 hours + job = new CoordinatorJobBean(); + action = new CoordinatorActionBean(); + job.setFrequency("10"); + job.setTimeUnit(CoordinatorJob.Timeunit.HOUR); + job.setTimeZone("America/Los_Angeles"); + job.setStartTime(DateUtils.parseDateOozieTZ("2015-07-21T14:00Z")); + job.setEndTime(DateUtils.parseDateOozieTZ("2015-07-25T23:00Z")); + job.setJobXml("<coordinator-app name=\"test\" end_of_duration='NONE'></coordinator-app>"); + // @5 occurs 50 hours after the start time (23T06:00) + action.setNominalTime(DateUtils.parseDateOozieTZ("2015-07-23T06:00Z")); + action.setActionNumber(5); + // The next nominal time should be 10 hours after that (23T16:00) + assertEquals(DateUtils.parseDateOozieTZ("2015-07-23T16:00Z"), CoordCommandUtils.computeNextNominalTime(job, action)); + job.setEndTime(DateUtils.parseDateOozieTZ("2015-07-23T09:00Z")); + assertNull(CoordCommandUtils.computeNextNominalTime(job, action)); + + // Every 10 days, at the end of the day + job = new CoordinatorJobBean(); + action = new CoordinatorActionBean(); + job.setFrequency("10"); + job.setTimeUnit(CoordinatorJob.Timeunit.END_OF_DAY); + job.setTimeZone("America/Los_Angeles"); + job.setStartTime(DateUtils.parseDateOozieTZ("2015-07-21T14:00Z")); + job.setEndTime(DateUtils.parseDateOozieTZ("2015-09-21T23:00Z")); + job.setJobXml("<coordinator-app name=\"test\" end_of_duration='END_OF_DAY'></coordinator-app>"); + // @5 occurs at 50 days, at the end of the day, after the starting time (08-31T07:00Z) + // (Timezone is America/Los_Angeles so end of day is -0700) + action.setNominalTime(DateUtils.parseDateOozieTZ("2015-08-31T07:00Z")); + action.setActionNumber(5); + // The next nominal time should be 10 days after that, at the end of the day + assertEquals(DateUtils.parseDateOozieTZ("2015-09-10T07:00Z"), CoordCommandUtils.computeNextNominalTime(job, action)); + job.setEndTime(DateUtils.parseDateOozieTZ("2015-09-08T23:00Z")); + assertNull(CoordCommandUtils.computeNextNominalTime(job, action)); + } + protected CoordinatorJobBean addRecordToCoordJobTable(CoordinatorJob.Status status, Date startTime, Date endTime, String freq) throws Exception { CoordinatorJobBean coordJob = createCoordJob(status, startTime, endTime, false, false, 0); http://git-wip-us.apache.org/repos/asf/oozie/blob/08fa157d/core/src/test/java/org/apache/oozie/command/coord/TestCoordMaterializeTransitionXCommand.java ---------------------------------------------------------------------- diff --git a/core/src/test/java/org/apache/oozie/command/coord/TestCoordMaterializeTransitionXCommand.java b/core/src/test/java/org/apache/oozie/command/coord/TestCoordMaterializeTransitionXCommand.java index 77f5518..29e7ca1 100644 --- a/core/src/test/java/org/apache/oozie/command/coord/TestCoordMaterializeTransitionXCommand.java +++ b/core/src/test/java/org/apache/oozie/command/coord/TestCoordMaterializeTransitionXCommand.java @@ -929,29 +929,6 @@ public class TestCoordMaterializeTransitionXCommand extends XDataTestCase { } } - private CoordinatorJobBean addRecordToCoordJobTableForWaiting(String testFileName, CoordinatorJob.Status status, - Date start, Date end, boolean pending, boolean doneMatd, int lastActionNum) throws Exception { - - String testDir = getTestCaseDir(); - CoordinatorJobBean coordJob = createCoordJob(testFileName, status, start, end, pending, doneMatd, lastActionNum); - String appXml = getCoordJobXmlForWaiting(testFileName, testDir); - coordJob.setJobXml(appXml); - - try { - JPAService jpaService = Services.get().get(JPAService.class); - assertNotNull(jpaService); - CoordJobInsertJPAExecutor coordInsertCmd = new CoordJobInsertJPAExecutor(coordJob); - jpaService.execute(coordInsertCmd); - } - catch (JPAExecutorException je) { - je.printStackTrace(); - fail("Unable to insert the test coord job record to table"); - throw je; - } - - return coordJob; - } - private long getDSTOffset(TimeZone tz, Date d1, Date d2) { if (tz.inDaylightTime(d1) && !tz.inDaylightTime(d2)) { Calendar cal = Calendar.getInstance(tz); http://git-wip-us.apache.org/repos/asf/oozie/blob/08fa157d/core/src/test/java/org/apache/oozie/executor/jpa/TestCoordActionGetForInputCheckJPAExecutor.java ---------------------------------------------------------------------- diff --git a/core/src/test/java/org/apache/oozie/executor/jpa/TestCoordActionGetForInputCheckJPAExecutor.java b/core/src/test/java/org/apache/oozie/executor/jpa/TestCoordActionGetForInputCheckJPAExecutor.java index ca1d80a..3a3f53e 100644 --- a/core/src/test/java/org/apache/oozie/executor/jpa/TestCoordActionGetForInputCheckJPAExecutor.java +++ b/core/src/test/java/org/apache/oozie/executor/jpa/TestCoordActionGetForInputCheckJPAExecutor.java @@ -76,12 +76,13 @@ public class TestCoordActionGetForInputCheckJPAExecutor extends XDataTestCase { String actionNominalTime = getActionNominalTime(actionXml); // Pass the expected values - _testGetForInputCheckX(action.getId(), job.getId(), CoordinatorAction.Status.WAITING, 0, actionXml, XmlUtils - .prettyPrint(conf).toString(), DateUtils.parseDateOozieTZ(actionNominalTime), dummyCreationTime, missDeps); + _testGetForInputCheckX(action.getId(), action.getActionNumber(), job.getId(), CoordinatorAction.Status.WAITING, 0, + actionXml, XmlUtils.prettyPrint(conf).toString(), DateUtils.parseDateOozieTZ(actionNominalTime), + dummyCreationTime, missDeps); } - private void _testGetForInputCheckX(String actionId, String jobId, CoordinatorAction.Status status, int pending, + private void _testGetForInputCheckX(String actionId, int actionNum, String jobId, CoordinatorAction.Status status, int pending, String actionXml, String runConf, Date nominalTime, Date createdTime, String missDeps) throws Exception { try { JPAService jpaService = Services.get().get(JPAService.class); @@ -92,6 +93,7 @@ public class TestCoordActionGetForInputCheckJPAExecutor extends XDataTestCase { // Check for expected column values assertEquals(actionId, action.getId()); + assertEquals(actionNum, action.getActionNumber()); assertEquals(jobId, action.getJobId()); assertEquals(status, action.getStatus()); assertEquals(pending, action.getPending()); http://git-wip-us.apache.org/repos/asf/oozie/blob/08fa157d/core/src/test/java/org/apache/oozie/executor/jpa/TestCoordJobGetReadyActionsJPAExecutor.java ---------------------------------------------------------------------- diff --git a/core/src/test/java/org/apache/oozie/executor/jpa/TestCoordJobGetReadyActionsJPAExecutor.java b/core/src/test/java/org/apache/oozie/executor/jpa/TestCoordJobGetReadyActionsJPAExecutor.java index c666b87..4aa5de8 100644 --- a/core/src/test/java/org/apache/oozie/executor/jpa/TestCoordJobGetReadyActionsJPAExecutor.java +++ b/core/src/test/java/org/apache/oozie/executor/jpa/TestCoordJobGetReadyActionsJPAExecutor.java @@ -50,52 +50,22 @@ public class TestCoordJobGetReadyActionsJPAExecutor extends XDataTestCase { } public void testCoordActionGetFIFO() throws Exception { - CoordinatorJobBean job = addRecordToCoordJobTable(CoordinatorJob.Status.RUNNING, false, false); - Date nomTime = new Date(); - addRecordToCoordActionTable(job.getId(), 1, CoordinatorAction.Status.WAITING, "coord-action-get.xml", 0, nomTime); - nomTime.setTime(nomTime.getTime() + 1000); - addRecordToCoordActionTable(job.getId(), 2, CoordinatorAction.Status.WAITING, "coord-action-get.xml", 0, nomTime); - nomTime.setTime(nomTime.getTime() + 1000); - addRecordToCoordActionTable(job.getId(), 3, CoordinatorAction.Status.WAITING, "coord-action-get.xml", 0, nomTime); - _testGetReadyActions(job.getId(), 0, "FIFO"); - - cleanUpDBTables(); - job = addRecordToCoordJobTable(CoordinatorJob.Status.RUNNING, false, false); - nomTime = new Date(); - addRecordToCoordActionTable(job.getId(), 1, CoordinatorAction.Status.READY, "coord-action-get.xml", 0, nomTime); - nomTime.setTime(nomTime.getTime() + 1000); - addRecordToCoordActionTable(job.getId(), 2, CoordinatorAction.Status.READY, "coord-action-get.xml", 0, nomTime); - nomTime.setTime(nomTime.getTime() + 1000); - addRecordToCoordActionTable(job.getId(), 3, CoordinatorAction.Status.WAITING, "coord-action-get.xml", 0, nomTime); - List<CoordinatorActionBean> actions = _testGetReadyActions(job.getId(), 2, "FIFO"); - assertEquals((job.getId() + "@1"), actions.get(0).getId()); - assertEquals((job.getId() + "@2"), actions.get(1).getId()); + _testGetReadyActions("FIFO", false); } public void testCoordActionGetLIFO() throws Exception { - CoordinatorJobBean job = addRecordToCoordJobTable(CoordinatorJob.Status.RUNNING, false, false); - Date nomTime = new Date(); - addRecordToCoordActionTable(job.getId(), 1, CoordinatorAction.Status.WAITING, "coord-action-get.xml", 0, nomTime); - nomTime.setTime(nomTime.getTime() + 1000); - addRecordToCoordActionTable(job.getId(), 2, CoordinatorAction.Status.WAITING, "coord-action-get.xml", 0, nomTime); - nomTime.setTime(nomTime.getTime() + 1000); - addRecordToCoordActionTable(job.getId(), 3, CoordinatorAction.Status.WAITING, "coord-action-get.xml", 0, nomTime); - _testGetReadyActions(job.getId(), 0, "LIFO"); - - cleanUpDBTables(); - job = addRecordToCoordJobTable(CoordinatorJob.Status.RUNNING, false, false); - nomTime = new Date(); - addRecordToCoordActionTable(job.getId(), 1, CoordinatorAction.Status.READY, "coord-action-get.xml", 0, nomTime); - nomTime.setTime(nomTime.getTime() + 1000); - addRecordToCoordActionTable(job.getId(), 2, CoordinatorAction.Status.READY, "coord-action-get.xml", 0, nomTime); - nomTime.setTime(nomTime.getTime() + 1000); - addRecordToCoordActionTable(job.getId(), 3, CoordinatorAction.Status.WAITING, "coord-action-get.xml", 0, nomTime); - List<CoordinatorActionBean> actions = _testGetReadyActions(job.getId(), 2, "LIFO"); - assertEquals((job.getId() + "@2"), actions.get(0).getId()); - assertEquals((job.getId() + "@1"), actions.get(1).getId()); + _testGetReadyActions("LIFO", true); } public void testCoordActionGetLAST_ONLY() throws Exception { + _testGetReadyActions("LAST_ONLY", true); + } + + public void testCoordActionGetNONE() throws Exception { + _testGetReadyActions("NONE", true); + } + + private void _testGetReadyActions(String execution, boolean reverseOrder) throws Exception { CoordinatorJobBean job = addRecordToCoordJobTable(CoordinatorJob.Status.RUNNING, false, false); Date nomTime = new Date(); addRecordToCoordActionTable(job.getId(), 1, CoordinatorAction.Status.WAITING, "coord-action-get.xml", 0, nomTime); @@ -103,25 +73,40 @@ public class TestCoordJobGetReadyActionsJPAExecutor extends XDataTestCase { addRecordToCoordActionTable(job.getId(), 2, CoordinatorAction.Status.WAITING, "coord-action-get.xml", 0, nomTime); nomTime.setTime(nomTime.getTime() + 1000); addRecordToCoordActionTable(job.getId(), 3, CoordinatorAction.Status.WAITING, "coord-action-get.xml", 0, nomTime); - _testGetReadyActions(job.getId(), 0, "LAST_ONLY"); + getReadyActions(job.getId(), 0, execution); cleanUpDBTables(); job = addRecordToCoordJobTable(CoordinatorJob.Status.RUNNING, false, false); + int[] actionIndexes = reverseOrder ? new int[]{1, 0} : new int[]{0, 1}; nomTime = new Date(); - addRecordToCoordActionTable(job.getId(), 1, CoordinatorAction.Status.READY, "coord-action-get.xml", 0, nomTime); + Date[] createdTimes = new Date[2]; + Date[] nomTimes = new Date[2]; + nomTimes[actionIndexes[0]] = new Date(nomTime.getTime()); + createdTimes[actionIndexes[0]] = + addRecordToCoordActionTable(job.getId(), 1, CoordinatorAction.Status.READY, "coord-action-get.xml", 0, nomTime) + .getCreatedTime(); nomTime.setTime(nomTime.getTime() + 1000); - addRecordToCoordActionTable(job.getId(), 2, CoordinatorAction.Status.READY, "coord-action-get.xml", 0, nomTime); + nomTimes[actionIndexes[1]] = new Date(nomTime.getTime()); + createdTimes[actionIndexes[1]] = + addRecordToCoordActionTable(job.getId(), 2, CoordinatorAction.Status.READY, "coord-action-get.xml", 0, nomTime) + .getCreatedTime(); nomTime.setTime(nomTime.getTime() + 1000); addRecordToCoordActionTable(job.getId(), 3, CoordinatorAction.Status.WAITING, "coord-action-get.xml", 0, nomTime); - List<CoordinatorActionBean> actions = _testGetReadyActions(job.getId(), 2, "LAST_ONLY"); - assertEquals((job.getId() + "@2"), actions.get(0).getId()); - assertEquals((job.getId() + "@1"), actions.get(1).getId()); + List<CoordinatorActionBean> actions = getReadyActions(job.getId(), 2, execution); + for (int i : actionIndexes) { + assertEquals(job.getId() + "@" + (actionIndexes[i] + 1), actions.get(i).getId()); + assertEquals(actionIndexes[i] + 1, actions.get(i).getActionNumber()); + assertEquals(job.getId(), actions.get(i).getJobId()); + assertEquals(CoordinatorAction.Status.READY, actions.get(i).getStatus()); + assertEquals(nomTimes[i], actions.get(i).getNominalTime()); + assertEquals(createdTimes[i], actions.get(i).getCreatedTime()); + } } - private List<CoordinatorActionBean> _testGetReadyActions(String jobId, int expected, String execution) throws Exception { + private List<CoordinatorActionBean> getReadyActions(String jobId, int expected, String execution) throws Exception { JPAService jpaService = Services.get().get(JPAService.class); assertNotNull(jpaService); - CoordJobGetReadyActionsJPAExecutor actionGetCmd = new CoordJobGetReadyActionsJPAExecutor(jobId, 10, execution); + CoordJobGetReadyActionsJPAExecutor actionGetCmd = new CoordJobGetReadyActionsJPAExecutor(jobId, execution); List<CoordinatorActionBean> actions = jpaService.execute(actionGetCmd); assertEquals(expected, actions.size()); return actions; http://git-wip-us.apache.org/repos/asf/oozie/blob/08fa157d/core/src/test/java/org/apache/oozie/executor/jpa/TestCoordJobQueryExecutor.java ---------------------------------------------------------------------- diff --git a/core/src/test/java/org/apache/oozie/executor/jpa/TestCoordJobQueryExecutor.java b/core/src/test/java/org/apache/oozie/executor/jpa/TestCoordJobQueryExecutor.java index ee7ad6a..a0012fb 100644 --- a/core/src/test/java/org/apache/oozie/executor/jpa/TestCoordJobQueryExecutor.java +++ b/core/src/test/java/org/apache/oozie/executor/jpa/TestCoordJobQueryExecutor.java @@ -209,9 +209,10 @@ public class TestCoordJobQueryExecutor extends XDataTestCase { assertEquals(bean.getFrequency(), retBean.getFrequency()); assertEquals(bean.getTimeUnit(), retBean.getTimeUnit()); assertEquals(bean.getTimeZone(), retBean.getTimeZone()); + assertEquals(bean.getStartTime(), retBean.getStartTime()); assertEquals(bean.getEndTime(), retBean.getEndTime()); + assertEquals(bean.getJobXml(), retBean.getJobXml()); assertNull(retBean.getConf()); - assertNull(retBean.getJobXmlBlob()); assertNull(retBean.getOrigJobXmlBlob()); assertNull(retBean.getSlaXmlBlob()); // GET_COORD_JOB_ACTION_READY @@ -223,8 +224,13 @@ public class TestCoordJobQueryExecutor extends XDataTestCase { assertEquals(bean.getStatusStr(), retBean.getStatusStr()); assertEquals(bean.getExecution(), retBean.getExecution()); assertEquals(bean.getConcurrency(), retBean.getConcurrency()); + assertEquals(bean.getFrequency(), retBean.getFrequency()); + assertEquals(bean.getTimeUnit(), retBean.getTimeUnit()); + assertEquals(bean.getTimeZone(), retBean.getTimeZone()); + assertEquals(bean.getStartTime(), retBean.getStartTime()); + assertEquals(bean.getEndTime(), retBean.getEndTime()); + assertEquals(bean.getJobXml(), retBean.getJobXml()); assertNull(retBean.getConf()); - assertNull(retBean.getJobXmlBlob()); assertNull(retBean.getOrigJobXmlBlob()); assertNull(retBean.getSlaXmlBlob()); // GET_COORD_JOB_ACTION_KILL http://git-wip-us.apache.org/repos/asf/oozie/blob/08fa157d/core/src/test/java/org/apache/oozie/test/XDataTestCase.java ---------------------------------------------------------------------- diff --git a/core/src/test/java/org/apache/oozie/test/XDataTestCase.java b/core/src/test/java/org/apache/oozie/test/XDataTestCase.java index e6a7d9c..4c45dca 100644 --- a/core/src/test/java/org/apache/oozie/test/XDataTestCase.java +++ b/core/src/test/java/org/apache/oozie/test/XDataTestCase.java @@ -1674,9 +1674,20 @@ public abstract class XDataTestCase extends XHCatTestCase { protected CoordinatorJobBean addRecordToCoordJobTableForWaiting(String testFileName, CoordinatorJob.Status status, boolean pending, boolean doneMatd) throws Exception { + CoordinatorJobBean coordJob = createCoordJob(status, pending, doneMatd); + return addRecordToCoordJobTableForWaiting(testFileName, coordJob); + } + + protected CoordinatorJobBean addRecordToCoordJobTableForWaiting(String testFileName, CoordinatorJob.Status status, + Date start, Date end, boolean pending, boolean doneMatd, int lastActionNum) throws Exception { + CoordinatorJobBean coordJob = createCoordJob(testFileName, status, start, end, pending, doneMatd, lastActionNum); + return addRecordToCoordJobTableForWaiting(testFileName, coordJob); + } + + protected CoordinatorJobBean addRecordToCoordJobTableForWaiting(String testFileName, CoordinatorJobBean coordJob) + throws Exception { String testDir = getTestCaseDir(); - CoordinatorJobBean coordJob = createCoordJob(status, pending, doneMatd); String appXml = getCoordJobXmlForWaiting(testFileName, testDir); coordJob.setJobXml(appXml); @@ -1721,6 +1732,13 @@ public abstract class XDataTestCase extends XHCatTestCase { CoordActionQueryExecutor.getInstance().executeUpdate(CoordActionQuery.UPDATE_COORD_ACTION, action); } + protected void setCoordActionStatus(String actionId, CoordinatorAction.Status status) throws Exception { + JPAService jpaService = Services.get().get(JPAService.class); + CoordinatorActionBean action = jpaService.execute(new CoordActionGetJPAExecutor(actionId)); + action.setStatus(status); + CoordActionQueryExecutor.getInstance().executeUpdate(CoordActionQuery.UPDATE_COORD_ACTION, action); + } + protected void setMissingDependencies(String actionId, String missingDependencies) throws Exception { JPAService jpaService = Services.get().get(JPAService.class); CoordinatorActionBean action = jpaService.execute(new CoordActionGetJPAExecutor(actionId)); @@ -1752,4 +1770,23 @@ public abstract class XDataTestCase extends XHCatTestCase { } } + protected CoordinatorActionBean checkCoordActionStatus(final String actionId, final CoordinatorAction.Status stat) + throws Exception { + final JPAService jpaService = Services.get().get(JPAService.class); + waitFor(5 * 1000, new Predicate() { + @Override + public boolean evaluate() throws Exception { + try { + CoordinatorActionBean action = jpaService.execute(new CoordActionGetJPAExecutor(actionId)); + return stat.equals(action.getStatus()); + } + catch (JPAExecutorException se) { + throw new Exception("Action ID " + actionId + " was not stored properly in db"); + } + } + }); + CoordinatorActionBean action = jpaService.execute(new CoordActionGetJPAExecutor(actionId)); + assertEquals(stat, action.getStatus()); + return action; + } } http://git-wip-us.apache.org/repos/asf/oozie/blob/08fa157d/docs/src/site/twiki/CoordinatorFunctionalSpec.twiki ---------------------------------------------------------------------- diff --git a/docs/src/site/twiki/CoordinatorFunctionalSpec.twiki b/docs/src/site/twiki/CoordinatorFunctionalSpec.twiki index faff37f..4ada5cf 100644 --- a/docs/src/site/twiki/CoordinatorFunctionalSpec.twiki +++ b/docs/src/site/twiki/CoordinatorFunctionalSpec.twiki @@ -976,13 +976,13 @@ A synchronous coordinator definition is a is defined by a name, start time and e * *%BLUE% timezone:%ENDCOLOR%* The timezone of the coordinator application. * *%BLUE% frequency: %ENDCOLOR%* The frequency, in minutes, to materialize actions. Refer to section #4 'Time Interval Representation' for syntax details. * Control information: - * *%BLUE% timeout: %ENDCOLOR%* The maximum time, in minutes, that a materialized action will be waiting for the additional conditions to be satisfied before being discarded. A timeout of =0= indicates that at the time of materialization all the other conditions must be satisfied, else the action will be discarded. A timeout of =0= indicates that if all the input events are not satisfied at the time of action materialization, the action should timeout immediately. A timeout of =-1= indicates no timeout, the materialized action will wait forever for the other conditions to be satisfied. The default value is =-1=. + * *%BLUE% timeout: %ENDCOLOR%* The maximum time, in minutes, that a materialized action will be waiting for the additional conditions to be satisfied before being discarded. A timeout of =0= indicates that at the time of materialization all the other conditions must be satisfied, else the action will be discarded. A timeout of =0= indicates that if all the input events are not satisfied at the time of action materialization, the action should timeout immediately. A timeout of =-1= indicates no timeout, the materialized action will wait forever for the other conditions to be satisfied. The default value is =-1=. The timeout can only cause a =WAITING= action to transition to =TIMEDOUT=; once the data dependency is satisified, a =WAITING= action transitions to =READY=, and the timeout no longer has any affect, even if the action hasn't transitioned to =SUBMITTED= or =RUNNING= when it expires. * *%BLUE% concurrency: %ENDCOLOR%* The maximum number of actions for this job that can be running at the same time. This value allows to materialize and submit multiple instances of the coordinator app, and allows operations to catchup on delayed processing. The default value is =1=. * *%BLUE% execution: %ENDCOLOR%* Specifies the execution order if multiple instances of the coordinator job have satisfied their execution criteria. Valid values are: * =FIFO= (oldest first) *default*. * =LIFO= (newest first). * =LAST_ONLY= (see explanation below). - * =NONE= (discards all older materialization, see explanation below). + * =NONE= (see explanation below). * *%BLUE% throttle: %ENDCOLOR%* The maximum coordinator actions are allowed to be in WAITING state concurrently. The default value is =12=. * *%BLUE% datasets: %ENDCOLOR%* The datasets coordinator application uses. * *%BLUE% input-events: %ENDCOLOR%* The coordinator job input events. @@ -1004,23 +1004,23 @@ A synchronous coordinator definition is a is defined by a name, start time and e *LAST_ONLY:* While =FIFO= and =LIFO= simply specify the order in which READY actions should be executed, =LAST_ONLY= can actually cause some actions to be SKIPPED and is a little harder to understand. When =LAST_ONLY= is set, an action that is =WAITING= or =READY= will be =SKIPPED= when the current time is past the next action's nominal time. For example, suppose action 1 and 2 -are both =WAITING=, the current time is 5:00pm, and action 2's nominal time is 5:10pm. In 10 minutes from now, at 5:10pm, action 1 -will become SKIPPED, assuming it doesn't transition to =SUBMITTED= (or a terminal state) before then. Another way of thinking about -this is to view it as similar to setting the =timeout= equal to the =frequency=, except that the =SKIPPED= status doesn't cause the -coordinator job to eventually become =DONEWITHERROR= and can actually become =SUCCEEDED= (i.e. it's a "good" version -of =TIMEDOUT=). =LAST_ONLY= is useful if you want a recurring job, but do not actually care about the individual instances and just +are both =READY=, the current time is 5:00pm, and action 2's nominal time is 5:10pm. In 10 minutes from now, at 5:10pm, action 1 +will become SKIPPED, assuming it doesn't transition to =SUBMITTED= (or a terminal state) before then. This sounds similar to the +timeout control, but there are some important differences: + * The timeout time is configurable while the =LAST_ONLY= time is effectively the frequency. + * Reaching the timeout causes an action to transition to =TIMEDOUT=, which will cause the Coordinator Job to become =RUNNINGWITHERROR= and eventually =DONEWITHERROR=. With =LAST_ONLY=, an action becomes =SKIPPED= and the Coordinator Job remains =RUNNING= and eventually =DONE=. + * The timeout is looking satisfying the data dependency, while =LAST_ONLY= is looking at the action itself. This means that the timeout can only cause a transition from =WAITING=, while =LAST_ONLY= can cause a transition from =WAITING= or =READY=. + +=LAST_ONLY= is useful if you want a recurring job, but do not actually care about the individual instances and just always want the latest action. For example, if you have a coordinator running every 10 minutes and take Oozie down for 1 hour, when -Oozie comes back, there would normally be 6 actions =READY= to run. However, with =LAST_ONLY=, only the current one will go -to =SUBMITTED= and =RUNNING=; the others will go to SKIPPED. - -*NONE:* Similar to LAST_ONLY except all older materializations are skipped. When =NONE= is set, an action that is =WAITING= -or =READY= will be =SKIPPED= when the current time is more than a certain configured number of minutes (tolerance) past the action's -nominal time. By default, the threshold is 1 minute. For example, suppose action 1 and 2 -are both =WAITING=, the current time is 5:20pm, and both actions' nominal times are before 5:19pm. Both actions -will become SKIPPED, assuming they don't transition to =SUBMITTED= (or a terminal state) before then. Another way of thinking about -this is to view it as similar to setting the =timeout= equal to 1 minute which is the smallest time unit, except that the =SKIPPED= status doesn't cause the -coordinator job to eventually become =DONEWITHERROR= and can actually become =SUCCEEDED= (i.e. it's a "good" version -of =TIMEDOUT=). +Oozie comes back, there would normally be 6 actions =WAITING= or =READY= to run. However, with =LAST_ONLY=, only the current one +will go to =SUBMITTED= and then =RUNNING=; the others will go to =SKIPPED=. + +*NONE:* Similar to =LAST_ONLY= except instead of looking at the next action's nominal time, it looks +at =oozie.coord.execution.none.tolerance= in oozie-site.xml (default is 1 minute). When =NONE= is set, an action that is =WAITING= +or =READY= will be =SKIPPED= when the current time is more than the configured number of minutes (tolerance) past that action's +nominal time. For example, suppose action 1 and 2 are both =READY=, the current time is 5:20pm, and both actions' nominal times are +before 5:19pm. Both actions will become =SKIPPED=, assuming they don't transition to =SUBMITTED= (or a terminal state) before then. *%PURPLE% Syntax: %ENDCOLOR%* http://git-wip-us.apache.org/repos/asf/oozie/blob/08fa157d/release-log.txt ---------------------------------------------------------------------- diff --git a/release-log.txt b/release-log.txt index 6a045b0..f583515 100644 --- a/release-log.txt +++ b/release-log.txt @@ -1,5 +1,6 @@ -- Oozie 4.3.0 release (trunk - unreleased) +OOZIE-2397 LAST_ONLY and NONE don't properly handle READY actions (rkanter) OOZIE-2401 Typo in twiki docs with FilesAchives instead of FilesArchives ([email protected] via rkanter) OOZIE-2168 Oozie flow and action names have 50 char limit (akshayrai09, me.venkatr via rkanter) OOZIE-2346 Add sub-workflow information like the super-parent id and workflow depth into the 'oozie.job.info' property (akshayrai09 via puru)
