| /** |
| * 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.bundle; |
| |
| import java.util.Date; |
| |
| import org.apache.oozie.BundleActionBean; |
| import org.apache.oozie.BundleJobBean; |
| import org.apache.oozie.CoordinatorJobBean; |
| import org.apache.oozie.ErrorCode; |
| 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.executor.jpa.BundleActionInsertJPAExecutor; |
| import org.apache.oozie.executor.jpa.BundleActionQueryExecutor; |
| import org.apache.oozie.executor.jpa.BundleActionQueryExecutor.BundleActionQuery; |
| import org.apache.oozie.executor.jpa.BundleJobGetJPAExecutor; |
| import org.apache.oozie.executor.jpa.BundleJobQueryExecutor; |
| import org.apache.oozie.executor.jpa.BundleJobQueryExecutor.BundleJobQuery; |
| import org.apache.oozie.executor.jpa.CoordJobGetJPAExecutor; |
| import org.apache.oozie.executor.jpa.CoordJobQueryExecutor; |
| import org.apache.oozie.executor.jpa.CoordJobQueryExecutor.CoordJobQuery; |
| import org.apache.oozie.service.JPAService; |
| import org.apache.oozie.service.PauseTransitService.PauseTransitRunnable; |
| import org.apache.oozie.service.SchedulerService; |
| import org.apache.oozie.service.Services; |
| import org.apache.oozie.test.XDataTestCase; |
| import org.apache.oozie.util.DateUtils; |
| |
| public class TestBundleChangeXCommand extends XDataTestCase { |
| private Services services; |
| |
| @Override |
| protected void setUp() throws Exception { |
| super.setUp(); |
| services = new Services(); |
| services.init(); |
| services.get(SchedulerService.class).destroy(); |
| } |
| |
| @Override |
| protected void tearDown() throws Exception { |
| services.destroy(); |
| super.tearDown(); |
| } |
| |
| /** |
| * Test : Change pause time of a bundle |
| * |
| * @throws Exception |
| */ |
| public void testBundleChange1() throws Exception { |
| BundleJobBean job = this.addRecordToBundleJobTable(Job.Status.PREP, false); |
| String dateStr = "2099-01-01T01:00Z"; |
| |
| JPAService jpaService = Services.get().get(JPAService.class); |
| assertNotNull(jpaService); |
| BundleJobGetJPAExecutor bundleJobGetCmd = new BundleJobGetJPAExecutor(job.getId()); |
| job = jpaService.execute(bundleJobGetCmd); |
| assertEquals(job.getPauseTime(), null); |
| |
| new BundleJobChangeXCommand(job.getId(), "pausetime=" + dateStr).call(); |
| |
| job = jpaService.execute(bundleJobGetCmd); |
| assertEquals(job.getPauseTime(), DateUtils.parseDateOozieTZ(dateStr)); |
| } |
| |
| /** |
| * Test : Change pause time of a bundle that contains a SUCCEEDED coordinator job, |
| * The coordinator should also change its pause time. |
| * |
| * @throws Exception |
| */ |
| public void testBundleChange2() throws Exception { |
| BundleJobBean bundleJob = this.addRecordToBundleJobTable(Job.Status.RUNNING, false); |
| |
| CoordinatorJobBean coordJob = addRecordToCoordJobTable(CoordinatorJob.Status.SUCCEEDED, false, false); |
| coordJob.setBundleId(bundleJob.getId()); |
| final JPAService jpaService = Services.get().get(JPAService.class); |
| assertNotNull(jpaService); |
| CoordJobQueryExecutor.getInstance().executeUpdate(CoordJobQuery.UPDATE_COORD_JOB_BUNDLEID, coordJob); |
| |
| coordJob.setAppName("COORD-TEST1"); |
| assertNotNull(jpaService); |
| CoordJobQueryExecutor.getInstance().executeUpdate(CoordJobQuery.UPDATE_COORD_JOB, coordJob); |
| |
| |
| BundleActionBean bundleAction = new BundleActionBean(); |
| bundleAction.setBundleActionId(bundleJob.getId() + "_COORD-TEST1"); |
| bundleAction.setCoordId(coordJob.getId()); |
| bundleAction.setBundleId(bundleJob.getId()); |
| bundleAction.setStatus(Job.Status.SUCCEEDED); |
| jpaService.execute(new BundleActionInsertJPAExecutor(bundleAction)); |
| |
| String dateStr = "2099-01-01T01:00Z"; |
| BundleJobGetJPAExecutor bundleJobGetCmd = new BundleJobGetJPAExecutor(bundleJob.getId()); |
| bundleJob = jpaService.execute(bundleJobGetCmd); |
| assertEquals(bundleJob.getPauseTime(), null); |
| |
| new BundleJobChangeXCommand(bundleJob.getId(), "pausetime=" + dateStr).call(); |
| bundleJob = jpaService.execute(bundleJobGetCmd); |
| assertEquals(DateUtils.parseDateOozieTZ(dateStr), bundleJob.getPauseTime()); |
| |
| final String coordJobId = coordJob.getId(); |
| waitFor(60000, new Predicate() { |
| public boolean evaluate() throws Exception { |
| CoordinatorJobBean coordJob1 = jpaService.execute(new CoordJobGetJPAExecutor(coordJobId)); |
| return (coordJob1.getPauseTime() != null); |
| } |
| }); |
| |
| coordJob = jpaService.execute(new CoordJobGetJPAExecutor(coordJob.getId())); |
| assertEquals(DateUtils.parseDateOozieTZ(dateStr), coordJob.getPauseTime()); |
| } |
| |
| /** |
| * Negative Test : pause time is not a valid date |
| * |
| * @throws Exception |
| */ |
| public void testBundleChangeNegative1() throws Exception { |
| BundleJobBean job = this.addRecordToBundleJobTable(Job.Status.PREP, false); |
| String dateStr = "2099-01-01Ta1:00Z"; |
| |
| JPAService jpaService = Services.get().get(JPAService.class); |
| assertNotNull(jpaService); |
| BundleJobGetJPAExecutor bundleJobGetCmd = new BundleJobGetJPAExecutor(job.getId()); |
| job = jpaService.execute(bundleJobGetCmd); |
| assertEquals(job.getPauseTime(), null); |
| |
| try { |
| new BundleJobChangeXCommand(job.getId(), "pausetime=" + dateStr).call(); |
| fail("Should not reach here"); |
| } |
| catch (XException e) { |
| assertEquals(ErrorCode.E1317, e.getErrorCode()); |
| } |
| } |
| |
| /** |
| * Negative Test : pause time is a past time |
| * |
| * @throws Exception |
| */ |
| public void testBundleChangeNegative2() throws Exception { |
| BundleJobBean job = this.addRecordToBundleJobTable(Job.Status.PREP, false); |
| String dateStr = "2009-01-01T01:00Z"; |
| |
| JPAService jpaService = Services.get().get(JPAService.class); |
| assertNotNull(jpaService); |
| BundleJobGetJPAExecutor bundleJobGetCmd = new BundleJobGetJPAExecutor(job.getId()); |
| job = jpaService.execute(bundleJobGetCmd); |
| assertEquals(job.getPauseTime(), null); |
| |
| try { |
| new BundleJobChangeXCommand(job.getId(), "pausetime=" + dateStr).call(); |
| fail("Should not reach here"); |
| } |
| catch (XException e) { |
| assertEquals(ErrorCode.E1317, e.getErrorCode()); |
| } |
| } |
| |
| /** |
| * Test : Change end time of a bundle |
| * |
| * @throws Exception |
| */ |
| public void testBundleChange3() throws Exception { |
| BundleJobBean job = this.addRecordToBundleJobTable(Job.Status.PREP, false); |
| String dateStr = "2099-01-01T01:00Z"; |
| |
| JPAService jpaService = Services.get().get(JPAService.class); |
| assertNotNull(jpaService); |
| BundleJobGetJPAExecutor bundleJobGetCmd = new BundleJobGetJPAExecutor(job.getId()); |
| job = jpaService.execute(bundleJobGetCmd); |
| assertEquals(job.getEndTime(), null); |
| |
| new BundleJobChangeXCommand(job.getId(), "endtime=" + dateStr).call(); |
| |
| job = jpaService.execute(bundleJobGetCmd); |
| assertEquals(job.getEndTime(), DateUtils.parseDateOozieTZ(dateStr)); |
| } |
| |
| public void testBundlePauseExtendMaterializesCoordinator() throws Exception { |
| BundleJobBean bundle = this.addRecordToBundleJobTable(Job.Status.PAUSED, false); |
| Date startTime = new Date(); |
| Date endTime = new Date(startTime.getTime() + (20 * 60 * 1000)); |
| // coord job with num actions materialized = 1 |
| CoordinatorJobBean coord = addRecordToCoordJobTable(CoordinatorJob.Status.PAUSED, startTime, endTime, true, |
| false, 1); |
| coord.setPauseTime(startTime); // setting dummy old pause time |
| CoordJobQueryExecutor.getInstance().executeUpdate(CoordJobQuery.UPDATE_COORD_JOB_CHANGE, coord); |
| // persist corresponding bundle action bean |
| BundleActionBean bundleAction = this.addRecordToBundleActionTable(bundle.getId(), "COORD-TEST", 1, |
| Job.Status.PAUSED); |
| |
| coord.setBundleId(bundle.getId()); |
| CoordJobQueryExecutor.getInstance().executeUpdate(CoordJobQuery.UPDATE_COORD_JOB_BUNDLEID, coord); |
| bundleAction.setCoordId(coord.getId()); |
| BundleActionQueryExecutor.getInstance().executeUpdate( |
| BundleActionQuery.UPDATE_BUNDLE_ACTION_STATUS_PENDING_MODTIME_COORDID, bundleAction); |
| |
| Date later = new Date(System.currentTimeMillis() + 20 * 1000); // 20 sec later |
| Date evenLater = new Date(later.getTime() + 10 * 60 * 1000); // 10 min later |
| |
| bundle = BundleJobQueryExecutor.getInstance().get(BundleJobQuery.GET_BUNDLE_JOB, bundle.getId()); |
| assertEquals(Job.Status.PAUSED, bundle.getStatus()); |
| coord = CoordJobQueryExecutor.getInstance().get(CoordJobQuery.GET_COORD_JOB, coord.getId()); |
| assertEquals(Job.Status.PAUSED, coord.getStatus()); |
| |
| // coordinator has previous actions |
| coord.setNextMaterializedTime(new Date(startTime.getTime() + (300 * 1000))); // before the new pausetime |
| CoordJobQueryExecutor.getInstance().executeUpdate(CoordJobQuery.UPDATE_COORD_JOB, coord); |
| Date lastMod = coord.getLastModifiedTime(); |
| addRecordToCoordActionTable(coord.getId(), 1, CoordinatorAction.Status.SUCCEEDED, "coord-action-get.xml", 0); |
| |
| // change pausetime to even later |
| new BundleJobChangeXCommand(bundle.getId(), "pausetime=" + DateUtils.formatDateOozieTZ(evenLater)).call(); |
| sleep(1000); // time for the queued CoordChangeXCommand to complete |
| |
| Runnable runnable = new PauseTransitRunnable(); |
| runnable.run(); // simulating run which would happen usually after interval |
| Thread.sleep(1000); |
| |
| bundle = BundleJobQueryExecutor.getInstance().get(BundleJobQuery.GET_BUNDLE_JOB, bundle.getId()); |
| assertEquals(Job.Status.RUNNING, bundle.getStatus()); |
| assertEquals(DateUtils.formatDateOozieTZ(evenLater), DateUtils.formatDateOozieTZ(bundle.getPauseTime())); |
| coord = CoordJobQueryExecutor.getInstance().get(CoordJobQuery.GET_COORD_JOB, coord.getId()); |
| assertEquals(Job.Status.RUNNING, coord.getStatus()); |
| assertEquals(DateUtils.formatDateOozieTZ(evenLater), DateUtils.formatDateOozieTZ(coord.getPauseTime())); |
| assertTrue(coord.getLastModifiedTime().after(lastMod)); |
| |
| } |
| |
| //check command report |
| public void testBundleChangeReport() throws Exception { |
| BundleJobBean bundleJob = this.addRecordToBundleJobTable(Job.Status.RUNNING, false); |
| CoordinatorJobBean coordJob1 = addRecordToCoordJobTable(CoordinatorJob.Status.SUCCEEDED, false, false); |
| coordJob1.setBundleId(bundleJob.getId()); |
| coordJob1.setAppName("COORD-TEST1"); |
| final JPAService jpaService = Services.get().get(JPAService.class); |
| assertNotNull(jpaService); |
| CoordJobQueryExecutor.getInstance().executeUpdate(CoordJobQuery.UPDATE_COORD_JOB, coordJob1); |
| BundleActionBean bundleAction1 = new BundleActionBean(); |
| bundleAction1.setBundleActionId(bundleJob.getId() + "_COORD-TEST1"); |
| bundleAction1.setCoordId(coordJob1.getId()); |
| bundleAction1.setBundleId(bundleJob.getId()); |
| bundleAction1.setStatus(Job.Status.SUCCEEDED); |
| jpaService.execute(new BundleActionInsertJPAExecutor(bundleAction1)); |
| |
| CoordinatorJobBean coordJob2 = addRecordToCoordJobTable(CoordinatorJob.Status.KILLED, false, false); |
| coordJob2.setBundleId(bundleJob.getId()); |
| coordJob2.setAppName("COORD-TEST2"); |
| CoordJobQueryExecutor.getInstance().executeUpdate(CoordJobQuery.UPDATE_COORD_JOB, coordJob2); |
| BundleActionBean bundleAction2 = new BundleActionBean(); |
| bundleAction2.setBundleActionId(bundleJob.getId() + "_COORD-TEST2"); |
| bundleAction2.setCoordId(coordJob2.getId()); |
| bundleAction2.setBundleId(bundleJob.getId()); |
| bundleAction2.setStatus(Job.Status.KILLED); |
| jpaService.execute(new BundleActionInsertJPAExecutor(bundleAction2)); |
| |
| CoordinatorJobBean coordJob3 = addRecordToCoordJobTable(CoordinatorJob.Status.RUNNING, false, false); |
| addRecordToCoordActionTable(coordJob3.getId(), 1, CoordinatorAction.Status.SUCCEEDED, "coord-action-get.xml", 0); |
| addRecordToCoordActionTable(coordJob3.getId(), 2, CoordinatorAction.Status.SUCCEEDED, "coord-action-get.xml", |
| 0, DateUtils.parseDateOozieTZ("2013-08-01T02:00Z")); |
| addRecordToCoordActionTable(coordJob3.getId(), 3, CoordinatorAction.Status.RUNNING, "coord-action-get.xml", 0, |
| DateUtils.parseDateOozieTZ("2013-08-01T03:00Z")); |
| addRecordToCoordActionTable(coordJob3.getId(), 4, CoordinatorAction.Status.WAITING, "coord-action-get.xml", 0, |
| DateUtils.parseDateOozieTZ("2013-08-01T04:00Z")); |
| |
| coordJob3.setBundleId(bundleJob.getId()); |
| coordJob3.setAppName("COORD-TEST3"); |
| coordJob3.setLastActionNumber(4); |
| coordJob3.setEndTime(DateUtils.parseDateOozieTZ("2013-08-01T04:00Z")); |
| coordJob3.setStartTime(DateUtils.parseDateOozieTZ("2013-08-01T00:00Z")); |
| |
| CoordJobQueryExecutor.getInstance().executeUpdate(CoordJobQuery.UPDATE_COORD_JOB, coordJob3); |
| BundleActionBean bundleAction3 = new BundleActionBean(); |
| bundleAction3.setBundleActionId(bundleJob.getId() + "_COORD-TEST3"); |
| bundleAction3.setCoordId(coordJob3.getId()); |
| bundleAction3.setBundleId(bundleJob.getId()); |
| bundleAction3.setStatus(Job.Status.RUNNING); |
| jpaService.execute(new BundleActionInsertJPAExecutor(bundleAction3)); |
| BundleJobGetJPAExecutor bundleJobGetCmd = new BundleJobGetJPAExecutor(bundleJob.getId()); |
| |
| String dateStr = "2099-01-01T01:00Z"; |
| bundleJob = jpaService.execute(bundleJobGetCmd); |
| assertEquals(bundleJob.getPauseTime(), null); |
| String reports = null; |
| try { |
| new BundleJobChangeXCommand(bundleJob.getId(), "pausetime=" + dateStr).call(); |
| } |
| catch (Exception e) { |
| reports = e.getMessage(); |
| } |
| assertTrue(reports.contains(coordJob2.getId() + " : Coord is in killed state")); |
| |
| bundleJobGetCmd = new BundleJobGetJPAExecutor(bundleJob.getId()); |
| dateStr = "2013-08-01T03:00Z"; |
| |
| bundleJob = jpaService.execute(bundleJobGetCmd); |
| try { |
| new BundleJobChangeXCommand(bundleJob.getId(), "endtime=" + dateStr).call(); |
| } |
| catch (Exception e) { |
| reports = e.getMessage(); |
| } |
| assertTrue(reports.contains(coordJob2.getId() + " : Coord is in killed state")); |
| assertTrue(reports.contains(coordJob3.getId() + " : E1022: Cannot delete running/completed coordinator action")); |
| } |
| |
| // Check partial update. |
| // 3 coord job with status SUCCEEDED, PREP, RUNNING. |
| // Changing endtime of running coord will fail because end time is before one of running action nominal time. |
| public void testCheckBundleActionStatus() throws Exception { |
| BundleJobBean bundleJob = this.addRecordToBundleJobTable(Job.Status.RUNNING, false); |
| CoordinatorJobBean coordJob1 = addRecordToCoordJobTable(CoordinatorJob.Status.SUCCEEDED, false, false); |
| coordJob1.setBundleId(bundleJob.getId()); |
| coordJob1.setAppName("COORD-TEST1"); |
| coordJob1.setEndTime(DateUtils.parseDateOozieTZ("2013-08-01T02:00Z")); |
| coordJob1.setStartTime(DateUtils.parseDateOozieTZ("2013-08-01T00:00Z")); |
| |
| final JPAService jpaService = Services.get().get(JPAService.class); |
| assertNotNull(jpaService); |
| CoordJobQueryExecutor.getInstance().executeUpdate(CoordJobQuery.UPDATE_COORD_JOB, coordJob1); |
| BundleActionBean bundleAction1 = new BundleActionBean(); |
| bundleAction1.setBundleActionId(bundleJob.getId() + "_COORD-TEST1"); |
| bundleAction1.setCoordId(coordJob1.getId()); |
| bundleAction1.setBundleId(bundleJob.getId()); |
| bundleAction1.setStatus(Job.Status.SUCCEEDED); |
| jpaService.execute(new BundleActionInsertJPAExecutor(bundleAction1)); |
| |
| CoordinatorJobBean coordJob2 = addRecordToCoordJobTable(CoordinatorJob.Status.PREP, false, false); |
| coordJob2.setBundleId(bundleJob.getId()); |
| coordJob2.setAppName("COORD-TEST2"); |
| coordJob2.setStartTime(DateUtils.parseDateOozieTZ("2099-08-01T02:00Z")); |
| CoordJobQueryExecutor.getInstance().executeUpdate(CoordJobQuery.UPDATE_COORD_JOB, coordJob2); |
| BundleActionBean bundleAction2 = new BundleActionBean(); |
| bundleAction2.setBundleActionId(bundleJob.getId() + "_COORD-TEST2"); |
| bundleAction2.setCoordId(coordJob2.getId()); |
| bundleAction2.setBundleId(bundleJob.getId()); |
| bundleAction2.setStatus(Job.Status.PREP); |
| jpaService.execute(new BundleActionInsertJPAExecutor(bundleAction2)); |
| |
| CoordinatorJobBean coordJob3 = addRecordToCoordJobTable(CoordinatorJob.Status.RUNNING, false, false); |
| addRecordToCoordActionTable(coordJob3.getId(), 1, CoordinatorAction.Status.SUCCEEDED, "coord-action-get.xml", 0); |
| addRecordToCoordActionTable(coordJob3.getId(), 2, CoordinatorAction.Status.SUCCEEDED, "coord-action-get.xml", |
| 0, DateUtils.parseDateOozieTZ("2013-08-01T02:00Z")); |
| addRecordToCoordActionTable(coordJob3.getId(), 3, CoordinatorAction.Status.RUNNING, "coord-action-get.xml", 0, |
| DateUtils.parseDateOozieTZ("2013-08-01T03:00Z")); |
| addRecordToCoordActionTable(coordJob3.getId(), 4, CoordinatorAction.Status.RUNNING, "coord-action-get.xml", 0, |
| DateUtils.parseDateOozieTZ("2013-08-01T04:00Z")); |
| |
| coordJob3.setBundleId(bundleJob.getId()); |
| coordJob3.setAppName("COORD-TEST3"); |
| coordJob3.setLastActionNumber(4); |
| coordJob3.setEndTime(DateUtils.parseDateOozieTZ("2013-08-01T04:00Z")); |
| coordJob3.setStartTime(DateUtils.parseDateOozieTZ("2013-08-01T00:00Z")); |
| |
| CoordJobQueryExecutor.getInstance().executeUpdate(CoordJobQuery.UPDATE_COORD_JOB, coordJob3); |
| BundleActionBean bundleAction3 = new BundleActionBean(); |
| bundleAction3.setBundleActionId(bundleJob.getId() + "_COORD-TEST3"); |
| bundleAction3.setPending(1); |
| bundleAction3.setCoordId(coordJob3.getId()); |
| bundleAction3.setBundleId(bundleJob.getId()); |
| bundleAction3.setStatus(Job.Status.RUNNING); |
| jpaService.execute(new BundleActionInsertJPAExecutor(bundleAction3)); |
| |
| String dateStr = "2013-08-01T03:00Z"; |
| try { |
| new BundleJobChangeXCommand(bundleJob.getId(), "endtime=" + dateStr).call(); |
| fail("should throw exception"); |
| } |
| catch (Exception e) { |
| String reports = e.getMessage(); |
| assertTrue(reports.contains(coordJob3.getId() + " : E1022: Cannot delete running/completed coordinator action")); |
| } |
| BundleActionBean action1 = BundleActionQueryExecutor.getInstance().get(BundleActionQuery.GET_BUNDLE_ACTION, |
| bundleJob.getId() + "_COORD-TEST1"); |
| assertEquals(CoordinatorJob.Status.RUNNING, action1.getStatus()); |
| BundleActionBean action2 = BundleActionQueryExecutor.getInstance().get(BundleActionQuery.GET_BUNDLE_ACTION, |
| bundleJob.getId() + "_COORD-TEST2"); |
| assertEquals(CoordinatorJob.Status.SUCCEEDED, action2.getStatus()); |
| BundleActionBean action3 = BundleActionQueryExecutor.getInstance().get(BundleActionQuery.GET_BUNDLE_ACTION, |
| bundleJob.getId() + "_COORD-TEST3"); |
| //No change in running |
| assertEquals(CoordinatorJob.Status.RUNNING, action3.getStatus()); |
| assertEquals(action3.getPending(), 1); |
| } |
| |
| } |
