| /** |
| * 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 java.io.IOException; |
| import java.util.Arrays; |
| import java.util.Date; |
| import java.util.List; |
| |
| import org.apache.hadoop.fs.Path; |
| import org.apache.oozie.CoordinatorActionBean; |
| import org.apache.oozie.CoordinatorJobBean; |
| import org.apache.oozie.client.CoordinatorAction; |
| import org.apache.oozie.client.CoordinatorJob; |
| import org.apache.oozie.client.OozieClient; |
| import org.apache.oozie.client.CoordinatorJob.Execution; |
| import org.apache.oozie.client.CoordinatorJob.Timeunit; |
| import org.apache.oozie.command.CommandException; |
| import org.apache.oozie.coord.CoordELFunctions; |
| import org.apache.oozie.coord.input.dependency.CoordOldInputDependency; |
| import org.apache.oozie.executor.jpa.CoordActionGetForInputCheckJPAExecutor; |
| import org.apache.oozie.executor.jpa.CoordActionGetJPAExecutor; |
| import org.apache.oozie.executor.jpa.CoordActionInsertJPAExecutor; |
| import org.apache.oozie.executor.jpa.CoordActionQueryExecutor; |
| import org.apache.oozie.executor.jpa.CoordActionQueryExecutor.CoordActionQuery; |
| import org.apache.oozie.executor.jpa.CoordJobInsertJPAExecutor; |
| import org.apache.oozie.executor.jpa.CoordJobQueryExecutor; |
| import org.apache.oozie.executor.jpa.JPAExecutorException; |
| import org.apache.oozie.service.CallableQueueService; |
| import org.apache.oozie.service.HadoopAccessorService; |
| import org.apache.oozie.service.JPAService; |
| import org.apache.oozie.service.Services; |
| import org.apache.oozie.service.XLogService; |
| import org.apache.oozie.test.XDataTestCase; |
| import org.apache.oozie.util.DateUtils; |
| import org.apache.oozie.util.IOUtils; |
| import org.apache.oozie.util.XConfiguration; |
| import org.apache.oozie.util.XLog; |
| import org.apache.oozie.util.XmlUtils; |
| import org.jdom.Element; |
| import org.junit.Test; |
| |
| public class TestCoordActionInputCheckXCommand extends XDataTestCase { |
| protected Services services; |
| |
| protected String getProcessingTZ() { |
| return DateUtils.OOZIE_PROCESSING_TIMEZONE_DEFAULT; |
| } |
| |
| private String TZ; |
| |
| @Override |
| protected void setUp() throws Exception { |
| super.setUp(); |
| setSystemProperty(XLogService.LOG4J_FILE, "oozie-log4j.properties"); |
| setSystemProperty(DateUtils.OOZIE_PROCESSING_TIMEZONE_KEY, getProcessingTZ()); |
| services = new Services(); |
| services.init(); |
| TZ = (getProcessingTZ().equals(DateUtils.OOZIE_PROCESSING_TIMEZONE_DEFAULT)) |
| ? "Z" : getProcessingTZ().substring(3); |
| } |
| |
| @Override |
| protected void tearDown() throws Exception { |
| services.destroy(); |
| super.tearDown(); |
| } |
| |
| public class MyCoordActionInputCheckXCommand extends CoordActionInputCheckXCommand { |
| long executed = 0; |
| int wait; |
| |
| public MyCoordActionInputCheckXCommand(String actionId, int wait, String entityKey) { |
| super(actionId, entityKey); |
| this.wait = wait; |
| } |
| |
| @Override |
| public String toString() { |
| StringBuilder sb = new StringBuilder(); |
| sb.append("Type:").append(getType()); |
| sb.append(",Priority:").append(getPriority()); |
| return sb.toString(); |
| } |
| |
| @Override |
| protected Void execute() throws CommandException { |
| try { |
| Thread.sleep(wait); |
| } |
| catch (InterruptedException e) { |
| } |
| executed = System.currentTimeMillis(); |
| return null; |
| } |
| |
| } |
| |
| public void testCoordActionInputCheckXCommandUniqueness() throws Exception { |
| Date startTime = DateUtils.parseDateOozieTZ("2009-02-01T23:59" + TZ); |
| Date endTime = DateUtils.parseDateOozieTZ("2009-02-02T23:59" + TZ); |
| CoordinatorJobBean job = addRecordToCoordJobTableForWaiting("coord-job-for-action-input-check.xml", |
| CoordinatorJob.Status.RUNNING, startTime, endTime, false, true, 3); |
| |
| CoordinatorActionBean action1 = addRecordToCoordActionTableForWaiting(job.getId(), 1, |
| CoordinatorAction.Status.WAITING, "coord-action-for-action-input-check.xml"); |
| |
| createTestCaseSubDir("2009/01/29/_SUCCESS".split("/")); |
| createTestCaseSubDir("2009/01/22/_SUCCESS".split("/")); |
| createTestCaseSubDir("2009/01/15/_SUCCESS".split("/")); |
| createTestCaseSubDir("2009/01/08/_SUCCESS".split("/")); |
| |
| final MyCoordActionInputCheckXCommand callable1 = new MyCoordActionInputCheckXCommand(action1.getId(), 100, "1"); |
| final MyCoordActionInputCheckXCommand callable2 = new MyCoordActionInputCheckXCommand(action1.getId(), 100, "2"); |
| final MyCoordActionInputCheckXCommand callable3 = new MyCoordActionInputCheckXCommand(action1.getId(), 100, "3"); |
| |
| List<MyCoordActionInputCheckXCommand> callables = Arrays.asList(callable1, callable2, callable3); |
| |
| CallableQueueService queueservice = services.get(CallableQueueService.class); |
| |
| for (MyCoordActionInputCheckXCommand c : callables) { |
| queueservice.queue(c); |
| } |
| |
| waitFor(200, new Predicate() { |
| public boolean evaluate() throws Exception { |
| return callable1.executed != 0 && callable2.executed == 0 && callable3.executed == 0; |
| } |
| }); |
| |
| assertTrue(callable1.executed != 0); |
| assertTrue(callable2.executed == 0); |
| assertTrue(callable3.executed == 0); |
| } |
| |
| public void testActionInputCheck() throws Exception { |
| String jobId = "0000000-" + new Date().getTime() + "-TestCoordActionInputCheckXCommand-C"; |
| Date startTime = DateUtils.parseDateOozieTZ("2009-02-01T23:59" + TZ); |
| Date endTime = DateUtils.parseDateOozieTZ("2009-02-02T23:59" + TZ); |
| CoordinatorJobBean job = addRecordToCoordJobTable(jobId, startTime, endTime); |
| new CoordMaterializeTransitionXCommand(job.getId(), 3600).call(); |
| createTestCaseSubDir("2009/02/05/_SUCCESS".split("/")); |
| createTestCaseSubDir("2009/01/15/_SUCCESS".split("/")); |
| new CoordActionInputCheckXCommand(job.getId() + "@1", job.getId()).call(); |
| JPAService jpaService = Services.get().get(JPAService.class); |
| CoordinatorActionBean action = jpaService.execute(new CoordActionGetJPAExecutor(job.getId() + "@1")); |
| System.out.println("missingDeps " + action.getMissingDependencies() + " Xml " + action.getActionXml()); |
| if (action.getMissingDependencies().indexOf("/2009/02/05/") >= 0) { |
| fail("directory should be resolved :" + action.getMissingDependencies()); |
| } |
| if (action.getMissingDependencies().indexOf("/2009/01/15/") < 0) { |
| fail("directory should NOT be resolved :" + action.getMissingDependencies()); |
| } |
| } |
| |
| /** |
| * Test to check that missing dependencies list starts from the first unavailable dependency in synchronous order |
| * @throws Exception |
| */ |
| public void testActionInputMissingDependencies() throws Exception { |
| String jobId = "0000000-" + new Date().getTime() + "-TestCoordActionInputCheckXCommand-C"; |
| Date startTime = DateUtils.parseDateOozieTZ("2009-02-15T23:59" + TZ); |
| Date endTime = DateUtils.parseDateOozieTZ("2009-02-16T23:59" + TZ); |
| CoordinatorJobBean job = addRecordToCoordJobTable(jobId, startTime, endTime); |
| new CoordMaterializeTransitionXCommand(job.getId(), 3600).call(); |
| |
| // providing some of the dataset dirs required as per coordinator |
| // specification - /2009/02/19, /2009/02/12, /2009/02/05, /2009/01/29, /2009/01/22 |
| createTestCaseSubDir("2009/02/19/_SUCCESS".split("/")); |
| createTestCaseSubDir("2009/01/29/_SUCCESS".split("/")); |
| |
| new CoordActionInputCheckXCommand(job.getId() + "@1", job.getId()).call(); |
| CoordinatorActionBean action = null; |
| try { |
| JPAService jpaService = Services.get().get(JPAService.class); |
| action = jpaService.execute(new CoordActionGetJPAExecutor(job.getId() + "@1")); |
| } |
| catch (JPAExecutorException se) { |
| fail("Action ID " + job.getId() + "@1" + " was not stored properly in db"); |
| } |
| |
| // Missing dependencies recorded by the coordinator action after input check |
| String missDepsOrder = action.getMissingDependencies(); |
| // Expected missing dependencies are /2009/02/12, /2009/02/05, /2009/01/29, and /2009/01/22. |
| |
| int index = missDepsOrder.indexOf("/2009/02/19"); |
| if( index >= 0) { |
| fail("Dependency should be available! current list: " + missDepsOrder); |
| } |
| // Case when /2009/01/29 exists but checking stops since dataset synchronously expected before i.e. /2009/02/05 is missing |
| index = missDepsOrder.indexOf("/2009/01/29"); |
| if( index < 0) { |
| fail("Data should have been in missing dependency list! current list: " + missDepsOrder); |
| } |
| index = missDepsOrder.indexOf("/2009/02/05"); |
| if( index < 0) { |
| fail("Data should have been in missing dependency list! current list: " + missDepsOrder); |
| } |
| } |
| |
| public void testActionInputCheckLatestActionCreationTime() throws Exception { |
| Services.get().getConf().setBoolean(CoordELFunctions.LATEST_EL_USE_CURRENT_TIME, false); |
| |
| String jobId = "0000000-" + new Date().getTime() + "-TestCoordActionInputCheckXCommand-C"; |
| Date startTime = DateUtils.parseDateOozieTZ("2009-02-15T23:59" + TZ); |
| Date endTime = DateUtils.parseDateOozieTZ("2009-02-16T23:59" + TZ); |
| CoordinatorJobBean job = addRecordToCoordJobTable(jobId, startTime, endTime, "latest"); |
| new CoordMaterializeTransitionXCommand(job.getId(), 3600).call(); |
| |
| JPAService jpaService = Services.get().get(JPAService.class); |
| CoordinatorActionBean action = jpaService |
| .execute(new CoordActionGetForInputCheckJPAExecutor(job.getId() + "@1")); |
| assertEquals(CoordCommandUtils.RESOLVED_UNRESOLVED_SEPARATOR + "${coord:latestRange(-3,0)}", |
| action.getMissingDependencies()); |
| |
| // Update action creation time |
| String actionXML = action.getActionXml(); |
| String actionCreationTime = "2009-02-15T01:00" + TZ; |
| actionXML = actionXML.replaceAll("action-actual-time=\".*\">", "action-actual-time=\"" + actionCreationTime |
| + "\">"); |
| action.setActionXml(actionXML); |
| action.setCreatedTime(DateUtils.parseDateOozieTZ(actionCreationTime)); |
| CoordActionQueryExecutor.getInstance().executeUpdate( |
| CoordActionQueryExecutor.CoordActionQuery.UPDATE_COORD_ACTION_FOR_INPUTCHECK, |
| action); |
| action = jpaService.execute(new CoordActionGetForInputCheckJPAExecutor(job.getId() + "@1")); |
| assertTrue(action.getActionXml().contains("action-actual-time=\"2009-02-15T01:00")) ; |
| |
| Thread.sleep(1000); |
| // providing some of the dataset dirs required as per coordinator specification with holes |
| // before and after action creation time |
| createTestCaseSubDir("2009/03/05/_SUCCESS".split("/")); |
| createTestCaseSubDir("2009/02/19/_SUCCESS".split("/")); |
| createTestCaseSubDir("2009/02/12/_SUCCESS".split("/")); |
| createTestCaseSubDir("2009/02/05/_SUCCESS".split("/")); |
| createTestCaseSubDir("2009/01/22/_SUCCESS".split("/")); |
| createTestCaseSubDir("2009/01/08/_SUCCESS".split("/")); |
| |
| new CoordActionInputCheckXCommand(job.getId() + "@1", job.getId()).call(); |
| //Sleep for sometime as it gets requeued with 10ms delay on failure to acquire write lock |
| Thread.sleep(1000); |
| |
| action = jpaService.execute(new CoordActionGetForInputCheckJPAExecutor(job.getId() + "@1")); |
| actionXML = action.getActionXml(); |
| assertEquals("", action.getMissingDependencies()); |
| // Datasets only before action creation/actual time should be picked up. |
| String resolvedList = getTestCaseFileUri("2009/02/12") + CoordELFunctions.INSTANCE_SEPARATOR |
| + getTestCaseFileUri("2009/02/05") + CoordELFunctions.INSTANCE_SEPARATOR |
| + getTestCaseFileUri("2009/01/22") + CoordELFunctions.INSTANCE_SEPARATOR |
| + getTestCaseFileUri("2009/01/08"); |
| System.out.println("Expected: " + resolvedList); |
| System.out.println("Actual: " + actionXML.substring(actionXML.indexOf("<uris>") + 6, actionXML.indexOf("</uris>"))); |
| assertEquals(resolvedList, actionXML.substring(actionXML.indexOf("<uris>") + 6, actionXML.indexOf("</uris>"))); |
| } |
| |
| public void testActionInputCheckLatestActionCreationTimeWithPushDependency() throws Exception { |
| setupServicesForHCatalog(services); |
| Services.get().getConf().setBoolean(CoordELFunctions.LATEST_EL_USE_CURRENT_TIME, false); |
| services.init(); |
| String jobId = "0000000-" + new Date().getTime() + "-TestCoordActionInputCheckXCommand-C"; |
| Date startTime = DateUtils.parseDateOozieTZ("2009-02-15T23:59" + TZ); |
| Date endTime = DateUtils.parseDateOozieTZ("2009-02-16T23:59" + TZ); |
| CoordinatorJobBean job = addRecordToCoordJobTable(jobId, startTime, endTime, "latest"); |
| new CoordMaterializeTransitionXCommand(job.getId(), 3600).call(); |
| |
| // Set push missing dependency |
| JPAService jpaService = Services.get().get(JPAService.class); |
| CoordinatorActionBean action = jpaService |
| .execute(new CoordActionGetForInputCheckJPAExecutor(job.getId() + "@1")); |
| final String pushMissingDependency = getTestCaseFileUri("2009/02/05"); |
| action.setPushMissingDependencies(pushMissingDependency); |
| CoordActionQueryExecutor.getInstance().executeUpdate( |
| CoordActionQueryExecutor.CoordActionQuery.UPDATE_COORD_ACTION_FOR_PUSH_INPUTCHECK, |
| action); |
| |
| // Update action creation time |
| String actionXML = action.getActionXml(); |
| String actionCreationTime = "2009-02-15T01:00" + TZ; |
| actionXML = actionXML.replaceAll("action-actual-time=\".*\">", "action-actual-time=\"" + actionCreationTime |
| + "\">"); |
| action.setActionXml(actionXML); |
| action.setCreatedTime(DateUtils.parseDateOozieTZ(actionCreationTime)); |
| CoordActionQueryExecutor.getInstance().executeUpdate( |
| CoordActionQueryExecutor.CoordActionQuery.UPDATE_COORD_ACTION_FOR_INPUTCHECK, |
| action); |
| action = jpaService.execute(new CoordActionGetForInputCheckJPAExecutor(job.getId() + "@1")); |
| assertTrue(action.getActionXml().contains("action-actual-time=\"2009-02-15T01:00")) ; |
| |
| new CoordActionInputCheckXCommand(job.getId() + "@1", job.getId()).call(); |
| new CoordPushDependencyCheckXCommand(job.getId() + "@1").call(); |
| action = jpaService.execute(new CoordActionGetForInputCheckJPAExecutor(job.getId() + "@1")); |
| assertEquals(CoordCommandUtils.RESOLVED_UNRESOLVED_SEPARATOR + "${coord:latestRange(-3,0)}", |
| action.getMissingDependencies()); |
| assertEquals(pushMissingDependency, action.getPushMissingDependencies()); |
| |
| // providing some of the dataset dirs required as per coordinator specification with holes |
| // before and after action creation time |
| createTestCaseSubDir("2009/03/05/_SUCCESS".split("/")); |
| createTestCaseSubDir("2009/02/19/_SUCCESS".split("/")); |
| createTestCaseSubDir("2009/02/12/_SUCCESS".split("/")); |
| createTestCaseSubDir("2009/01/22/_SUCCESS".split("/")); |
| createTestCaseSubDir("2009/01/08/_SUCCESS".split("/")); |
| createTestCaseSubDir("2009/01/01/_SUCCESS".split("/")); |
| |
| // Run input check after making latest available |
| new CoordActionInputCheckXCommand(job.getId() + "@1", job.getId()).call(); |
| action = jpaService.execute(new CoordActionGetForInputCheckJPAExecutor(job.getId() + "@1")); |
| assertEquals(CoordCommandUtils.RESOLVED_UNRESOLVED_SEPARATOR + "${coord:latestRange(-3,0)}", |
| action.getMissingDependencies()); |
| assertEquals(pushMissingDependency, action.getPushMissingDependencies()); |
| |
| // Run input check after making push dependencies available |
| createTestCaseSubDir("2009/02/05/_SUCCESS".split("/")); |
| new CoordPushDependencyCheckXCommand(job.getId() + "@1").call(); |
| action = jpaService.execute(new CoordActionGetForInputCheckJPAExecutor(job.getId() + "@1")); |
| assertEquals("", action.getPushMissingDependencies()); |
| checkCoordAction(job.getId() + "@1", CoordCommandUtils.RESOLVED_UNRESOLVED_SEPARATOR |
| + "${coord:latestRange(-3,0)}", CoordinatorAction.Status.WAITING); |
| new CoordActionInputCheckXCommand(job.getId() + "@1", job.getId()).call(); |
| //Sleep for sometime as it gets requeued with 10ms delay on failure to acquire write lock |
| Thread.sleep(1000); |
| |
| action = jpaService.execute(new CoordActionGetForInputCheckJPAExecutor(job.getId() + "@1")); |
| assertEquals("", action.getMissingDependencies()); |
| actionXML = action.getActionXml(); |
| // Datasets only before action creation/actual time should be picked up. |
| String resolvedList = getTestCaseFileUri("2009/02/12") + CoordELFunctions.INSTANCE_SEPARATOR |
| + getTestCaseFileUri("2009/02/05") + CoordELFunctions.INSTANCE_SEPARATOR |
| + getTestCaseFileUri("2009/01/22") + CoordELFunctions.INSTANCE_SEPARATOR |
| + getTestCaseFileUri("2009/01/08"); |
| System.out.println("Expected: " + resolvedList); |
| System.out.println("Actual: " + actionXML.substring(actionXML.indexOf("<uris>") + 6, actionXML.indexOf("</uris>"))); |
| assertEquals(resolvedList, actionXML.substring(actionXML.indexOf("<uris>") + 6, actionXML.indexOf("</uris>"))); |
| } |
| |
| public void testActionInputCheckLatestCurrentTime() throws Exception { |
| Services.get().getConf().setBoolean(CoordELFunctions.LATEST_EL_USE_CURRENT_TIME, true); |
| |
| String jobId = "0000000-" + new Date().getTime() + "-TestCoordActionInputCheckXCommand-C"; |
| Date startTime = DateUtils.parseDateOozieTZ("2009-02-15T23:59" + TZ); |
| Date endTime = DateUtils.parseDateOozieTZ("2009-02-16T23:59" + TZ); |
| CoordinatorJobBean job = addRecordToCoordJobTable(jobId, startTime, endTime, "latest"); |
| new CoordMaterializeTransitionXCommand(job.getId(), 3600).call(); |
| |
| JPAService jpaService = Services.get().get(JPAService.class); |
| CoordinatorActionBean action = jpaService |
| .execute(new CoordActionGetForInputCheckJPAExecutor(job.getId() + "@1")); |
| assertEquals(CoordCommandUtils.RESOLVED_UNRESOLVED_SEPARATOR + "${coord:latestRange(-3,0)}", |
| action.getMissingDependencies()); |
| |
| // Update action creation time |
| String actionXML = action.getActionXml(); |
| String actionCreationTime = "2009-02-15T01:00" + TZ; |
| actionXML = actionXML.replaceAll("action-actual-time=\".*\">", "action-actual-time=\"" + actionCreationTime |
| + "\">"); |
| action.setActionXml(actionXML); |
| action.setCreatedTime(DateUtils.parseDateOozieTZ(actionCreationTime)); |
| CoordActionQueryExecutor.getInstance().executeUpdate( |
| CoordActionQueryExecutor.CoordActionQuery.UPDATE_COORD_ACTION_FOR_INPUTCHECK, |
| action); |
| action = jpaService.execute(new CoordActionGetForInputCheckJPAExecutor(job.getId() + "@1")); |
| assertTrue(action.getActionXml().contains("action-actual-time=\"2009-02-15T01:00")) ; |
| |
| // providing some of the dataset dirs required as per coordinator |
| // specification with holes |
| // before and after action creation time |
| createTestCaseSubDir("2009/03/05/_SUCCESS".split("/")); |
| createTestCaseSubDir("2009/02/19/_SUCCESS".split("/")); |
| createTestCaseSubDir("2009/02/12/_SUCCESS".split("/")); |
| createTestCaseSubDir("2009/02/05/_SUCCESS".split("/")); |
| createTestCaseSubDir("2009/01/22/_SUCCESS".split("/")); |
| createTestCaseSubDir("2009/01/08/_SUCCESS".split("/")); |
| |
| new CoordActionInputCheckXCommand(job.getId() + "@1", job.getId()).call(); |
| //Sleep for sometime as it gets requeued with 10ms delay on failure to acquire write lock |
| Thread.sleep(1000); |
| |
| action = jpaService.execute(new CoordActionGetJPAExecutor(job.getId() + "@1")); |
| actionXML = action.getActionXml(); |
| assertEquals("", action.getMissingDependencies()); |
| // Datasets should be picked up based on current time and not action creation/actual time. |
| String resolvedList = getTestCaseFileUri("2009/03/05") + CoordELFunctions.INSTANCE_SEPARATOR |
| + getTestCaseFileUri("2009/02/19") + CoordELFunctions.INSTANCE_SEPARATOR |
| + getTestCaseFileUri("2009/02/12") + CoordELFunctions.INSTANCE_SEPARATOR |
| + getTestCaseFileUri("2009/02/05"); |
| assertEquals(resolvedList, actionXML.substring(actionXML.indexOf("<uris>") + 6, actionXML.indexOf("</uris>"))); |
| } |
| |
| public void testActionInputCheckLatestCurrentTimeWithPushDependency() throws Exception { |
| setupServicesForHCatalog(services); |
| Services.get().getConf().setBoolean(CoordELFunctions.LATEST_EL_USE_CURRENT_TIME, true); |
| services.init(); |
| |
| String jobId = "0000000-" + new Date().getTime() + "-TestCoordActionInputCheckXCommand-C"; |
| Date startTime = DateUtils.parseDateOozieTZ("2009-02-15T23:59" + TZ); |
| Date endTime = DateUtils.parseDateOozieTZ("2009-02-16T23:59" + TZ); |
| CoordinatorJobBean job = addRecordToCoordJobTable(jobId, startTime, endTime, "latest"); |
| new CoordMaterializeTransitionXCommand(job.getId(), 3600).call(); |
| |
| // Set push missing dependency |
| JPAService jpaService = Services.get().get(JPAService.class); |
| CoordinatorActionBean action = jpaService |
| .execute(new CoordActionGetForInputCheckJPAExecutor(job.getId() + "@1")); |
| final String pushMissingDependency = getTestCaseFileUri("2009/02/05"); |
| action.setPushMissingDependencies(pushMissingDependency); |
| CoordActionQueryExecutor.getInstance().executeUpdate( |
| CoordActionQueryExecutor.CoordActionQuery.UPDATE_COORD_ACTION_FOR_PUSH_INPUTCHECK, |
| action); |
| |
| // Update action creation time |
| String actionXML = action.getActionXml(); |
| String actionCreationTime = "2009-02-15T01:00" + TZ; |
| actionXML = actionXML.replaceAll("action-actual-time=\".*\">", "action-actual-time=\"" + actionCreationTime |
| + "\">"); |
| action.setActionXml(actionXML); |
| action.setCreatedTime(DateUtils.parseDateOozieTZ(actionCreationTime)); |
| CoordActionQueryExecutor.getInstance().executeUpdate( |
| CoordActionQueryExecutor.CoordActionQuery.UPDATE_COORD_ACTION_FOR_INPUTCHECK, |
| action); |
| action = jpaService.execute(new CoordActionGetForInputCheckJPAExecutor(job.getId() + "@1")); |
| assertTrue(action.getActionXml().contains("action-actual-time=\"2009-02-15T01:00")) ; |
| |
| new CoordActionInputCheckXCommand(job.getId() + "@1", job.getId()).call(); |
| new CoordPushDependencyCheckXCommand(job.getId() + "@1").call(); |
| action = jpaService.execute(new CoordActionGetForInputCheckJPAExecutor(job.getId() + "@1")); |
| assertEquals(CoordCommandUtils.RESOLVED_UNRESOLVED_SEPARATOR + "${coord:latestRange(-3,0)}", |
| action.getMissingDependencies()); |
| assertEquals(pushMissingDependency, action.getPushMissingDependencies()); |
| |
| // providing some of the dataset dirs required as per coordinator specification with holes |
| // before and after action creation time |
| createTestCaseSubDir("2009/03/05/_SUCCESS".split("/")); |
| createTestCaseSubDir("2009/02/19/_SUCCESS".split("/")); |
| createTestCaseSubDir("2009/02/12/_SUCCESS".split("/")); |
| createTestCaseSubDir("2009/01/22/_SUCCESS".split("/")); |
| createTestCaseSubDir("2009/01/08/_SUCCESS".split("/")); |
| createTestCaseSubDir("2009/01/05/_SUCCESS".split("/")); |
| |
| // Run input check after making latest available |
| new CoordActionInputCheckXCommand(job.getId() + "@1", job.getId()).call(); |
| action = jpaService.execute(new CoordActionGetForInputCheckJPAExecutor(job.getId() + "@1")); |
| assertEquals(CoordCommandUtils.RESOLVED_UNRESOLVED_SEPARATOR + "${coord:latestRange(-3,0)}", |
| action.getMissingDependencies()); |
| assertEquals(pushMissingDependency, action.getPushMissingDependencies()); |
| |
| // Run input check after making push dependencies available |
| createTestCaseSubDir("2009/02/05/_SUCCESS".split("/")); |
| new CoordPushDependencyCheckXCommand(job.getId() + "@1").call(); |
| action = jpaService.execute(new CoordActionGetForInputCheckJPAExecutor(job.getId() + "@1")); |
| assertEquals("", action.getPushMissingDependencies()); |
| checkCoordAction(job.getId() + "@1", CoordCommandUtils.RESOLVED_UNRESOLVED_SEPARATOR |
| + "${coord:latestRange(-3,0)}", CoordinatorAction.Status.WAITING); |
| new CoordActionInputCheckXCommand(job.getId() + "@1", job.getId()).call(); |
| //Sleep for sometime as it gets requeued with 10ms delay on failure to acquire write lock |
| Thread.sleep(1000); |
| |
| action = jpaService.execute(new CoordActionGetForInputCheckJPAExecutor(job.getId() + "@1")); |
| assertEquals("", action.getMissingDependencies()); |
| actionXML = action.getActionXml(); |
| // Datasets should be picked up based on current time and not action creation/actual time. |
| String resolvedList = getTestCaseFileUri("2009/03/05") + CoordELFunctions.INSTANCE_SEPARATOR |
| + getTestCaseFileUri("2009/02/19") + CoordELFunctions.INSTANCE_SEPARATOR |
| + getTestCaseFileUri("2009/02/12") + CoordELFunctions.INSTANCE_SEPARATOR |
| + getTestCaseFileUri("2009/02/05"); |
| System.out.println("Expected: " + resolvedList); |
| System.out.println("Actual: " + actionXML.substring(actionXML.indexOf("<uris>") + 6, actionXML.indexOf("</uris>"))); |
| assertEquals(resolvedList, actionXML.substring(actionXML.indexOf("<uris>") + 6, actionXML.indexOf("</uris>"))); |
| } |
| |
| public void testActionInputCheckFuture() throws Exception { |
| String jobId = "0000000-" + new Date().getTime() + "-TestCoordActionInputCheckXCommand-C"; |
| Date startTime = DateUtils.parseDateOozieTZ("2009-02-15T23:59" + TZ); |
| Date endTime = DateUtils.parseDateOozieTZ("2009-02-16T23:59" + TZ); |
| CoordinatorJobBean job = addRecordToCoordJobTable(jobId, startTime, endTime, "future"); |
| new CoordMaterializeTransitionXCommand(job.getId(), 3600).call(); |
| |
| // providing some of the dataset dirs required as per coordinator specification with holes |
| createTestCaseSubDir("2009/02/12/_SUCCESS".split("/")); |
| createTestCaseSubDir("2009/02/26/_SUCCESS".split("/")); |
| createTestCaseSubDir("2009/03/05/_SUCCESS".split("/")); |
| createTestCaseSubDir("2009/03/26/_SUCCESS".split("/")); //limit is 5. So this should be ignored |
| |
| new CoordActionInputCheckXCommand(job.getId() + "@1", job.getId()).call(); |
| CoordinatorActionBean action = null; |
| JPAService jpaService = Services.get().get(JPAService.class); |
| try { |
| action = jpaService.execute(new CoordActionGetJPAExecutor(job.getId() + "@1")); |
| } |
| catch (JPAExecutorException se) { |
| fail("Action ID " + job.getId() + "@1" + " was not stored properly in db"); |
| } |
| |
| assertEquals(CoordCommandUtils.RESOLVED_UNRESOLVED_SEPARATOR + "${coord:futureRange(0,3,'5')}", |
| action.getMissingDependencies()); |
| |
| createTestCaseSubDir("2009/03/12/_SUCCESS".split("/")); |
| |
| new CoordActionInputCheckXCommand(job.getId() + "@1", job.getId()).call(); |
| try { |
| action = jpaService.execute(new CoordActionGetJPAExecutor(job.getId() + "@1")); |
| } |
| catch (JPAExecutorException se) { |
| fail("Action ID " + job.getId() + "@1" + " was not stored properly in db"); |
| } |
| |
| assertEquals("", action.getMissingDependencies()); |
| String actionXML = action.getActionXml(); |
| String resolvedList = getTestCaseFileUri("2009/02/12") + CoordELFunctions.INSTANCE_SEPARATOR |
| + getTestCaseFileUri("2009/02/26") + CoordELFunctions.INSTANCE_SEPARATOR |
| + getTestCaseFileUri("2009/03/05") + CoordELFunctions.INSTANCE_SEPARATOR |
| + getTestCaseFileUri("2009/03/12"); |
| assertEquals(resolvedList, actionXML.substring(actionXML.indexOf("<uris>") + 6, actionXML.indexOf("</uris>"))); |
| } |
| /** |
| * Testing a non existing namenode path |
| * |
| * @throws Exception |
| */ |
| public void testNonExistingNameNode() throws Exception { |
| String jobId = "0000000-" + new Date().getTime() + "-TestCoordActionInputCheckXCommand-C"; |
| Date startTime = DateUtils.parseDateUTC("2009-02-01T23:59Z"); |
| Date endTime = DateUtils.parseDateUTC("2009-02-02T23:59Z"); |
| CoordinatorJobBean job = addRecordToCoordJobTable(jobId, startTime, endTime); |
| new CoordMaterializeTransitionXCommand(job.getId(), 3600).call(); |
| CoordActionInputCheckXCommand caicc = new CoordActionInputCheckXCommand(job.getId() + "@1", job.getId()); |
| caicc.loadState(); |
| |
| // Override the name node while list for testing purpose only. |
| String[] whiteList = new String[1]; |
| whiteList[0] = "localhost:5330"; |
| services.destroy(); |
| setSystemProperty(HadoopAccessorService.NAME_NODE_WHITELIST, whiteList[0]); |
| services = new Services(); |
| services.init(); |
| |
| // setting the test path with nonExistDir |
| Path appPath = new Path(getFsTestCaseDir(), "coord"); |
| String inputDir = appPath.toString() + "/coord-input/2010/07/09/01/00"; |
| String nonExistDir = inputDir.replaceFirst("localhost", "nonExist"); |
| CoordinatorActionBean actionBean = new CoordinatorActionBean(); |
| try { |
| new CoordOldInputDependency().pathExists(actionBean, nonExistDir, new XConfiguration(), getTestUser()); |
| fail("Should throw exception due to non-existent NN path. Therefore fail"); |
| } |
| catch (IOException ioe) { |
| assertEquals(actionBean.getErrorCode(), "E0901"); |
| assertTrue(actionBean.getErrorMessage().contains("not in Oozie's whitelist")); |
| } |
| } |
| |
| /** |
| * This test case verifies if getCoordInputCheckRequeueInterval picks up the |
| * overridden value. In reality, the value could be overridden in |
| * oozie-site.xml. |
| * |
| * @throws Exception |
| */ |
| public void testRequeueInterval() throws Exception { |
| /* |
| * Create a dummy Coordinator Job to pass to |
| * CoordActionInputCheckXCommand constructor. |
| */ |
| String jobId = "0000000-" + new Date().getTime() + "-TestCoordActionInputCheckXCommand-C"; |
| Date startTime = DateUtils.parseDateOozieTZ("2009-02-01T23:59" + TZ); |
| Date endTime = DateUtils.parseDateOozieTZ("2009-02-02T23:59" + TZ); |
| CoordinatorJobBean job = addRecordToCoordJobTable(jobId, startTime, endTime); |
| /* Override the property value for testing purpose only. */ |
| long testedValue = 12000; |
| Services.get().getConf().setLong(CoordActionInputCheckXCommand.CONF_COORD_INPUT_CHECK_REQUEUE_INTERVAL, |
| testedValue); |
| |
| CoordActionInputCheckXCommand caicc = new CoordActionInputCheckXCommand(job.getId() + "@1", job.getId()); |
| |
| long effectiveValue = caicc.getCoordInputCheckRequeueInterval(); |
| // Verify if two values are same. |
| assertEquals(testedValue, effectiveValue); |
| } |
| |
| public void testResolveCoordConfiguration() { |
| try { |
| CoordinatorJobBean job = addRecordToCoordJobTableForWaiting("coord-job-for-action-input-check.xml", |
| CoordinatorJob.Status.RUNNING, false, true); |
| |
| CoordinatorActionBean action = addRecordToCoordActionTableForWaiting(job.getId(), 1, |
| CoordinatorAction.Status.WAITING, "coord-action-for-action-input-check.xml"); |
| |
| createTestCaseSubDir("2009/01/29/_SUCCESS".split("/")); |
| createTestCaseSubDir("2009/01/22/_SUCCESS".split("/")); |
| createTestCaseSubDir("2009/01/15/_SUCCESS".split("/")); |
| createTestCaseSubDir("2009/01/08/_SUCCESS".split("/")); |
| sleep(3000); |
| new CoordActionInputCheckXCommand(action.getId(), job.getId()).call(); |
| sleep(3000); |
| final JPAService jpaService = Services.get().get(JPAService.class); |
| CoordActionGetJPAExecutor coordGetCmd = new CoordActionGetJPAExecutor(action.getId()); |
| CoordinatorActionBean caBean = jpaService.execute(coordGetCmd); |
| |
| Element eAction = XmlUtils.parseXml(caBean.getActionXml()); |
| Element configElem = eAction.getChild("action", eAction.getNamespace()) |
| .getChild("workflow", eAction.getNamespace()).getChild("configuration", eAction.getNamespace()); |
| List<?> elementList = configElem.getChildren("property", configElem.getNamespace()); |
| Element e1 = (Element) elementList.get(0); |
| Element e2 = (Element) elementList.get(1); |
| assertEquals( |
| "file://,testDir/2009/29,file://,testDir/2009/22,file://,testDir/2009/15,file://,testDir/2009/08", |
| e1.getChild("value", e1.getNamespace()).getValue()); |
| assertEquals("file://,testDir/2009/29", e2.getChild("value", e1.getNamespace()).getValue()); |
| |
| } |
| catch (Exception e) { |
| e.printStackTrace(); |
| fail("Unexpected exception"); |
| } |
| } |
| |
| /** |
| * This test verifies that for a coordinator with no input dependencies |
| * action is not stuck in WAITING |
| * |
| * @throws Exception |
| */ |
| public void testNoDatasetDependency() throws Exception { |
| /* |
| * create coordinator job |
| */ |
| CoordinatorJobBean coordJob = new CoordinatorJobBean(); |
| coordJob.setId("0000000" + new Date().getTime() + "-TestCoordActionInputCheckXCommand-C"); |
| coordJob.setAppName("testApp"); |
| coordJob.setAppPath("testAppPath"); |
| coordJob.setStatus(CoordinatorJob.Status.RUNNING); |
| coordJob.setCreatedTime(new Date()); |
| coordJob.setLastModifiedTime(new Date()); |
| coordJob.setUser("test"); |
| coordJob.setGroup("testGroup"); |
| coordJob.setTimeZone("UTC"); |
| coordJob.setTimeUnit(Timeunit.DAY); |
| coordJob.setMatThrottling(2); |
| try { |
| coordJob.setStartTime(DateUtils.parseDateOozieTZ("2009-02-01T23:59" + TZ)); |
| coordJob.setEndTime(DateUtils.parseDateOozieTZ("2009-02-02T23:59" + TZ)); |
| } |
| catch (Exception e) { |
| e.printStackTrace(); |
| fail("Could not set Date/time"); |
| } |
| XConfiguration jobConf = new XConfiguration(); |
| jobConf.set(OozieClient.USER_NAME, getTestUser()); |
| String confStr = jobConf.toXmlString(false); |
| coordJob.setConf(confStr); |
| String wfXml = IOUtils.getResourceAsString("wf-no-op.xml", -1); |
| writeToFile(wfXml, getFsTestCaseDir(), "workflow.xml"); |
| String appXml = "<coordinator-app xmlns='uri:oozie:coordinator:0.2' name='NAME' frequency=\"1\" start='2009-02-01T01:00" |
| + TZ + "' end='2009-02-03T23:59" + TZ + "' timezone='UTC' freq_timeunit='DAY' end_of_duration='NONE'>"; |
| appXml += "<output-events>"; |
| appXml += "<data-out name='LOCAL_A' dataset='local_a'>"; |
| appXml += "<dataset name='local_a' frequency='7' initial-instance='2009-01-01T01:00" + TZ |
| + "' timezone='UTC' freq_timeunit='DAY' end_of_duration='NONE'>"; |
| appXml += "<uri-template>file://" + getFsTestCaseDir() + "/${YEAR}/${MONTH}/${DAY}</uri-template>"; |
| appXml += "</dataset>"; |
| appXml += "<start-instance>${coord:current(-3)}</start-instance>"; |
| appXml += "<instance>${coord:current(0)}</instance>"; |
| appXml += "</data-out>"; |
| appXml += "</output-events>"; |
| appXml += "<action>"; |
| appXml += "<workflow>"; |
| appXml += "<app-path>" + getFsTestCaseDir() + "/workflow.xml</app-path>"; |
| appXml += "</workflow>"; |
| appXml += "</action>"; |
| appXml += "</coordinator-app>"; |
| coordJob.setJobXml(appXml); |
| coordJob.setLastActionNumber(0); |
| coordJob.setFrequency("1"); |
| coordJob.setConcurrency(1); |
| JPAService jpaService = Services.get().get(JPAService.class); |
| if (jpaService != null) { |
| try { |
| jpaService.execute(new CoordJobInsertJPAExecutor(coordJob)); |
| } |
| catch (JPAExecutorException e) { |
| throw new CommandException(e); |
| } |
| } |
| else { |
| fail("Unable to insert the test job record to table"); |
| } |
| new CoordMaterializeTransitionXCommand(coordJob.getId(), 3600).call(); |
| /* |
| * check coord action READY |
| */ |
| new CoordActionInputCheckXCommand(coordJob.getId() + "@1", coordJob.getId()).call(); |
| CoordinatorActionBean action = null; |
| try { |
| jpaService = Services.get().get(JPAService.class); |
| action = jpaService.execute(new CoordActionGetJPAExecutor(coordJob.getId() + "@1")); |
| } |
| catch (JPAExecutorException se) { |
| fail("Action ID " + coordJob.getId() + "@1" + " was not stored properly in db"); |
| } |
| // for no dataset dependency, action will proceed directly to start |
| // workflow synchronously after being READY (since n(actions) < concurrency) |
| assertEquals(CoordinatorAction.Status.RUNNING, action.getStatus()); |
| |
| action.setMissingDependencies(""); |
| action.setStatus(CoordinatorAction.Status.WAITING); |
| action.setExternalId(null); |
| try { |
| jpaService = Services.get().get(JPAService.class); |
| CoordActionQueryExecutor.getInstance().executeUpdate(CoordActionQuery.UPDATE_COORD_ACTION, action); |
| } |
| catch (JPAExecutorException se) { |
| fail("Action ID " + coordJob.getId() + "@1" + " was not stored properly in db"); |
| } |
| |
| new CoordActionInputCheckXCommand(coordJob.getId() + "@1", coordJob.getId()).call(); |
| try { |
| jpaService = Services.get().get(JPAService.class); |
| action = jpaService.execute(new CoordActionGetJPAExecutor(coordJob.getId() + "@1")); |
| } |
| catch (JPAExecutorException se) { |
| fail("Action ID " + coordJob.getId() + "@1" + " was not stored properly in db"); |
| } |
| // for no dataset dependency, action will proceed directly to start |
| // workflow synchronously after being READY (since n(actions) < concurrency) |
| assertEquals(CoordinatorAction.Status.RUNNING, action.getStatus()); |
| } |
| |
| @Test |
| public void testTimeout() throws Exception { |
| String missingDeps = "hdfs:///dirx/filex"; |
| String actionId = addInitRecords(missingDeps, null, TZ); |
| new CoordActionInputCheckXCommand(actionId, actionId.substring(0, actionId.indexOf("@"))).call(); |
| // Timeout is 10 mins. Change action creation time to before 12 min to make the action |
| // timeout. |
| long timeOutCreationTime = System.currentTimeMillis() - (12 * 60 * 1000); |
| setCoordActionCreationTime(actionId, timeOutCreationTime); |
| new CoordActionInputCheckXCommand(actionId, actionId.substring(0, actionId.indexOf("@"))).call(); |
| checkCoordActionStatus(actionId, CoordinatorAction.Status.TIMEDOUT); |
| checkCoordAction(actionId, missingDeps, CoordinatorAction.Status.TIMEDOUT); |
| } |
| |
| @Test |
| public void testTimeoutWithUnResolved() throws Exception { |
| String jobId = "0000000-" + new Date().getTime() + "-TestCoordActionInputCheckXCommand-C"; |
| Date startTime = DateUtils.parseDateOozieTZ("2009-02-15T23:59" + TZ); |
| Date endTime = DateUtils.parseDateOozieTZ("2009-02-16T23:59" + TZ); |
| CoordinatorJobBean job = addRecordToCoordJobTable(jobId, startTime, endTime, "latest"); |
| new CoordMaterializeTransitionXCommand(job.getId(), 3600).call(); |
| CoordinatorActionBean action = CoordActionQueryExecutor.getInstance() |
| .get(CoordActionQuery.GET_COORD_ACTION, job.getId() + "@1"); |
| assertEquals(CoordCommandUtils.RESOLVED_UNRESOLVED_SEPARATOR + "${coord:latestRange(-3,0)}", |
| action.getMissingDependencies()); |
| long timeOutCreationTime = System.currentTimeMillis() - (13 * 60 * 1000); |
| setCoordActionCreationTime(action.getId(), timeOutCreationTime); |
| new CoordActionInputCheckXCommand(action.getId(), action.getJobId()).call(); |
| checkCoordActionStatus(action.getId(), CoordinatorAction.Status.TIMEDOUT); |
| } |
| |
| @Test |
| public void testTimeoutWithException() throws Exception { |
| String missingDeps = "nofs:///dirx/filex"; |
| String actionId = addInitRecords(missingDeps, null, TZ); |
| try { |
| new CoordActionInputCheckXCommand(actionId, actionId.substring(0, actionId.indexOf("@"))).call(); |
| fail(); |
| } |
| catch (Exception e) { |
| assertTrue(e.getMessage().contains("No FileSystem for scheme")); |
| } |
| // Timeout is 10 mins. Change action created time to before 12 min to make the action |
| // timeout. |
| long timeOutCreationTime = System.currentTimeMillis() - (12 * 60 * 1000); |
| setCoordActionCreationTime(actionId, timeOutCreationTime); |
| try { |
| new CoordActionInputCheckXCommand(actionId, actionId.substring(0, actionId.indexOf("@"))).call(); |
| fail(); |
| } |
| catch (Exception e) { |
| assertTrue(e.getMessage().contains("No FileSystem for scheme")); |
| } |
| checkCoordActionStatus(actionId, CoordinatorAction.Status.TIMEDOUT); |
| checkCoordAction(actionId, missingDeps, CoordinatorAction.Status.TIMEDOUT); |
| } |
| |
| @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, 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 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(start.getTime()); // 65 min ago |
| setCoordActionNominalTime(actionId1, nomTime.getTime()); |
| new CoordActionInputCheckXCommand(actionId1, job.getId()).call(); |
| checkCoordActionStatus(actionId1, CoordinatorAction.Status.SKIPPED); |
| |
| // @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(start.getTime() + 68 * 60 * 1000); // 3 min from now |
| job.setEndTime(endTime); |
| CoordJobQueryExecutor.getInstance().executeUpdate(CoordJobQueryExecutor.CoordJobQuery.UPDATE_COORD_JOB, job); |
| new CoordActionInputCheckXCommand(actionId7, job.getId()).call(); |
| checkCoordActionStatus(actionId7, CoordinatorAction.Status.WAITING); |
| |
| // @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); |
| setCoordActionStatus(actionId1, CoordinatorAction.Status.WAITING); |
| new CoordActionInputCheckXCommand(actionId1, job.getId()).call(); |
| checkCoordActionStatus(actionId1, CoordinatorAction.Status.WAITING); |
| } |
| |
| @Test |
| public void testNone() throws Exception { |
| CoordinatorJobBean job = addRecordToCoordJobTableForWaiting("coord-job-for-action-input-check.xml", |
| CoordinatorJob.Status.RUNNING, false, true); |
| job.setExecutionOrder(CoordinatorJobBean.Execution.NONE); |
| job.setFrequency("10"); |
| job.setTimeUnit(Timeunit.MINUTE); |
| CoordJobQueryExecutor.getInstance().executeUpdate(CoordJobQueryExecutor.CoordJobQuery.UPDATE_COORD_JOB, job); |
| String missingDeps = "hdfs:///dirx/filex"; |
| |
| // nominal time is one hour past. So action will be skipped |
| String actionId1 = addInitRecords(missingDeps, null, TZ, job, 1); |
| Date nomTime = new Date(new Date().getTime() - 60 * 60 * 1000); // 1 hour ago |
| setCoordActionNominalTime(actionId1, nomTime.getTime()); |
| new CoordActionInputCheckXCommand(actionId1, job.getId()).call(); |
| checkCoordActionStatus(actionId1, CoordinatorAction.Status.SKIPPED); |
| |
| // Nominal time is 60 minutes from now, action should be in WAITING stage |
| 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); |
| |
| // Nominal times of both actions are more than one minute past, so both will be skipped |
| 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.SKIPPED); |
| |
| String actionId4 = addInitRecords(missingDeps, null, TZ, job, 4); |
| nomTime = new Date(new Date().getTime() - 2 * 60 * 1000); // 2 minutes ago |
| setCoordActionNominalTime(actionId4, nomTime.getTime()); |
| new CoordActionInputCheckXCommand(actionId4, job.getId()).call(); |
| checkCoordActionStatus(actionId4, CoordinatorAction.Status.SKIPPED); |
| |
| |
| } |
| |
| public void testExceptionOnInvalidElFunction() { |
| try { |
| CoordinatorJobBean job = addRecordToCoordJobTableForWaiting("coord-hdfsinput-invalid-elfunction.xml", |
| CoordinatorJob.Status.RUNNING, false, true); |
| |
| CoordinatorActionBean action = addRecordToCoordActionTableForWaiting(job.getId(), 1, |
| CoordinatorAction.Status.WAITING, "coord-hdfsinput-invalid-elfunction.xml"); |
| |
| createTestCaseSubDir("2009/01/29/_SUCCESS".split("/")); |
| createTestCaseSubDir("2009/01/22/_SUCCESS".split("/")); |
| createTestCaseSubDir("2009/01/15/_SUCCESS".split("/")); |
| createTestCaseSubDir("2009/01/08/_SUCCESS".split("/")); |
| sleep(3000); |
| final String actionId = action.getId(); |
| try { |
| new CoordActionInputCheckXCommand(action.getId(), job.getId()).call(); |
| waitFor(6000, new Predicate() { |
| @Override |
| public boolean evaluate() throws Exception { |
| CoordinatorActionBean action = CoordActionQueryExecutor.getInstance() |
| .get(CoordActionQueryExecutor.CoordActionQuery.GET_COORD_ACTION, actionId); |
| return action.getStatus() == CoordinatorAction.Status.FAILED; |
| } |
| }); |
| fail("Should throw an exception"); |
| } |
| catch (Exception e) { |
| assertTrue(e.getMessage().contains("Coord Action Input Check Error")); |
| } |
| } |
| catch (Exception e) { |
| e.printStackTrace(System.out); |
| fail("Unexpected exception"); |
| } |
| } |
| protected CoordinatorActionBean addRecordToCoordActionTableForWaiting(String jobId, int actionNum, |
| CoordinatorAction.Status status, String resourceXmlName) throws Exception { |
| CoordinatorActionBean action = createCoordAction(jobId, actionNum, status, resourceXmlName, 0, TZ, null); |
| String missDeps = getTestCaseFileUri("2009/01/29/_SUCCESS") + "#" |
| + getTestCaseFileUri("2009/01/22/_SUCCESS") + "#" |
| + getTestCaseFileUri("2009/01/15/_SUCCESS") + "#" |
| + getTestCaseFileUri("2009/01/08/_SUCCESS"); |
| |
| action.setMissingDependencies(missDeps); |
| |
| try { |
| JPAService jpaService = Services.get().get(JPAService.class); |
| assertNotNull(jpaService); |
| CoordActionInsertJPAExecutor coordActionInsertCmd = new CoordActionInsertJPAExecutor(action); |
| jpaService.execute(coordActionInsertCmd); |
| } |
| catch (JPAExecutorException je) { |
| je.printStackTrace(); |
| fail("Unable to insert the test coord action record to table"); |
| throw je; |
| } |
| return action; |
| } |
| |
| private CoordinatorJobBean addRecordToCoordJobTable(String jobId, Date start, Date end) throws CommandException { |
| return addRecordToCoordJobTable(jobId, start, end, "current"); |
| } |
| |
| private CoordinatorJobBean addRecordToCoordJobTable(String jobId, Date start, Date end, |
| String dataInType) throws CommandException { |
| CoordinatorJobBean coordJob = new CoordinatorJobBean(); |
| coordJob.setId(jobId); |
| coordJob.setAppName("testApp"); |
| coordJob.setAppPath("testAppPath"); |
| coordJob.setStatus(CoordinatorJob.Status.RUNNING); |
| coordJob.setCreatedTime(new Date()); |
| coordJob.setLastModifiedTime(new Date()); |
| coordJob.setUser("test"); |
| coordJob.setGroup("testGroup"); |
| coordJob.setTimeZone("UTC"); |
| coordJob.setTimeUnit(Timeunit.DAY); |
| coordJob.setMatThrottling(2); |
| try { |
| coordJob.setStartTime(start); |
| coordJob.setEndTime(end); |
| } |
| catch (Exception e) { |
| e.printStackTrace(); |
| fail("Could not set Date/time"); |
| } |
| |
| String testDir = getTestCaseDir(); |
| |
| XConfiguration jobConf = new XConfiguration(); |
| jobConf.set(OozieClient.USER_NAME, getTestUser()); |
| |
| String confStr = jobConf.toXmlString(false); |
| coordJob.setConf(confStr); |
| String appXml = "<coordinator-app xmlns='uri:oozie:coordinator:0.2' name='NAME' frequency=\"1\" start='2009-02-01T01:00" |
| + TZ + "' end='2009-02-03T23:59" + TZ + "' timezone='UTC' freq_timeunit='DAY' end_of_duration='NONE'>"; |
| appXml += "<controls>"; |
| appXml += "<timeout>10</timeout>"; |
| appXml += "<concurrency>2</concurrency>"; |
| appXml += "<execution>LIFO</execution>"; |
| appXml += "</controls>"; |
| appXml += "<input-events>"; |
| appXml += "<data-in name='A' dataset='a'>"; |
| appXml += "<dataset name='a' frequency='7' initial-instance='2009-01-01T01:00" + TZ + "' timezone='UTC'" |
| + " freq_timeunit='DAY' end_of_duration='NONE'>"; |
| appXml += "<uri-template>" + getTestCaseFileUri("${YEAR}/${MONTH}/${DAY}" )+ "</uri-template>"; |
| appXml += "</dataset>"; |
| if (dataInType.equals("future")) { |
| appXml += "<start-instance>${coord:" + dataInType + "(0,5)}</start-instance>"; |
| appXml += "<end-instance>${coord:" + dataInType + "(3,5)}</end-instance>"; |
| } |
| else if (dataInType.equals("latest")) { |
| appXml += "<start-instance>${coord:" + dataInType + "(-3)}</start-instance>"; |
| appXml += "<end-instance>${coord:" + dataInType + "(0)}</end-instance>"; |
| } |
| else if (dataInType.equals("current")) { |
| appXml += "<start-instance>${coord:" + dataInType + "(-3)}</start-instance>"; |
| appXml += "<end-instance>${coord:" + dataInType + "(1)}</end-instance>"; |
| } |
| appXml += "</data-in>"; |
| appXml += "</input-events>"; |
| appXml += "<output-events>"; |
| appXml += "<data-out name='LOCAL_A' dataset='local_a'>"; |
| appXml += "<dataset name='local_a' frequency='7' initial-instance='2009-01-01T01:00" + TZ + "' timezone='UTC'" |
| + " freq_timeunit='DAY' end_of_duration='NONE'>"; |
| appXml += "<uri-template>" + getTestCaseFileUri("${YEAR}/${MONTH}/${DAY}" )+ "</uri-template>"; |
| appXml += "</dataset>"; |
| appXml += "<start-instance>${coord:current(-3)}</start-instance>"; |
| appXml += "<instance>${coord:current(0)}</instance>"; |
| appXml += "</data-out>"; |
| appXml += "</output-events>"; |
| appXml += "<action>"; |
| appXml += "<workflow>"; |
| appXml += "<app-path>hdfs:///tmp/workflows/</app-path>"; |
| appXml += "<configuration>"; |
| appXml += "<property>"; |
| appXml += "<name>inputA</name>"; |
| appXml += "<value>${coord:dataIn('A')}</value>"; |
| appXml += "</property>"; |
| appXml += "<property>"; |
| appXml += "<name>inputB</name>"; |
| appXml += "<value>${coord:dataOut('LOCAL_A')}</value>"; |
| appXml += "</property>"; |
| appXml += "</configuration>"; |
| appXml += "</workflow>"; |
| appXml += "</action>"; |
| appXml += "</coordinator-app>"; |
| coordJob.setJobXml(appXml); |
| coordJob.setLastActionNumber(0); |
| coordJob.setFrequency("1"); |
| coordJob.setExecutionOrder(Execution.FIFO); |
| coordJob.setConcurrency(1); |
| |
| JPAService jpaService = Services.get().get(JPAService.class); |
| if (jpaService != null) { |
| try { |
| jpaService.execute(new CoordJobInsertJPAExecutor(coordJob)); |
| } |
| catch (JPAExecutorException e) { |
| throw new CommandException(e); |
| } |
| } |
| else { |
| fail("Unable to insert the test job record to table"); |
| } |
| return coordJob; |
| } |
| |
| private CoordinatorActionBean checkCoordAction(String actionId, String expDeps, CoordinatorAction.Status stat) |
| throws Exception { |
| try { |
| JPAService jpaService = Services.get().get(JPAService.class); |
| CoordinatorActionBean action = jpaService.execute(new CoordActionGetJPAExecutor(actionId)); |
| String missDeps = action.getMissingDependencies(); |
| assertEquals(expDeps, missDeps); |
| assertEquals(stat, action.getStatus()); |
| |
| return action; |
| } |
| catch (JPAExecutorException se) { |
| throw new Exception("Action ID " + actionId + " was not stored properly in db"); |
| } |
| } |
| |
| public void testHarFileInputCheck() throws Exception { |
| CoordinatorJobBean job = addRecordToCoordJobTableForWaiting("coord-job-for-action-input-check.xml", |
| CoordinatorJob.Status.RUNNING, false, true); |
| String pathName = createTestCaseSubDir("2009/01/29".split("/")); |
| CoordJobQueryExecutor.getInstance().executeUpdate(CoordJobQueryExecutor.CoordJobQuery.UPDATE_COORD_JOB, job); |
| String missingDeps = "file://" + pathName + CoordELFunctions.INSTANCE_SEPARATOR |
| + "har:///dirx/archive.har/data"; |
| String actionId1 = addInitRecords(missingDeps, null, TZ, job, 1); |
| new CoordActionInputCheckXCommand(actionId1, job.getId()).call(); |
| CoordinatorActionBean coordAction = CoordActionQueryExecutor.getInstance() |
| .get(CoordActionQuery.GET_COORD_ACTION, actionId1); |
| // the directory (2009/01/29) exists and successfully removed from |
| // missing dependency of the coord action |
| assertEquals(coordAction.getMissingDependencies(), "har:///dirx/archive.har/data"); |
| } |
| } |
