Repository: oozie Updated Branches: refs/heads/master dc22c9022 -> 3de3c523d
OOZIE-1735 Support resuming of failed coordinator job and rerun of a failed coordinator action (puru via rohini) Project: http://git-wip-us.apache.org/repos/asf/oozie/repo Commit: http://git-wip-us.apache.org/repos/asf/oozie/commit/3de3c523 Tree: http://git-wip-us.apache.org/repos/asf/oozie/tree/3de3c523 Diff: http://git-wip-us.apache.org/repos/asf/oozie/diff/3de3c523 Branch: refs/heads/master Commit: 3de3c523d4645b2150e21c12f599de0cb5b333b4 Parents: dc22c90 Author: Rohini Palaniswamy <[email protected]> Authored: Thu Mar 27 17:04:29 2014 -0700 Committer: Rohini Palaniswamy <[email protected]> Committed: Thu Mar 27 17:04:29 2014 -0700 ---------------------------------------------------------------------- .../java/org/apache/oozie/cli/OozieCLI.java | 2 +- .../org/apache/oozie/client/OozieClient.java | 2 + .../command/coord/CoordChangeXCommand.java | 62 ++++++++++-- .../oozie/command/coord/CoordRerunXCommand.java | 10 -- .../command/coord/TestCoordChangeXCommand.java | 99 ++++++++++++++++++++ .../command/coord/TestCoordRerunXCommand.java | 37 ++++++++ docs/src/site/twiki/DG_CommandLineTool.twiki | 12 ++- release-log.txt | 1 + 8 files changed, 205 insertions(+), 20 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/oozie/blob/3de3c523/client/src/main/java/org/apache/oozie/cli/OozieCLI.java ---------------------------------------------------------------------- diff --git a/client/src/main/java/org/apache/oozie/cli/OozieCLI.java b/client/src/main/java/org/apache/oozie/cli/OozieCLI.java index 6dc4a3b..e1f551d 100644 --- a/client/src/main/java/org/apache/oozie/cli/OozieCLI.java +++ b/client/src/main/java/org/apache/oozie/cli/OozieCLI.java @@ -720,7 +720,7 @@ public class OozieCLI { * @return change value specified by -value. * @throws OozieCLIException */ - private String getChangeValue(CommandLine commandLine) throws OozieCLIException { + private String getChangeValue(CommandLine commandLine) throws OozieCLIException { String changeValue = commandLine.getOptionValue(CHANGE_VALUE_OPTION); if (changeValue == null) { http://git-wip-us.apache.org/repos/asf/oozie/blob/3de3c523/client/src/main/java/org/apache/oozie/client/OozieClient.java ---------------------------------------------------------------------- diff --git a/client/src/main/java/org/apache/oozie/client/OozieClient.java b/client/src/main/java/org/apache/oozie/client/OozieClient.java index 6164447..46c4288 100644 --- a/client/src/main/java/org/apache/oozie/client/OozieClient.java +++ b/client/src/main/java/org/apache/oozie/client/OozieClient.java @@ -148,6 +148,8 @@ public class OozieClient { public static final String CHANGE_VALUE_CONCURRENCY = "concurrency"; + public static final String CHANGE_VALUE_STATUS = "status"; + public static final String LIBPATH = "oozie.libpath"; public static final String USE_SYSTEM_LIBPATH = "oozie.use.system.libpath"; http://git-wip-us.apache.org/repos/asf/oozie/blob/3de3c523/core/src/main/java/org/apache/oozie/command/coord/CoordChangeXCommand.java ---------------------------------------------------------------------- diff --git a/core/src/main/java/org/apache/oozie/command/coord/CoordChangeXCommand.java b/core/src/main/java/org/apache/oozie/command/coord/CoordChangeXCommand.java index 4957330..5ecd5d5 100644 --- a/core/src/main/java/org/apache/oozie/command/coord/CoordChangeXCommand.java +++ b/core/src/main/java/org/apache/oozie/command/coord/CoordChangeXCommand.java @@ -26,6 +26,7 @@ import java.util.Map; import java.util.Map.Entry; import java.util.Set; +import org.apache.commons.lang.StringUtils; import org.apache.oozie.CoordinatorActionBean; import org.apache.oozie.CoordinatorJobBean; import org.apache.oozie.ErrorCode; @@ -33,6 +34,7 @@ import org.apache.oozie.XException; import org.apache.oozie.client.CoordinatorAction; import org.apache.oozie.client.CoordinatorJob; import org.apache.oozie.client.Job; +import org.apache.oozie.client.Job.Status; import org.apache.oozie.client.OozieClient; import org.apache.oozie.client.rest.JsonBean; import org.apache.oozie.command.CommandException; @@ -66,6 +68,7 @@ public class CoordChangeXCommand extends CoordinatorXCommand<Void> { private Date newPauseTime = null; private Date oldPauseTime = null; private boolean resetPauseTime = false; + private CoordinatorJob.Status jobStatus = null; private CoordinatorJobBean coordJob; private JPAService jpaService = null; private Job.Status prevStatus; @@ -77,6 +80,8 @@ public class CoordChangeXCommand extends CoordinatorXCommand<Void> { ALLOWED_CHANGE_OPTIONS.add("endtime"); ALLOWED_CHANGE_OPTIONS.add("concurrency"); ALLOWED_CHANGE_OPTIONS.add("pausetime"); + ALLOWED_CHANGE_OPTIONS.add(OozieClient.CHANGE_VALUE_STATUS); + } /** @@ -103,7 +108,7 @@ public class CoordChangeXCommand extends CoordinatorXCommand<Void> { Map<String, String> map = JobUtils.parseChangeValue(changeValue); if (map.size() > ALLOWED_CHANGE_OPTIONS.size()) { - throw new CommandException(ErrorCode.E1015, changeValue, "must change endtime|concurrency|pausetime"); + throw new CommandException(ErrorCode.E1015, changeValue, "must change endtime|concurrency|pausetime|status"); } java.util.Iterator<Entry<String, String>> iter = map.entrySet().iterator(); @@ -113,7 +118,7 @@ public class CoordChangeXCommand extends CoordinatorXCommand<Void> { String value = entry.getValue(); if (!ALLOWED_CHANGE_OPTIONS.contains(key)) { - throw new CommandException(ErrorCode.E1015, changeValue, "must change endtime|concurrency|pausetime"); + throw new CommandException(ErrorCode.E1015, changeValue, "must change endtime|concurrency|pausetime|status"); } if (!key.equals(OozieClient.CHANGE_VALUE_PAUSETIME) && value.equalsIgnoreCase("")) { @@ -155,6 +160,13 @@ public class CoordChangeXCommand extends CoordinatorXCommand<Void> { } } } + + if (map.containsKey(OozieClient.CHANGE_VALUE_STATUS)) { + String value = map.get(OozieClient.CHANGE_VALUE_STATUS); + if (!StringUtils.isEmpty(value)) { + jobStatus = CoordinatorJob.Status.valueOf(value); + } + } } /** @@ -215,6 +227,27 @@ public class CoordChangeXCommand extends CoordinatorXCommand<Void> { } /** + * Check if status change is valid. + * + * @param coordJob the coord job + * @param jobStatus the job status + * @throws CommandException the command exception + */ + private void checkStatusChange(CoordinatorJobBean coordJob, CoordinatorJob.Status jobStatus) + throws CommandException { + if (!jobStatus.equals(CoordinatorJob.Status.RUNNING)) { + throw new CommandException(ErrorCode.E1015, jobStatus, " must be RUNNING"); + } + + if (!(coordJob.getStatus().equals(CoordinatorJob.Status.FAILED) || coordJob.getStatus().equals( + CoordinatorJob.Status.KILLED))) { + throw new CommandException(ErrorCode.E1015, jobStatus, + " Only FAILED or KILLED job can be changed to RUNNING. Current job status is " + + coordJob.getStatus()); + } + } + + /** * Process lookahead created actions that become invalid because of the new pause time, * These actions will be deleted from DB, also the coordinator job will be updated accordingly * @@ -301,10 +334,13 @@ public class CoordChangeXCommand extends CoordinatorXCommand<Void> { * @param newPauseTime new pause time. * @throws CommandException thrown if new values are not valid. */ - private void check(CoordinatorJobBean coordJob, Date newEndTime, Integer newConcurrency, Date newPauseTime) - throws CommandException { + private void check(CoordinatorJobBean coordJob, Date newEndTime, Integer newConcurrency, Date newPauseTime, + CoordinatorJob.Status jobStatus) throws CommandException { + if (coordJob.getStatus() == CoordinatorJob.Status.KILLED) { - throw new CommandException(ErrorCode.E1016); + if (jobStatus == null || (newEndTime != null || newConcurrency != null || newPauseTime != null)) { + throw new CommandException(ErrorCode.E1016); + } } if (newEndTime != null) { @@ -314,6 +350,9 @@ public class CoordChangeXCommand extends CoordinatorXCommand<Void> { if (newPauseTime != null) { checkPauseTime(coordJob, newPauseTime); } + if (jobStatus != null) { + checkStatusChange(coordJob, jobStatus); + } } /* (non-Javadoc) @@ -379,6 +418,17 @@ public class CoordChangeXCommand extends CoordinatorXCommand<Void> { processLookaheadActions(coordJob, newPauseTime); } } + if (jobStatus != null) { + coordJob.setStatus(jobStatus); + LOG.info("Coord status is changed to RUNNING from " + prevStatus); + coordJob.setPending(); + if (jobStatus.equals(CoordinatorJob.Status.RUNNING)) { + if (coordJob.getNextMaterializedTime() != null + && coordJob.getEndTime().after(coordJob.getNextMaterializedTime())) { + coordJob.resetDoneMaterialization(); + } + } + } if (coordJob.getNextMaterializedTime() != null && coordJob.getEndTime().compareTo(coordJob.getNextMaterializedTime()) <= 0) { LOG.info("[" + coordJob.getId() + "]: all actions have been materialized, job status = " + coordJob.getStatus() @@ -442,7 +492,7 @@ public class CoordChangeXCommand extends CoordinatorXCommand<Void> { */ @Override protected void verifyPrecondition() throws CommandException,PreconditionException { - check(this.coordJob, newEndTime, newConcurrency, newPauseTime); + check(this.coordJob, newEndTime, newConcurrency, newPauseTime, jobStatus); } /* (non-Javadoc) http://git-wip-us.apache.org/repos/asf/oozie/blob/3de3c523/core/src/main/java/org/apache/oozie/command/coord/CoordRerunXCommand.java ---------------------------------------------------------------------- diff --git a/core/src/main/java/org/apache/oozie/command/coord/CoordRerunXCommand.java b/core/src/main/java/org/apache/oozie/command/coord/CoordRerunXCommand.java index 301737b..78a2614 100644 --- a/core/src/main/java/org/apache/oozie/command/coord/CoordRerunXCommand.java +++ b/core/src/main/java/org/apache/oozie/command/coord/CoordRerunXCommand.java @@ -262,16 +262,6 @@ public class CoordRerunXCommand extends RerunTransitionXCommand<CoordinatorActio @Override protected void verifyPrecondition() throws CommandException, PreconditionException { BundleStatusUpdateXCommand bundleStatusUpdate = new BundleStatusUpdateXCommand(coordJob, coordJob.getStatus()); - if (coordJob.getStatus() == CoordinatorJob.Status.FAILED) { - LOG.info("CoordRerunXCommand is not able to run, job status=" + coordJob.getStatus() + ", jobid=" + jobId); - // Call the parent so the pending flag is reset and state transition - // of bundle can happen - if (coordJob.getBundleId() != null) { - bundleStatusUpdate.call(); - } - throw new CommandException(ErrorCode.E1018, - "coordinator job is failed. So all actions are not eligible to rerun!"); - } // no actioins have been created for PREP job if (coordJob.getStatus() == CoordinatorJob.Status.PREP) { http://git-wip-us.apache.org/repos/asf/oozie/blob/3de3c523/core/src/test/java/org/apache/oozie/command/coord/TestCoordChangeXCommand.java ---------------------------------------------------------------------- diff --git a/core/src/test/java/org/apache/oozie/command/coord/TestCoordChangeXCommand.java b/core/src/test/java/org/apache/oozie/command/coord/TestCoordChangeXCommand.java index b9bbf16..fdc0a55 100644 --- a/core/src/test/java/org/apache/oozie/command/coord/TestCoordChangeXCommand.java +++ b/core/src/test/java/org/apache/oozie/command/coord/TestCoordChangeXCommand.java @@ -542,6 +542,105 @@ public class TestCoordChangeXCommand extends XDataTestCase { } } + // Status change from failed- successful + public void testCoordStatus_Failed() throws Exception { + Date start = new Date(); + Date end = new Date(start.getTime() + (5 * 60 * 60 * 1000)); // 5 hrs + String status = "status=RUNNING"; + final CoordinatorJobBean job = addRecordToCoordJobTableForPauseTimeTest(CoordinatorJob.Status.FAILED, start, + end, end, true, false, 4); + addRecordToCoordActionTable(job.getId(), 1, CoordinatorAction.Status.SUCCEEDED, "coord-action-get.xml", 0); + addRecordToCoordActionTable(job.getId(), 2, CoordinatorAction.Status.SUCCEEDED, "coord-action-get.xml", 0); + addRecordToCoordActionTable(job.getId(), 3, CoordinatorAction.Status.RUNNING, "coord-action-get.xml", 0); + addRecordToCoordActionTable(job.getId(), 4, CoordinatorAction.Status.WAITING, "coord-action-get.xml", 0); + + try { + new CoordChangeXCommand(job.getId(), status).call(); + JPAService jpaService = Services.get().get(JPAService.class); + + CoordJobGetJPAExecutor coordGetCmd = new CoordJobGetJPAExecutor(job.getId()); + CoordinatorJobBean coordJob = jpaService.execute(coordGetCmd); + + assertEquals(coordJob.getStatusStr(), "RUNNING"); + } + catch (CommandException e) { + e.printStackTrace(); + if (e.getErrorCode() != ErrorCode.E1022) { + fail("Error code should be E1022"); + } + } + } + + // Status change from Killed- successful + public void testCoordStatus_Killed() throws Exception { + Date start = new Date(); + Date end = new Date(start.getTime() + (5 * 60 * 60 * 1000)); // 5 hrs + String status = "status=RUNNING"; + final CoordinatorJobBean job = addRecordToCoordJobTableForPauseTimeTest(CoordinatorJob.Status.KILLED, start, + end, end, true, false, 4); + addRecordToCoordActionTable(job.getId(), 1, CoordinatorAction.Status.SUCCEEDED, "coord-action-get.xml", 0); + addRecordToCoordActionTable(job.getId(), 2, CoordinatorAction.Status.SUCCEEDED, "coord-action-get.xml", 0); + addRecordToCoordActionTable(job.getId(), 3, CoordinatorAction.Status.RUNNING, "coord-action-get.xml", 0); + addRecordToCoordActionTable(job.getId(), 4, CoordinatorAction.Status.WAITING, "coord-action-get.xml", 0); + + try { + new CoordChangeXCommand(job.getId(), status).call(); + JPAService jpaService = Services.get().get(JPAService.class); + + CoordJobGetJPAExecutor coordGetCmd = new CoordJobGetJPAExecutor(job.getId()); + CoordinatorJobBean coordJob = jpaService.execute(coordGetCmd); + + assertEquals(coordJob.getStatusStr(), "RUNNING"); + } + catch (CommandException e) { + e.printStackTrace(); + if (e.getErrorCode() != ErrorCode.E1022) { + fail("Error code should be E1022"); + } + } + } + + // Check status change from Succeeded- exception + public void testCoordStatus_Changefailed() throws Exception { + Date start = new Date(); + Date end = new Date(start.getTime() + (4 * 60 * 60 * 1000)); // 5 hrs + String status = "status=RUNNING"; + final CoordinatorJobBean job = addRecordToCoordJobTableForPauseTimeTest(CoordinatorJob.Status.SUCCEEDED, start, + end, end, true, false, 4); + addRecordToCoordActionTable(job.getId(), 1, CoordinatorAction.Status.SUCCEEDED, "coord-action-get.xml", 0); + addRecordToCoordActionTable(job.getId(), 2, CoordinatorAction.Status.SUCCEEDED, "coord-action-get.xml", 0); + addRecordToCoordActionTable(job.getId(), 3, CoordinatorAction.Status.RUNNING, "coord-action-get.xml", 0); + addRecordToCoordActionTable(job.getId(), 4, CoordinatorAction.Status.WAITING, "coord-action-get.xml", 0); + + try { + new CoordChangeXCommand(job.getId(), status).call(); + } + catch (CommandException e) { + assertTrue(e.getMessage().contains("Invalid coordinator job change value RUNNING")); + } + } + + // Check status change - with multiple option. Pause can't be applied to killed job, old behavior. + public void testCoord_throwException() throws Exception { + Date start = new Date(); + Date end = new Date(start.getTime() + (4 * 60 * 60 * 1000)); // 4 hrs + String status = "status=RUNNING;pausetime=" + DateUtils.formatDateOozieTZ(end); + final CoordinatorJobBean job = addRecordToCoordJobTableForPauseTimeTest(CoordinatorJob.Status.KILLED, start, + end, end, true, false, 4); + addRecordToCoordActionTable(job.getId(), 1, CoordinatorAction.Status.SUCCEEDED, "coord-action-get.xml", 0); + addRecordToCoordActionTable(job.getId(), 2, CoordinatorAction.Status.SUCCEEDED, "coord-action-get.xml", 0); + addRecordToCoordActionTable(job.getId(), 3, CoordinatorAction.Status.RUNNING, "coord-action-get.xml", 0); + addRecordToCoordActionTable(job.getId(), 4, CoordinatorAction.Status.RUNNING, "coord-action-get.xml", 0); + + try { + new CoordChangeXCommand(job.getId(), status).call(); + fail("should throw Exception"); + } + catch (CommandException e) { + assertTrue(e.getMessage().contains("Cannot change a killed coordinator job")); + } + } + protected CoordinatorJobBean addRecordToCoordJobTableForPauseTimeTest(CoordinatorJob.Status status, Date start, Date end, Date lastActionTime, boolean pending, boolean doneMatd, int lastActionNum) throws Exception { CoordinatorJobBean coordJob = createCoordJob(status, start, end, pending, doneMatd, lastActionNum); http://git-wip-us.apache.org/repos/asf/oozie/blob/3de3c523/core/src/test/java/org/apache/oozie/command/coord/TestCoordRerunXCommand.java ---------------------------------------------------------------------- diff --git a/core/src/test/java/org/apache/oozie/command/coord/TestCoordRerunXCommand.java b/core/src/test/java/org/apache/oozie/command/coord/TestCoordRerunXCommand.java index 3cee71a..92b2190 100644 --- a/core/src/test/java/org/apache/oozie/command/coord/TestCoordRerunXCommand.java +++ b/core/src/test/java/org/apache/oozie/command/coord/TestCoordRerunXCommand.java @@ -821,6 +821,43 @@ public class TestCoordRerunXCommand extends XDataTestCase { store2.closeTrx(); } + /* + * Test rerun of failed action + */ + public void testCoordRerunFailedCoordAction() throws Exception { + final String jobId = "0000000-" + new Date().getTime() + "-testCoordRerun-C"; + final int actionNum = 1; + final String actionId = jobId + "@" + actionNum; + CoordinatorStore store = Services.get().get(StoreService.class).getStore(CoordinatorStore.class); + store.beginTrx(); + try { + addRecordToJobTable(jobId, store, CoordinatorJob.Status.RUNNING); + addRecordToActionTable(jobId, actionNum, actionId, store, CoordinatorAction.Status.FAILED, + "coord-rerun-action1.xml"); + store.commitTrx(); + } + catch (Exception e) { + e.printStackTrace(); + fail("Could not update db."); + } + finally { + store.closeTrx(); + } + + try { + final OozieClient coordClient = LocalOozie.getCoordClient(); + coordClient.reRunCoord(jobId, RestConstants.JOB_COORD_SCOPE_ACTION, Integer.toString(actionNum), false, + true); + } + catch (OozieClientException ex) { + ex.printStackTrace(); + fail("Coord rerun failed"); + } + CoordinatorStore store2 = Services.get().get(StoreService.class).getStore(CoordinatorStore.class); + CoordinatorActionBean action2 = store2.getCoordinatorAction(actionId, false); + assertEquals(action2.getStatus(), CoordinatorAction.Status.WAITING); + } + /** * Tests functionality of the coord rerun for backward support is true. </p> Insert a coordinator job with SUCCEEDED * and coordinator actions with pending false, but one of action is FAILED. http://git-wip-us.apache.org/repos/asf/oozie/blob/3de3c523/docs/src/site/twiki/DG_CommandLineTool.twiki ---------------------------------------------------------------------- diff --git a/docs/src/site/twiki/DG_CommandLineTool.twiki b/docs/src/site/twiki/DG_CommandLineTool.twiki index ee4c7d9..351f0f2 100644 --- a/docs/src/site/twiki/DG_CommandLineTool.twiki +++ b/docs/src/site/twiki/DG_CommandLineTool.twiki @@ -321,23 +321,29 @@ $oozie job -kill <coord_Job_id> [-action 1, 3-4, 7-40] [-date 2009-01-01T01:00Z: * Single date specified in -date must be able to find an action with matched nominal time to be effective. * After the command is executed the killed coordiator action will have =KILLED= status. ----+++ Changing endtime/concurrency/pausetime of a Coordinator Job +---+++ Changing endtime/concurrency/pausetime/status of a Coordinator Job Example: <verbatim> $ oozie job -oozie http://localhost:11000/oozie -change 14-20090525161321-oozie-joe -value endtime=2011-12-01T05:00Z\;concurrency=100\;2011-10-01T05:00Z +$ oozie job -oozie http://localhost:11000/oozie -change 0000001-140321155112907-oozie-puru-C -value status=RUNNING </verbatim> -The =change= option changes a coordinator job that is not in =KILLED= status. +The =endtime/concurrency/pausetime= option changes a coordinator job that is not in =KILLED= status. Valid value names are: * endtime: the end time of the coordinator job. * concurrency: the concurrency of the coordinator job. * pausetime: the pause time of the coordinator job. + * status: new status for coordinator job. + +Repeated value names are not allowed. New end time must not be before job's start time and last action time. Currently status only +takes RUNNING and can be used to change the status of FAILED or KILLED coordinator job to RUNNING and resuming materialization. +This status change command does not affect the status of already materialized actions in the coordinator. If there are FAILED or +KILLED coordinator actions they have to be rerun separately. -Repeated value names are not allowed. New end time must not be before job's start time and last action time. New concurrency value has to be a valid integer. All lookahead actions will be revoked according to the new pause time. Also empty string "" can be used to reset pause time to none. After the command is executed the job's end time, concurrency or pause time should be changed. If an already-succeeded job changes its end time, its status will become running. http://git-wip-us.apache.org/repos/asf/oozie/blob/3de3c523/release-log.txt ---------------------------------------------------------------------- diff --git a/release-log.txt b/release-log.txt index 3e5279b..db6017a 100644 --- a/release-log.txt +++ b/release-log.txt @@ -1,5 +1,6 @@ -- Oozie 4.1.0 release (trunk - unreleased) +OOZIE-1735 Support resuming of failed coordinator job and rerun of a failed coordinator action (puru via rohini) OOZIE-1746 New API to fetch workflows corresponding to coordinator action reruns (mona) OOZIE-1713 Avoid creating dummy input file for each launcher job (puru via rohini) OOZIE-1701 TestXTestCase#testWaitFor and TestXTestCase#testBaseDir have the errors (omaliuvanchuk via rohini)
