| /** |
| * 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.falcon.resource; |
| |
| import com.sun.jersey.api.client.ClientResponse; |
| import org.apache.commons.io.FileUtils; |
| import org.apache.falcon.FalconException; |
| import org.apache.falcon.client.FalconCLIException; |
| import org.apache.falcon.entity.v0.Entity; |
| import org.apache.falcon.entity.v0.EntityType; |
| import org.apache.falcon.entity.v0.feed.Cluster; |
| import org.apache.falcon.entity.v0.feed.Feed; |
| import org.apache.falcon.entity.v0.feed.Location; |
| import org.apache.falcon.entity.v0.feed.LocationType; |
| import org.apache.falcon.entity.v0.feed.Locations; |
| import org.apache.falcon.entity.v0.process.Input; |
| import org.apache.falcon.entity.v0.process.Process; |
| import org.apache.falcon.entity.v0.process.Property; |
| import org.apache.falcon.entity.v0.process.Validity; |
| import org.apache.falcon.hadoop.HadoopClientFactory; |
| import org.apache.falcon.unit.FalconUnit; |
| import org.apache.falcon.util.BuildProperties; |
| import org.apache.falcon.util.DeploymentProperties; |
| import org.apache.falcon.util.FalconTestUtil; |
| import org.apache.falcon.util.OozieTestUtils; |
| import org.apache.hadoop.fs.FileStatus; |
| import org.apache.hadoop.fs.FileSystem; |
| import org.apache.hadoop.fs.Path; |
| import org.apache.oozie.client.BundleJob; |
| 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.testng.Assert; |
| import org.testng.annotations.AfterMethod; |
| import org.testng.annotations.BeforeClass; |
| import org.testng.annotations.Test; |
| |
| import javax.ws.rs.core.MediaType; |
| import java.io.File; |
| import java.io.IOException; |
| import java.util.Calendar; |
| import java.util.Date; |
| import java.util.HashMap; |
| import java.util.Iterator; |
| import java.util.List; |
| import java.util.Map; |
| import java.util.TimeZone; |
| import java.util.concurrent.Callable; |
| import java.util.concurrent.ExecutorService; |
| import java.util.concurrent.Executors; |
| import java.util.concurrent.Future; |
| import java.util.regex.Pattern; |
| |
| /** |
| * Test class for Entity REST APIs. |
| * |
| * Tests should be enabled only in local environments as they need running instance of the web server. |
| */ |
| @Test(groups = {"exhaustive"}) |
| public class EntityManagerJerseyIT extends AbstractSchedulerManagerJerseyIT { |
| |
| private static final String START_INSTANCE = "2012-04-20T00:00Z"; |
| private static final String SLEEP_WORKFLOW = "sleepWorkflow.xml"; |
| |
| @BeforeClass |
| @Override |
| public void setup() throws Exception { |
| String version = System.getProperty("project.version"); |
| String buildDir = System.getProperty("project.build.directory"); |
| System.setProperty("falcon.libext", buildDir + "/../../unit/target/falcon-unit-" + version + ".jar"); |
| super.setup(); |
| } |
| |
| @AfterMethod |
| @Override |
| public void cleanUpActionXml() throws IOException, FalconException { |
| //Needed since oozie writes action xml to current directory. |
| FileUtils.deleteQuietly(new File("action.xml")); |
| FileUtils.deleteQuietly(new File(".action.xml.crc")); |
| contexts.remove(); |
| } |
| |
| private ThreadLocal<UnitTestContext> contexts = new ThreadLocal<UnitTestContext>(); |
| |
| private UnitTestContext newContext() throws FalconException, IOException { |
| contexts.set(new UnitTestContext()); |
| return contexts.get(); |
| } |
| |
| static void assertLibs(FileSystem fs, Path path) throws IOException { |
| FileStatus[] libs = fs.listStatus(path); |
| Assert.assertNotNull(libs); |
| } |
| |
| private Entity getDefinition(EntityType type, String name) throws Exception { |
| Entity entity = falconUnitClient.getDefinition(type.name(), name, null); |
| return entity; |
| } |
| |
| private void updateEndtime(Process process) { |
| Validity processValidity = process.getClusters().getClusters().get(0).getValidity(); |
| processValidity.setEnd(new Date(new Date().getTime() + 2 * 24 * 60 * 60 * 1000)); |
| } |
| |
| @Test |
| public void testLibExtensions() throws Exception { |
| UnitTestContext context = newContext(); |
| submitCluster(context); |
| FileSystem fs = FalconUnit.getFileSystem(); |
| assertLibs(fs, new Path("/projects/falcon/working/libext/FEED/retention")); |
| assertLibs(fs, new Path("/projects/falcon/working/libext/PROCESS")); |
| |
| String tmpFileName = TestContext.overlayParametersOverTemplate(UnitTestContext.FEED_TEMPLATE1, context.overlay); |
| Feed feed = (Feed) EntityType.FEED.getUnmarshaller().unmarshal(new File(tmpFileName)); |
| Location location = new Location(); |
| location.setPath("fsext://global:00/falcon/test/input/${YEAR}/${MONTH}/${DAY}/${HOUR}"); |
| location.setType(LocationType.DATA); |
| Cluster cluster = feed.getClusters().getClusters().get(0); |
| cluster.setLocations(new Locations()); |
| feed.getClusters().getClusters().get(0).getLocations().getLocations().add(location); |
| |
| File tmpFile = UnitTestContext.getTempFile(); |
| EntityType.FEED.getMarshaller().marshal(feed, tmpFile); |
| |
| APIResult result = falconUnitClient.submitAndSchedule(EntityType.FEED.name(), tmpFile.getAbsolutePath(), true, |
| null, null); |
| Assert.assertEquals(result.getStatus(), APIResult.Status.SUCCEEDED); |
| } |
| |
| @Test |
| public void testUpdateCheckUser() throws Exception { |
| UnitTestContext context = newContext(); |
| String tmpFileName = TestContext.overlayParametersOverTemplate(UnitTestContext.PROCESS_TEMPLATE, |
| context.overlay); |
| Process process = (Process) EntityType.PROCESS.getUnmarshaller().unmarshal(new File(tmpFileName)); |
| updateEndtime(process); |
| File tmpFile = UnitTestContext.getTempFile(); |
| EntityType.PROCESS.getMarshaller().marshal(process, tmpFile); |
| submitCluster(context); |
| context.prepare(); |
| submitFeeds(context.overlay); |
| submitProcess(tmpFile.getAbsolutePath(), context.overlay); |
| scheduleProcess(context.getProcessName(), context.getClusterName(), getAbsolutePath(SLEEP_WORKFLOW)); |
| waitForStatus(EntityType.PROCESS.name(), context.getProcessName(), START_INSTANCE, |
| InstancesResult.WorkflowStatus.RUNNING); |
| |
| List<BundleJob> bundles = OozieTestUtils.getBundles(context); |
| Assert.assertEquals(bundles.size(), 1); |
| Assert.assertEquals(bundles.get(0).getUser(), TestContext.REMOTE_USER); |
| |
| Feed feed = (Feed) getDefinition(EntityType.FEED, context.outputFeedName); |
| |
| //change output feed path and update feed as another user |
| feed.getLocations().getLocations().get(0).setPath("/falcon/test/output2/${YEAR}/${MONTH}/${DAY}"); |
| tmpFile = TestContext.getTempFile(); |
| feed.getEntityType().getMarshaller().marshal(feed, tmpFile); |
| APIResult result = falconUnitClient.update(EntityType.FEED.name(), feed.getName(), |
| tmpFile.getAbsolutePath(), true, null); |
| Assert.assertEquals(result.getStatus(), APIResult.Status.SUCCEEDED); |
| |
| bundles = OozieTestUtils.getBundles(context); |
| Assert.assertEquals(bundles.size(), 2); |
| Assert.assertEquals(bundles.get(0).getUser(), TestContext.REMOTE_USER); |
| Assert.assertEquals(bundles.get(1).getUser(), TestContext.REMOTE_USER); |
| } |
| |
| public void testOptionalInput() throws Exception { |
| UnitTestContext context = newContext(); |
| String tmpFileName = TestContext.overlayParametersOverTemplate(TestContext.PROCESS_TEMPLATE, context.overlay); |
| Process process = (Process) EntityType.PROCESS.getUnmarshaller().unmarshal(new File(tmpFileName)); |
| |
| Input in1 = process.getInputs().getInputs().get(0); |
| Input in2 = new Input(); |
| in2.setFeed(in1.getFeed()); |
| in2.setName("input2"); |
| in2.setOptional(true); |
| in2.setPartition(in1.getPartition()); |
| in2.setStart("now(-1,0)"); |
| in2.setEnd("now(0,0)"); |
| process.getInputs().getInputs().add(in2); |
| |
| File tmpFile = TestContext.getTempFile(); |
| EntityType.PROCESS.getMarshaller().marshal(process, tmpFile); |
| schedule(context); |
| waitForStatus(EntityType.PROCESS.name(), context.getProcessName(), START_INSTANCE, |
| InstancesResult.WorkflowStatus.SUCCEEDED); |
| } |
| |
| public void testDryRun() throws Exception { |
| //Schedule of invalid process should fail because of dryRun, and should pass when dryrun is skipped |
| UnitTestContext context = newContext(); |
| String tmpFileName = TestContext.overlayParametersOverTemplate(UnitTestContext.PROCESS_TEMPLATE, |
| context.overlay); |
| Process process = (Process) EntityType.PROCESS.getUnmarshaller().unmarshal(new File(tmpFileName)); |
| Property prop = new Property(); |
| prop.setName("newProp"); |
| prop.setValue("${instanceTim()}"); //invalid property |
| process.getProperties().getProperties().add(prop); |
| File tmpFile = TestContext.getTempFile(); |
| EntityType.PROCESS.getMarshaller().marshal(process, tmpFile); |
| |
| try { |
| falconUnitClient.validate(EntityType.PROCESS.name(), tmpFile.getAbsolutePath(), |
| true, null); |
| Assert.fail("Exception should be Thrown"); |
| } catch (Exception e) { |
| //ignore |
| } |
| schedule(context); |
| |
| //Fix the process and then submitAndSchedule should succeed |
| Iterator<Property> itr = process.getProperties().getProperties().iterator(); |
| while (itr.hasNext()) { |
| Property myProp = itr.next(); |
| if (myProp.getName().equals("newProp")) { |
| itr.remove(); |
| } |
| } |
| tmpFile = TestContext.getTempFile(); |
| process.setName("process" + System.currentTimeMillis()); |
| EntityType.PROCESS.getMarshaller().marshal(process, tmpFile); |
| APIResult result = falconUnitClient.submitAndSchedule(EntityType.PROCESS.name(), |
| tmpFile.getAbsolutePath(), true, null, null); |
| Assert.assertEquals(result.getStatus(), APIResult.Status.SUCCEEDED); |
| |
| // update where dryrun is disabled should succeed. |
| tmpFile = TestContext.getTempFile(); |
| process.getEntityType().getMarshaller().marshal(process, tmpFile); |
| result = falconUnitClient.update(EntityType.PROCESS.name(), process.getName(), |
| tmpFile.getAbsolutePath(), true, null); |
| Assert.assertEquals(result.getStatus(), APIResult.Status.SUCCEEDED); |
| } |
| |
| @Test |
| public void testUpdateSuspendedEntity() throws Exception { |
| UnitTestContext context = newContext(); |
| schedule(context); |
| waitForStatus(EntityType.PROCESS.name(), context.processName, START_INSTANCE, |
| InstancesResult.WorkflowStatus.RUNNING); |
| |
| //Suspend entity |
| Process process = (Process) getDefinition(EntityType.PROCESS, context.processName); |
| APIResult result = falconUnitClient.suspend(EntityType.PROCESS, process.getName(), context.colo, |
| null); |
| Assert.assertEquals(result.getStatus(), APIResult.Status.SUCCEEDED); |
| result = falconUnitClient.getStatus(EntityType.PROCESS, context.processName, context.clusterName, |
| null, false); |
| Assert.assertEquals(result.getStatus(), APIResult.Status.SUCCEEDED); |
| Assert.assertEquals(result.getMessage(), "SUSPENDED"); |
| |
| process.getProperties().getProperties().get(0).setName("newprop"); |
| Date endTime = getEndTime(); |
| process.getClusters().getClusters().get(0).getValidity().setEnd(endTime); |
| File tmpFile = TestContext.getTempFile(); |
| process.getEntityType().getMarshaller().marshal(process, tmpFile); |
| result = falconUnitClient.update(EntityType.PROCESS.name(), context.getProcessName(), |
| tmpFile.getAbsolutePath(), true, null); |
| Assert.assertEquals(result.getStatus(), APIResult.Status.SUCCEEDED); |
| |
| //Since the process endtime = update effective time, it shouldn't create new bundle |
| List<BundleJob> bundles = OozieTestUtils.getBundles(context); |
| Assert.assertEquals(bundles.size(), 1); |
| |
| //Since the entity was suspended before update, it should still be suspended |
| Assert.assertEquals(bundles.get(0).getStatus(), Status.SUSPENDED); |
| } |
| |
| @Test |
| public void testProcessInputUpdate() throws Exception { |
| UnitTestContext context = newContext(); |
| |
| schedule(context); |
| waitForStatus(EntityType.PROCESS.name(), context.processName, START_INSTANCE, |
| InstancesResult.WorkflowStatus.RUNNING); |
| List<BundleJob> bundles = OozieTestUtils.getBundles(context); |
| Assert.assertEquals(bundles.size(), 1); |
| OozieClient ozClient = OozieTestUtils.getOozieClient(context); |
| String bundle = bundles.get(0).getId(); |
| String coordId = ozClient.getBundleJobInfo(bundle).getCoordinators().get(0).getId(); |
| |
| Process process = (Process) getDefinition(EntityType.PROCESS, context.processName); |
| |
| String feed3 = "f3" + System.currentTimeMillis(); |
| Map<String, String> overlay = new HashMap<String, String>(); |
| overlay.put("inputFeedName", feed3); |
| overlay.put("cluster", context.clusterName); |
| overlay.put("user", System.getProperty("user.name")); |
| submitFeed(UnitTestContext.FEED_TEMPLATE1, overlay); |
| |
| |
| Input input = new Input(); |
| input.setFeed(feed3); |
| input.setName("inputData2"); |
| input.setStart("today(20,0)"); |
| input.setEnd("today(20,20)"); |
| process.getInputs().getInputs().add(input); |
| |
| updateEndtime(process); |
| Date endTime = getEndTime(); |
| File tmpFile = TestContext.getTempFile(); |
| process.getEntityType().getMarshaller().marshal(process, tmpFile); |
| APIResult result = falconUnitClient.update(EntityType.PROCESS.name(), context.getProcessName(), |
| tmpFile.getAbsolutePath(), true, null); |
| Assert.assertEquals(result.getStatus(), APIResult.Status.SUCCEEDED); |
| |
| //Assert that update creates new bundle and old coord is running |
| bundles = OozieTestUtils.getBundles(context); |
| Assert.assertEquals(bundles.size(), 2); |
| CoordinatorJob coord = ozClient.getCoordJobInfo(coordId); |
| Assert.assertEquals(coord.getStatus(), Status.RUNNING); |
| Assert.assertEquals(coord.getEndTime(), endTime); |
| |
| //Assert on new bundle/coord |
| String newBundle = null; |
| for (BundleJob myBundle : bundles) { |
| if (!myBundle.getId().equals(bundle)) { |
| newBundle = myBundle.getId(); |
| break; |
| } |
| } |
| |
| assert newBundle != null; |
| OozieTestUtils.waitForBundleStart(context, newBundle, Job.Status.RUNNING, Status.PREP); |
| coord = ozClient.getCoordJobInfo(ozClient.getBundleJobInfo(newBundle).getCoordinators().get(0).getId()); |
| Assert.assertTrue(coord.getStatus() == Status.RUNNING || coord.getStatus() == Status.PREP); |
| Assert.assertEquals(coord.getStartTime(), endTime); |
| } |
| |
| @Test |
| public void testProcessEndtimeUpdate() throws Exception { |
| UnitTestContext context = newContext(); |
| schedule(context); |
| OozieTestUtils.waitForBundleStart(context, Job.Status.RUNNING); |
| |
| Process process = (Process) getDefinition(EntityType.PROCESS, context.processName); |
| |
| updateEndtime(process); |
| File tmpFile = TestContext.getTempFile(); |
| process.getEntityType().getMarshaller().marshal(process, tmpFile); |
| APIResult result = falconUnitClient.update(EntityType.PROCESS.name(), context.getProcessName(), |
| tmpFile.getAbsolutePath(), true, null); |
| Assert.assertEquals(result.getStatus(), APIResult.Status.SUCCEEDED); |
| |
| //Assert that update does not create new bundle |
| List<BundleJob> bundles = OozieTestUtils.getBundles(context); |
| Assert.assertEquals(bundles.size(), 1); |
| } |
| |
| @Test |
| public void testTouchEntity() throws Exception { |
| UnitTestContext context = newContext(); |
| schedule(context); |
| OozieTestUtils.waitForBundleStart(context, Job.Status.RUNNING); |
| List<BundleJob> bundles = OozieTestUtils.getBundles(context); |
| Assert.assertEquals(bundles.size(), 1); |
| OozieClient ozClient = OozieTestUtils.getOozieClient(context); |
| String bundle = bundles.get(0).getId(); |
| String coordId = ozClient.getBundleJobInfo(bundle).getCoordinators().get(0).getId(); |
| |
| //Update end time of process required for touch |
| Process process = (Process) getDefinition(EntityType.PROCESS, context.processName); |
| updateEndtime(process); |
| File tmpFile = TestContext.getTempFile(); |
| process.getEntityType().getMarshaller().marshal(process, tmpFile); |
| APIResult result = falconUnitClient.update(EntityType.PROCESS.name(), context.getProcessName(), |
| tmpFile.getAbsolutePath(), true, null); |
| Assert.assertEquals(result.getStatus(), APIResult.Status.SUCCEEDED); |
| bundles = OozieTestUtils.getBundles(context); |
| Assert.assertEquals(bundles.size(), 1); |
| |
| result = falconUnitClient.touch(EntityType.PROCESS.name(), context.getProcessName(), context.colo, true, null); |
| Assert.assertEquals(result.getStatus(), APIResult.Status.SUCCEEDED); |
| OozieTestUtils.waitForBundleStart(context, Status.PREP, Status.RUNNING); |
| |
| //Assert that touch creates new bundle and old coord is running |
| bundles = OozieTestUtils.getBundles(context); |
| Assert.assertEquals(bundles.size(), 2); |
| CoordinatorJob coord = ozClient.getCoordJobInfo(coordId); |
| Assert.assertTrue(coord.getStatus() == Status.RUNNING || coord.getStatus() == Status.SUCCEEDED); |
| |
| //Assert on new bundle/coord |
| String newBundle = null; |
| for (BundleJob myBundle : bundles) { |
| if (!myBundle.getId().equals(bundle)) { |
| newBundle = myBundle.getId(); |
| break; |
| } |
| } |
| |
| assert newBundle != null; |
| coord = ozClient.getCoordJobInfo(ozClient.getBundleJobInfo(newBundle).getCoordinators().get(0).getId()); |
| Assert.assertTrue(coord.getStatus() == Status.RUNNING || coord.getStatus() == Status.PREP); |
| } |
| |
| public void testStatus() throws Exception { |
| UnitTestContext context = newContext(); |
| |
| submitCluster(context); |
| |
| context.prepare(); |
| submitFeed(UnitTestContext.FEED_TEMPLATE1, context.overlay); |
| |
| APIResult result = falconUnitClient.getStatus(EntityType.FEED, context.overlay.get("inputFeedName"), |
| context.colo, null, false); |
| Assert.assertEquals(result.getStatus(), APIResult.Status.SUCCEEDED); |
| Assert.assertEquals(result.getMessage(), "SUBMITTED"); |
| } |
| |
| public void testIdempotentSubmit() throws Exception { |
| UnitTestContext context = newContext(); |
| |
| submitCluster(context); |
| |
| submitCluster(context); |
| } |
| |
| public void testNotFoundStatus() throws FalconException, IOException, FalconCLIException { |
| String feed1 = "f1" + System.currentTimeMillis(); |
| try { |
| falconUnitClient.getStatus(EntityType.FEED, feed1, null, null, false); |
| Assert.fail("Exception should be Thrown"); |
| } catch (Exception e) { |
| //ignore |
| } |
| } |
| |
| public void testVersion() throws FalconException, IOException, FalconCLIException { |
| String json = falconUnitClient.getVersion(null); |
| String buildVersion = BuildProperties.get().getProperty("build.version"); |
| String deployMode = DeploymentProperties.get().getProperty("deploy.mode"); |
| Assert.assertTrue(Pattern.matches( |
| ".*\\{\\s*\"key\"\\s*:\\s*\"Version\"\\s*,\\s*\"value\"\\s*:\\s*\"" |
| + buildVersion + "\"\\s*}.*", json), |
| "No build.version found in /api/admin/version"); |
| Assert.assertTrue(Pattern.matches( |
| ".*\\{\\s*\"key\"\\s*:\\s*\"Mode\"\\s*,\\s*\"value\"\\s*:\\s*\"" |
| + deployMode + "\"\\s*}.*", json), |
| "No deploy.mode found in /api/admin/version"); |
| } |
| |
| public void testValidate() throws FalconException, IOException { |
| UnitTestContext context = newContext(); |
| try { |
| falconUnitClient.validate(EntityType.PROCESS.name(), |
| UnitTestContext.class.getResource(UnitTestContext.SAMPLE_PROCESS_XML).getPath(), true, null); |
| Assert.fail("Exception should be Thrown"); |
| } catch (Exception e) { |
| //ignore |
| } |
| } |
| |
| public void testClusterValidate() throws Exception { |
| UnitTestContext context = newContext(); |
| |
| String tmpFileName = TestContext.overlayParametersOverTemplate(TestContext.CLUSTER_TEMPLATE, context.overlay); |
| File tmpFile = new File(tmpFileName); |
| fs.mkdirs(new Path(STAGING_PATH), HadoopClientFactory.ALL_PERMISSION); |
| fs.mkdirs(new Path(WORKING_PATH), HadoopClientFactory.READ_EXECUTE_PERMISSION); |
| APIResult result = falconUnitClient.validate(EntityType.CLUSTER.name(), tmpFile.getAbsolutePath(), |
| true, null); |
| Assert.assertEquals(result.getStatus(), APIResult.Status.SUCCEEDED); |
| } |
| |
| ClientResponse suspend(TestContext context, Entity entity) { |
| return suspend(context, entity.getEntityType(), entity.getName()); |
| } |
| |
| private ClientResponse suspend(TestContext context, EntityType entityType, String name) { |
| return context.service |
| .path("api/entities/suspend/" + entityType.name().toLowerCase() + "/" + name) |
| .header("Cookie", context.getAuthenticationToken()) |
| .accept(MediaType.TEXT_XML).type(MediaType.TEXT_XML) |
| .post(ClientResponse.class); |
| } |
| |
| public void testClusterSubmitScheduleSuspendResumeDelete() throws Exception { |
| UnitTestContext context = newContext(); |
| |
| submitCluster(context); |
| |
| try { |
| falconUnitClient.schedule(EntityType.CLUSTER, context.clusterName, null, true, null, null); |
| Assert.fail("Exception should be Thrown"); |
| } catch (Exception e) { |
| //ignore |
| } |
| |
| try { |
| falconUnitClient.suspend(EntityType.CLUSTER, context.clusterName, context.colo, null); |
| Assert.fail("Exception should be Thrown"); |
| } catch (Exception e) { |
| //ignore |
| } |
| |
| try { |
| falconUnitClient.resume(EntityType.CLUSTER, context.clusterName, context.colo, null); |
| Assert.fail("Exception should be Thrown"); |
| } catch (Exception e) { |
| //ignore |
| } |
| |
| APIResult result = falconUnitClient.delete(EntityType.CLUSTER, context.getClusterName(), null); |
| Assert.assertEquals(result.getStatus(), APIResult.Status.SUCCEEDED); |
| } |
| |
| public void testSubmit() throws Exception { |
| UnitTestContext context = newContext(); |
| ClientResponse response; |
| |
| submitCluster(context); |
| |
| submitFeed(UnitTestContext.FEED_TEMPLATE1, context.overlay); |
| |
| submitFeed(UnitTestContext.FEED_TEMPLATE2, context.overlay); |
| |
| submitProcess(UnitTestContext.PROCESS_TEMPLATE, context.overlay); |
| } |
| |
| @Test |
| public void testDuplicateSubmitCommands() throws Exception { |
| UnitTestContext context = newContext(); |
| |
| submitCluster(context); |
| |
| ExecutorService service = Executors.newSingleThreadExecutor(); |
| ExecutorService duplicateService = Executors.newSingleThreadExecutor(); |
| |
| Future<APIResult> future = service.submit(new SubmitCommand(context, context.overlay)); |
| Future<APIResult> duplicateFuture = duplicateService.submit(new SubmitCommand(context, context.overlay)); |
| |
| // since there are duplicate threads for submits, there is no guarantee which request will succeed. |
| try { |
| APIResult response = future.get(); |
| APIResult duplicateSubmitThreadResponse = duplicateFuture.get(); |
| Assert.fail("Exception should be Thrown"); |
| } catch (Exception e) { |
| //ignore |
| } |
| } |
| |
| @Test |
| public void testDuplicateDeleteCommands() throws Exception { |
| UnitTestContext context = newContext(); |
| Map<String, String> overlay = context.overlay; |
| submitCluster(context); |
| submitFeed(UnitTestContext.FEED_TEMPLATE1, overlay); |
| |
| ExecutorService service = Executors.newFixedThreadPool(2); |
| |
| Future<APIResult> future = service.submit(new DeleteCommand(context, overlay.get("inputFeedName"), |
| "feed")); |
| Future<APIResult> duplicateFuture = service.submit(new DeleteCommand(context, |
| overlay.get("inputFeedName"), "feed")); |
| |
| // since there are duplicate threads for submits, there is no guarantee which request will succeed. |
| try { |
| APIResult response = future.get(); |
| APIResult duplicateSubmitThreadResponse = duplicateFuture.get(); |
| Assert.fail("Exception should be Thrown"); |
| } catch (Exception e) { |
| //ignore |
| } |
| } |
| |
| public void testProcesssScheduleAndDelete() throws Exception { |
| scheduleAndDeleteProcess(false); |
| } |
| |
| public void testProcesssScheduleAndDeleteWithDoAs() throws Exception { |
| scheduleAndDeleteProcess(true); |
| } |
| |
| private void scheduleAndDeleteProcess(boolean withDoAs) throws Exception { |
| UnitTestContext context = newContext(); |
| submitCluster(context); |
| context.prepare(); |
| submitFeeds(context.overlay); |
| ClientResponse clientResponse; |
| String tmpFileName = TestContext.overlayParametersOverTemplate(TestContext.PROCESS_TEMPLATE, context.overlay); |
| Process process = (Process) EntityType.PROCESS.getUnmarshaller().unmarshal(new File(tmpFileName)); |
| updateEndtime(process); |
| File tmpFile = TestContext.getTempFile(); |
| EntityType.PROCESS.getMarshaller().marshal(process, tmpFile); |
| submitProcess(tmpFile.getAbsolutePath(), context.overlay); |
| if (withDoAs) { |
| falconUnitClient.schedule(EntityType.PROCESS, context.getProcessName(), context.getClusterName(), false, |
| FalconTestUtil.TEST_USER_2, null); |
| } else { |
| falconUnitClient.schedule(EntityType.PROCESS, context.getProcessName(), context.getClusterName(), false, "", |
| "key1:value1"); |
| } |
| OozieTestUtils.waitForBundleStart(context, Status.RUNNING); |
| |
| APIResult result; |
| if (withDoAs) { |
| result = falconUnitClient.delete(EntityType.PROCESS, context.getProcessName(), FalconTestUtil.TEST_USER_2); |
| } else { |
| result = falconUnitClient.delete(EntityType.PROCESS, context.getProcessName(), null); |
| } |
| assertStatus(result); |
| } |
| |
| public void testGetEntityDefinition() throws Exception { |
| UnitTestContext context = newContext(); |
| |
| submitCluster(context); |
| |
| context.prepare(); |
| APIResult result = submitFeed(UnitTestContext.FEED_TEMPLATE1, context.overlay); |
| Assert.assertEquals(result.getStatus(), APIResult.Status.SUCCEEDED); |
| |
| Feed feed = (Feed) falconUnitClient.getDefinition(EntityType.FEED.name(), |
| context.overlay.get("inputFeedName"), null); |
| Assert.assertEquals(feed.getName(), context.overlay.get("inputFeedName")); |
| } |
| |
| public void testInvalidGetEntityDefinition() throws FalconException, IOException, FalconCLIException { |
| try { |
| falconUnitClient.getDefinition(EntityType.PROCESS.name(), "sample1", null); |
| Assert.fail("Exception should be Thrown"); |
| } catch (Exception e) { |
| //ignore |
| } |
| } |
| |
| public void testScheduleSuspendResume() throws Exception { |
| UnitTestContext context = newContext(); |
| |
| schedule(context); |
| waitForStatus(EntityType.PROCESS.name(), context.processName, START_INSTANCE, |
| InstancesResult.WorkflowStatus.RUNNING); |
| |
| APIResult result = falconUnitClient.suspend(EntityType.PROCESS, context.getProcessName(), |
| context.colo, null); |
| Assert.assertEquals(result.getStatus(), APIResult.Status.SUCCEEDED); |
| result = falconUnitClient.getStatus(EntityType.PROCESS, context.processName, context.clusterName, |
| null, false); |
| Assert.assertEquals(result.getStatus(), APIResult.Status.SUCCEEDED); |
| Assert.assertEquals(result.getMessage(), "SUSPENDED"); |
| |
| result = falconUnitClient.resume(EntityType.PROCESS, context.getProcessName(), |
| context.colo, null); |
| Assert.assertEquals(result.getStatus(), APIResult.Status.SUCCEEDED); |
| result = falconUnitClient.getStatus(EntityType.PROCESS, context.processName, context.clusterName, |
| null, false); |
| Assert.assertEquals(result.getStatus(), APIResult.Status.SUCCEEDED); |
| Assert.assertEquals(result.getMessage(), "RUNNING"); |
| } |
| |
| public void testFeedSchedule() throws Exception { |
| UnitTestContext context = newContext(); |
| |
| submitCluster(context); |
| |
| context.prepare(); |
| APIResult result = submitFeed(UnitTestContext.FEED_TEMPLATE1, context.overlay); |
| Assert.assertEquals(result.getStatus(), APIResult.Status.SUCCEEDED); |
| |
| createTestData(); |
| result = falconUnitClient.schedule(EntityType.FEED, context.overlay.get("inputFeedName"), null, true, null, |
| null); |
| Assert.assertEquals(result.getStatus(), APIResult.Status.SUCCEEDED); |
| } |
| |
| public void testDeleteDataSet() throws Exception { |
| UnitTestContext context = newContext(); |
| |
| submitCluster(context); |
| |
| context.prepare(); |
| APIResult result = submitFeed(UnitTestContext.FEED_TEMPLATE1, context.overlay); |
| Assert.assertEquals(result.getStatus(), APIResult.Status.SUCCEEDED); |
| |
| result = falconUnitClient.delete(EntityType.FEED, context.overlay.get("inputFeedName"), null); |
| Assert.assertEquals(result.getStatus(), APIResult.Status.SUCCEEDED); |
| } |
| |
| public void testDelete() throws Exception { |
| UnitTestContext context = newContext(); |
| |
| submitCluster(context); |
| |
| context.prepare(); |
| APIResult result = submitFeed(UnitTestContext.FEED_TEMPLATE1, context.overlay); |
| Assert.assertEquals(result.getStatus(), APIResult.Status.SUCCEEDED); |
| |
| try { |
| falconUnitClient.delete(EntityType.CLUSTER, context.getClusterName(), null); |
| Assert.fail("Exception should be Thrown"); |
| } catch (Exception e) { |
| //ignore |
| } |
| |
| result = submitFeed(UnitTestContext.FEED_TEMPLATE2, context.overlay); |
| Assert.assertEquals(result.getStatus(), APIResult.Status.SUCCEEDED); |
| |
| submitProcess(UnitTestContext.PROCESS_TEMPLATE, context.overlay); |
| |
| //Delete a referred feed |
| try { |
| falconUnitClient.delete(EntityType.FEED, context.overlay.get("inputFeedName"), null); |
| Assert.fail("Exception should be Thrown"); |
| } catch (Exception e) { |
| //ignore |
| } |
| |
| //Delete a submitted process |
| result = falconUnitClient.delete(EntityType.PROCESS, context.getProcessName(), null); |
| Assert.assertEquals(result.getStatus(), APIResult.Status.SUCCEEDED); |
| |
| submitProcess(UnitTestContext.PROCESS_TEMPLATE, context.overlay); |
| |
| result = falconUnitClient.schedule(EntityType.PROCESS, context.getProcessName(), null, true, null, |
| null); |
| Assert.assertEquals(result.getStatus(), APIResult.Status.SUCCEEDED); |
| |
| //Delete a scheduled process |
| result = falconUnitClient.delete(EntityType.PROCESS, context.getProcessName(), null); |
| Assert.assertEquals(result.getStatus(), APIResult.Status.SUCCEEDED); |
| } |
| |
| @Test |
| public void testGetEntityList() throws Exception { |
| EntityList result = falconUnitClient.getEntityList(EntityType.PROCESS.name(), "", "", null, null, |
| null, null, null, new Integer(0), new Integer(1), null); |
| Assert.assertNotNull(result); |
| for (EntityList.EntityElement entityElement : result.getElements()) { |
| Assert.assertNull(entityElement.status); // status is null |
| } |
| |
| result = falconUnitClient.getEntityList(EntityType.CLUSTER.name(), "", "", null, null, |
| null, null, null, new Integer(0), new Integer(1), null); |
| Assert.assertNotNull(result); |
| for (EntityList.EntityElement entityElement : result.getElements()) { |
| Assert.assertNull(entityElement.status); // status is null |
| } |
| |
| result = falconUnitClient.getEntityList(EntityType.FEED.name() + "," + EntityType.PROCESS.name(), |
| "", "", null, null, null, null, null, new Integer(0), new Integer(1), null); |
| Assert.assertNotNull(result); |
| for (EntityList.EntityElement entityElement : result.getElements()) { |
| Assert.assertNull(entityElement.status); // status is null |
| } |
| |
| result = falconUnitClient.getEntityList(null, "", "", null, null, null, null, null, new Integer(0), |
| new Integer(1), null); |
| Assert.assertNotNull(result); |
| for (EntityList.EntityElement entityElement : result.getElements()) { |
| Assert.assertNull(entityElement.status); // status is null |
| } |
| } |
| |
| @Test |
| public void testDuplicateUpdateCommands() throws Exception { |
| UnitTestContext context = newContext(); |
| schedule(context); |
| OozieTestUtils.waitForBundleStart(context, Job.Status.RUNNING); |
| List<BundleJob> bundles = OozieTestUtils.getBundles(context); |
| Assert.assertEquals(bundles.size(), 1); |
| |
| Process process = (Process) getDefinition(EntityType.PROCESS, context.processName); |
| |
| String feed3 = "f3" + System.currentTimeMillis(); |
| Map<String, String> overlay = new HashMap<String, String>(); |
| overlay.put("inputFeedName", feed3); |
| overlay.put("cluster", context.clusterName); |
| overlay.put("user", System.getProperty("user.name")); |
| APIResult result = submitFeed(UnitTestContext.FEED_TEMPLATE1, context.overlay); |
| Assert.assertEquals(result.getStatus(), APIResult.Status.SUCCEEDED); |
| |
| Input input = new Input(); |
| input.setFeed(feed3); |
| input.setName("inputData2"); |
| input.setStart("today(20,0)"); |
| input.setEnd("today(20,20)"); |
| process.getInputs().getInputs().add(input); |
| |
| updateEndtime(process); |
| Date endTime = getEndTime(); |
| ExecutorService service = Executors.newSingleThreadExecutor(); |
| ExecutorService duplicateService = Executors.newSingleThreadExecutor(); |
| |
| Future<APIResult> future = service.submit(new UpdateCommand(context, process, endTime)); |
| Future<APIResult> duplicateFuture = duplicateService.submit(new UpdateCommand(context, process, endTime)); |
| |
| // since there are duplicate threads for updates, there is no guarantee which request will succeed |
| try { |
| future.get(); |
| duplicateFuture.get(); |
| Assert.fail("Exception should be Thrown"); |
| } catch (Exception e) { |
| //ignore |
| } |
| } |
| |
| public Date getEndTime() { |
| Calendar cal = Calendar.getInstance(TimeZone.getTimeZone("UTC")); |
| cal.add(Calendar.DAY_OF_MONTH, 1); |
| cal.set(Calendar.HOUR_OF_DAY, 0); |
| cal.set(Calendar.MINUTE, 0); |
| cal.set(Calendar.SECOND, 0); |
| cal.set(Calendar.MILLISECOND, 0); |
| return cal.getTime(); |
| } |
| |
| class UpdateCommand implements Callable<APIResult> { |
| private UnitTestContext context; |
| private Process process; |
| private Date endTime; |
| |
| public UnitTestContext getContext() { |
| return context; |
| } |
| public Process getProcess() { |
| return process; |
| } |
| |
| public UpdateCommand(UnitTestContext context, Process process, Date endTime) { |
| this.context = context; |
| this.process = process; |
| this.endTime = endTime; |
| } |
| |
| @Override |
| public APIResult call() throws Exception { |
| File tmpFile = TestContext.getTempFile(); |
| process.getEntityType().getMarshaller().marshal(process, tmpFile); |
| return falconUnitClient.update(EntityType.PROCESS.name(), context.getProcessName(), |
| tmpFile.getAbsolutePath(), true, null); |
| } |
| } |
| |
| class SubmitCommand implements Callable<APIResult> { |
| private Map<String, String> overlay; |
| private UnitTestContext context; |
| |
| public UnitTestContext getContext() { |
| return context; |
| } |
| |
| public Map<String, String> getOverlay() { |
| return overlay; |
| } |
| |
| public SubmitCommand(UnitTestContext context, Map<String, String> overlay) { |
| this.context = context; |
| this.overlay = overlay; |
| } |
| |
| @Override |
| public APIResult call() throws Exception { |
| return submitFeed(UnitTestContext.FEED_TEMPLATE1, context.overlay); |
| } |
| } |
| |
| class DeleteCommand implements Callable<APIResult> { |
| private UnitTestContext context; |
| private String entityName; |
| private String entityType; |
| |
| public UnitTestContext getContext() { |
| return context; |
| } |
| |
| public DeleteCommand(UnitTestContext context, String entityName, String entityType) { |
| this.context = context; |
| this.entityName = entityName; |
| this.entityType = entityType; |
| } |
| |
| @Override |
| public APIResult call() throws Exception { |
| return falconUnitClient.delete(EntityType.valueOf(entityType), entityName, null); |
| } |
| } |
| } |
