core/src/test/java/org/apache/oozie/TestLocalOozieClientCoord.java - oozie - Git at Google

 /**
  * 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;

 import java.io.File;
 import java.io.FileOutputStream;
 import java.io.IOException;
 import java.io.OutputStreamWriter;
 import java.io.PrintWriter;
 import java.net.URI;
 import java.nio.charset.StandardCharsets;
 import java.util.Iterator;
 import java.util.List;
 import java.util.NoSuchElementException;
 import java.util.Properties;

 import org.apache.commons.lang3.StringUtils;
 import org.apache.oozie.client.CoordinatorJob;
 import org.apache.oozie.client.Job;
 import org.apache.oozie.client.OozieClient;
 import org.apache.oozie.client.OozieClientException;
 import org.apache.oozie.local.LocalOozie;
 import org.apache.oozie.service.Services;
 import org.apache.oozie.test.XDataTestCase;
 import org.json.simple.JSONArray;
 import org.json.simple.JSONObject;

 public class TestLocalOozieClientCoord extends XDataTestCase {

     private Services services;

     @Override
     protected void setUp() throws Exception {
         super.setUp();

         /*
          * DB cleanup below is needed to clean up the job records possibly left by previously executed tests.
          * For example, such records are left by test org.apache.oozie.executor.jpa.TestCoordActionUpdateJPAExecutor.
          * Note that by default Oozie tests are executed in "filesystem" (in fact, arbitrary) order. This way problems caused
          * by left records can be flaky (not reproduced constantly).
          * Services re-init is needed for the DB cleanup.
          */
         services = new Services();
         services.init();

         LocalOozie.start();
     }

     @Override
     protected void tearDown() throws Exception {
         LocalOozie.stop();
         services.destroy();
         super.tearDown();
     }

     public void testGetOozieUrl() {
         OozieClient client = LocalOozie.getCoordClient();
         assertEquals("localoozie", client.getOozieUrl());
     }

     public void testGetProtocolUrl() throws IOException, OozieClientException {
         OozieClient client = LocalOozie.getCoordClient();
         assertEquals("localoozie", client.getProtocolUrl());
     }

     public void testValidateWSVersion() throws IOException, OozieClientException {
         OozieClient client = LocalOozie.getCoordClient();
         client.validateWSVersion();
     }

     public void testHeaderMethods() {
         OozieClient client = LocalOozie.getCoordClient();
         client.setHeader("h", "v");
         assertTrue("no-op, should be null/empty", StringUtils.isBlank(client.getHeader("h")));
         Iterator<String> hit = client.getHeaderNames();
         assertFalse(hit.hasNext());
         try {
             hit.next();
             fail("NoSuchElementException expected.");
         }
         catch (NoSuchElementException nsee) {
             // expected
         }
         client.removeHeader("h");
         assertTrue("no-op, should be null/empty", StringUtils.isBlank(client.getHeader("h")));
     }

     public void testGetJobsInfo() {
         OozieClient client = LocalOozie.getCoordClient();
         try {
             client.getJobsInfo("foo");
             fail("OozieClientException expected.");
         }
         catch (OozieClientException oce) {
             assertEquals(ErrorCode.E0301.toString(), oce.getErrorCode());
         }
         try {
             client.getJobsInfo("foo", 0, 5);
             fail("OozieClientException expected.");
         }
         catch (OozieClientException oce) {
             assertEquals(ErrorCode.E0301.toString(), oce.getErrorCode());
         }
         try {
             client.getJobInfo("foo-id");
             fail("OozieClientException expected.");
         }
         catch (OozieClientException oce) {
             assertEquals(ErrorCode.E0301.toString(), oce.getErrorCode());
         }
     }

     public void testReRun2() {
         OozieClient client = LocalOozie.getCoordClient();
         try {
             client.reRun("foo-id", client.createConfiguration());
             fail("OozieClientException expected.");
         }
         catch (OozieClientException oce) {
             assertEquals(ErrorCode.E0301.toString(), oce.getErrorCode());
         }
     }

     private void writeToFile(String appXml, String appPath) throws Exception {
         File wf = new File(new URI(appPath).getPath());
         PrintWriter out = null;
         try {
             out = new PrintWriter(new OutputStreamWriter(new FileOutputStream(wf), StandardCharsets.UTF_8));
             out.println(appXml);
         }
         catch (IOException iex) {
             throw iex;
         }
         finally {
             if (out != null) {
                 out.close();
             }
         }
     }

     public void testJobMethods() throws Exception {
         final OozieClient client = LocalOozie.getCoordClient();

         // Just in case, check that there are no Coord job records left by previous tests:
         List<CoordinatorJob> list0 = client.getCoordJobsInfo("", 1, 100);
         assertEquals(0, list0.size());

         Properties conf = client.createConfiguration();

         String appPath = getTestCaseFileUri("coordinator.xml");
         String appXml = "<coordinator-app name=\"NAME\" frequency=\"${coord:minutes(20)}\" "
                 + "start=\"2009-02-01T01:00Z\" end=\"2009-02-03T23:59Z\" timezone=\"UTC\" "
                 + "xmlns=\"uri:oozie:coordinator:0.1\"> <controls> <timeout>10</timeout> <concurrency>1</concurrency> "
                 + "<execution>LIFO</execution> </controls> <datasets> "
                 + "<dataset name=\"a\" frequency=\"${coord:minutes(20)}\" initial-instance=\"2009-02-01T01:00Z\" "
                 + "timezone=\"UTC\"> <uri-template>" + getTestCaseFileUri("coord/workflows/${YEAR}/${DAY}")
                 + "</uri-template> </dataset> "
                 + "<dataset name=\"local_a\" frequency=\"${coord:minutes(20)}\" initial-instance=\"2009-02-01T01:00Z\" "
                 + "timezone=\"UTC\"> <uri-template>" + getTestCaseFileUri("coord/workflows/${YEAR}/${DAY}")
                 + "</uri-template> </dataset> "
                 + "</datasets> <input-events> "
                 + "<data-in name=\"A\" dataset=\"a\"> <instance>${coord:latest(0)}</instance> </data-in>  " + "</input-events> "
                 + "<output-events> <data-out name=\"LOCAL_A\" dataset=\"local_a\"> "
                 + "<instance>${coord:current(-1)}</instance> </data-out> </output-events> <action> <workflow> "
                 + "<app-path>hdfs:///tmp/workflows/</app-path> "
                 + "<configuration> <property> <name>inputA</name> <value>${coord:dataIn('A')}</value> </property> "
                 + "<property> <name>inputB</name> <value>${coord:dataOut('LOCAL_A')}</value> "
                 + "</property></configuration> </workflow> </action> </coordinator-app>";
         writeToFile(appXml, appPath);

         conf.setProperty(OozieClient.COORDINATOR_APP_PATH, appPath);
         String jobId0 = client.submit(conf);
         client.kill(jobId0);

         String jobId = client.run(conf);
         client.suspend(jobId);
         client.resume(jobId);
         client.kill(jobId);

         CoordinatorJob job = client.getCoordJobInfo(jobId);
         String appName = job.getAppName();
         assertEquals("NAME", appName);

         List<CoordinatorJob> list = client.getCoordJobsInfo("", 1, 5);
         assertEquals(2, list.size());
     }

     public void testJobsOperations() throws Exception {
         final OozieClient client = LocalOozie.getCoordClient();

         // Just in case, check that there are no Coord job records left by previous tests:
         List<CoordinatorJob> list0 = client.getCoordJobsInfo("", 1, 100);
         assertEquals(0, list0.size());
         Properties conf = client.createConfiguration();
         String appPath = storedCoordAppPath();

         conf.setProperty(OozieClient.COORDINATOR_APP_PATH, appPath);
         final String jobId0 = client.run(conf);
         final String jobId1 = client.run(conf);
         final String jobId2 = client.run(conf);
         waitFor(client, jobId0);
         waitFor(client, jobId1);
         waitFor(client, jobId2);
         list0 = client.getCoordJobsInfo("name=NAME", 1, 10);
         assertEquals(3, list0.size());

         JSONObject jsonObject = client.suspendJobs("name=NAME", "coord", 1, 3);
         assertEquals(3, jsonObject.get("total"));
         assertEquals(3, ((JSONArray) jsonObject.get("coordinatorjobs")).size());
         assertEquals(Job.Status.SUSPENDED, client.getCoordJobInfo(jobId0).getStatus());
         assertEquals(Job.Status.SUSPENDED, client.getCoordJobInfo(jobId1).getStatus());
         assertEquals(Job.Status.SUSPENDED, client.getCoordJobInfo(jobId2).getStatus());

         jsonObject = client.resumeJobs("name=NAME", "coord", 1, 3);
         assertEquals(3, jsonObject.get("total"));
         assertEquals(3, ((JSONArray) jsonObject.get("coordinatorjobs")).size());
         assertEquals(Job.Status.RUNNING, client.getCoordJobInfo(jobId0).getStatus());
         assertEquals(Job.Status.RUNNING, client.getCoordJobInfo(jobId1).getStatus());
         assertEquals(Job.Status.RUNNING, client.getCoordJobInfo(jobId2).getStatus());

         jsonObject = client.killJobs("name=NAME", "coord", 1, 3);
         assertEquals(3, jsonObject.get("total"));
         assertEquals(3, ((JSONArray) jsonObject.get("coordinatorjobs")).size());
         assertEquals(Job.Status.KILLED, client.getCoordJobInfo(jobId0).getStatus());
         assertEquals(Job.Status.KILLED, client.getCoordJobInfo(jobId1).getStatus());
         assertEquals(Job.Status.KILLED, client.getCoordJobInfo(jobId2).getStatus());
     }

     private void waitFor(final OozieClient client, final String jobId) {
         waitFor(10 * 1000, new Predicate() {
             @Override
             public boolean evaluate() throws Exception {
                 Job.Status status = client.getCoordJobInfo(jobId).getStatus();
                 return status != Job.Status.PREP;
             }
         });
     }

     private String storedCoordAppPath() throws Exception {
         String appPath = getTestCaseFileUri("coordinator.xml");
         String appXml = "<coordinator-app name=\"NAME\" frequency=\"${coord:minutes(20)}\" "
                 + "start=\"2009-02-01T01:00Z\" end=\"2009-02-01T03:00Z\" timezone=\"UTC\" "
                 + "xmlns=\"uri:oozie:coordinator:0.1\"> <controls> <timeout>10</timeout> <concurrency>1</concurrency> "
                 + "<execution>LIFO</execution> </controls> "
                 + " <action> <workflow> "
                 + "<app-path>hdfs:///tmp/workflows/</app-path> "
                 +" </workflow> </action> </coordinator-app>";
         writeToFile(appXml, appPath);
         return appPath;
     }
 }
	/**
	* 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;

	import java.io.File;
	import java.io.FileOutputStream;
	import java.io.IOException;
	import java.io.OutputStreamWriter;
	import java.io.PrintWriter;
	import java.net.URI;
	import java.nio.charset.StandardCharsets;
	import java.util.Iterator;
	import java.util.List;
	import java.util.NoSuchElementException;
	import java.util.Properties;

	import org.apache.commons.lang3.StringUtils;
	import org.apache.oozie.client.CoordinatorJob;
	import org.apache.oozie.client.Job;
	import org.apache.oozie.client.OozieClient;
	import org.apache.oozie.client.OozieClientException;
	import org.apache.oozie.local.LocalOozie;
	import org.apache.oozie.service.Services;
	import org.apache.oozie.test.XDataTestCase;
	import org.json.simple.JSONArray;
	import org.json.simple.JSONObject;

	public class TestLocalOozieClientCoord extends XDataTestCase {

	private Services services;

	@Override
	protected void setUp() throws Exception {
	super.setUp();

	/*
	* DB cleanup below is needed to clean up the job records possibly left by previously executed tests.
	* For example, such records are left by test org.apache.oozie.executor.jpa.TestCoordActionUpdateJPAExecutor.
	* Note that by default Oozie tests are executed in "filesystem" (in fact, arbitrary) order. This way problems caused
	* by left records can be flaky (not reproduced constantly).
	* Services re-init is needed for the DB cleanup.
	*/
	services = new Services();
	services.init();

	LocalOozie.start();
	}

	@Override
	protected void tearDown() throws Exception {
	LocalOozie.stop();
	services.destroy();
	super.tearDown();
	}

	public void testGetOozieUrl() {
	OozieClient client = LocalOozie.getCoordClient();
	assertEquals("localoozie", client.getOozieUrl());
	}

	public void testGetProtocolUrl() throws IOException, OozieClientException {
	OozieClient client = LocalOozie.getCoordClient();
	assertEquals("localoozie", client.getProtocolUrl());
	}

	public void testValidateWSVersion() throws IOException, OozieClientException {
	OozieClient client = LocalOozie.getCoordClient();
	client.validateWSVersion();
	}

	public void testHeaderMethods() {
	OozieClient client = LocalOozie.getCoordClient();
	client.setHeader("h", "v");
	assertTrue("no-op, should be null/empty", StringUtils.isBlank(client.getHeader("h")));
	Iterator<String> hit = client.getHeaderNames();
	assertFalse(hit.hasNext());
	try {
	hit.next();
	fail("NoSuchElementException expected.");
	}
	catch (NoSuchElementException nsee) {
	// expected
	}
	client.removeHeader("h");
	assertTrue("no-op, should be null/empty", StringUtils.isBlank(client.getHeader("h")));
	}

	public void testGetJobsInfo() {
	OozieClient client = LocalOozie.getCoordClient();
	try {
	client.getJobsInfo("foo");
	fail("OozieClientException expected.");
	}
	catch (OozieClientException oce) {
	assertEquals(ErrorCode.E0301.toString(), oce.getErrorCode());
	}
	try {
	client.getJobsInfo("foo", 0, 5);
	fail("OozieClientException expected.");
	}
	catch (OozieClientException oce) {
	assertEquals(ErrorCode.E0301.toString(), oce.getErrorCode());
	}
	try {
	client.getJobInfo("foo-id");
	fail("OozieClientException expected.");
	}
	catch (OozieClientException oce) {
	assertEquals(ErrorCode.E0301.toString(), oce.getErrorCode());
	}
	}

	public void testReRun2() {
	OozieClient client = LocalOozie.getCoordClient();
	try {
	client.reRun("foo-id", client.createConfiguration());
	fail("OozieClientException expected.");
	}
	catch (OozieClientException oce) {
	assertEquals(ErrorCode.E0301.toString(), oce.getErrorCode());
	}
	}

	private void writeToFile(String appXml, String appPath) throws Exception {
	File wf = new File(new URI(appPath).getPath());
	PrintWriter out = null;
	try {
	out = new PrintWriter(new OutputStreamWriter(new FileOutputStream(wf), StandardCharsets.UTF_8));
	out.println(appXml);
	}
	catch (IOException iex) {
	throw iex;
	}
	finally {
	if (out != null) {
	out.close();
	}
	}
	}

	public void testJobMethods() throws Exception {
	final OozieClient client = LocalOozie.getCoordClient();

	// Just in case, check that there are no Coord job records left by previous tests:
	List<CoordinatorJob> list0 = client.getCoordJobsInfo("", 1, 100);
	assertEquals(0, list0.size());

	Properties conf = client.createConfiguration();

	String appPath = getTestCaseFileUri("coordinator.xml");
	String appXml = "<coordinator-app name=\"NAME\" frequency=\"${coord:minutes(20)}\" "
	+ "start=\"2009-02-01T01:00Z\" end=\"2009-02-03T23:59Z\" timezone=\"UTC\" "
	+ "xmlns=\"uri:oozie:coordinator:0.1\"> <controls> <timeout>10</timeout> <concurrency>1</concurrency> "
	+ "<execution>LIFO</execution> </controls> <datasets> "
	+ "<dataset name=\"a\" frequency=\"${coord:minutes(20)}\" initial-instance=\"2009-02-01T01:00Z\" "
	+ "timezone=\"UTC\"> <uri-template>" + getTestCaseFileUri("coord/workflows/${YEAR}/${DAY}")
	+ "</uri-template> </dataset> "
	+ "<dataset name=\"local_a\" frequency=\"${coord:minutes(20)}\" initial-instance=\"2009-02-01T01:00Z\" "
	+ "timezone=\"UTC\"> <uri-template>" + getTestCaseFileUri("coord/workflows/${YEAR}/${DAY}")
	+ "</uri-template> </dataset> "
	+ "</datasets> <input-events> "
	+ "<data-in name=\"A\" dataset=\"a\"> <instance>${coord:latest(0)}</instance> </data-in> " + "</input-events> "
	+ "<output-events> <data-out name=\"LOCAL_A\" dataset=\"local_a\"> "
	+ "<instance>${coord:current(-1)}</instance> </data-out> </output-events> <action> <workflow> "
	+ "<app-path>hdfs:///tmp/workflows/</app-path> "
	+ "<configuration> <property> <name>inputA</name> <value>${coord:dataIn('A')}</value> </property> "
	+ "<property> <name>inputB</name> <value>${coord:dataOut('LOCAL_A')}</value> "
	+ "</property></configuration> </workflow> </action> </coordinator-app>";
	writeToFile(appXml, appPath);

	conf.setProperty(OozieClient.COORDINATOR_APP_PATH, appPath);
	String jobId0 = client.submit(conf);
	client.kill(jobId0);

	String jobId = client.run(conf);
	client.suspend(jobId);
	client.resume(jobId);
	client.kill(jobId);

	CoordinatorJob job = client.getCoordJobInfo(jobId);
	String appName = job.getAppName();
	assertEquals("NAME", appName);

	List<CoordinatorJob> list = client.getCoordJobsInfo("", 1, 5);
	assertEquals(2, list.size());
	}

	public void testJobsOperations() throws Exception {
	final OozieClient client = LocalOozie.getCoordClient();

	// Just in case, check that there are no Coord job records left by previous tests:
	List<CoordinatorJob> list0 = client.getCoordJobsInfo("", 1, 100);
	assertEquals(0, list0.size());
	Properties conf = client.createConfiguration();
	String appPath = storedCoordAppPath();

	conf.setProperty(OozieClient.COORDINATOR_APP_PATH, appPath);
	final String jobId0 = client.run(conf);
	final String jobId1 = client.run(conf);
	final String jobId2 = client.run(conf);
	waitFor(client, jobId0);
	waitFor(client, jobId1);
	waitFor(client, jobId2);
	list0 = client.getCoordJobsInfo("name=NAME", 1, 10);
	assertEquals(3, list0.size());

	JSONObject jsonObject = client.suspendJobs("name=NAME", "coord", 1, 3);
	assertEquals(3, jsonObject.get("total"));
	assertEquals(3, ((JSONArray) jsonObject.get("coordinatorjobs")).size());
	assertEquals(Job.Status.SUSPENDED, client.getCoordJobInfo(jobId0).getStatus());
	assertEquals(Job.Status.SUSPENDED, client.getCoordJobInfo(jobId1).getStatus());
	assertEquals(Job.Status.SUSPENDED, client.getCoordJobInfo(jobId2).getStatus());

	jsonObject = client.resumeJobs("name=NAME", "coord", 1, 3);
	assertEquals(3, jsonObject.get("total"));
	assertEquals(3, ((JSONArray) jsonObject.get("coordinatorjobs")).size());
	assertEquals(Job.Status.RUNNING, client.getCoordJobInfo(jobId0).getStatus());
	assertEquals(Job.Status.RUNNING, client.getCoordJobInfo(jobId1).getStatus());
	assertEquals(Job.Status.RUNNING, client.getCoordJobInfo(jobId2).getStatus());

	jsonObject = client.killJobs("name=NAME", "coord", 1, 3);
	assertEquals(3, jsonObject.get("total"));
	assertEquals(3, ((JSONArray) jsonObject.get("coordinatorjobs")).size());
	assertEquals(Job.Status.KILLED, client.getCoordJobInfo(jobId0).getStatus());
	assertEquals(Job.Status.KILLED, client.getCoordJobInfo(jobId1).getStatus());
	assertEquals(Job.Status.KILLED, client.getCoordJobInfo(jobId2).getStatus());
	}

	private void waitFor(final OozieClient client, final String jobId) {
	waitFor(10 * 1000, new Predicate() {
	@Override
	public boolean evaluate() throws Exception {
	Job.Status status = client.getCoordJobInfo(jobId).getStatus();
	return status != Job.Status.PREP;
	}
	});
	}

	private String storedCoordAppPath() throws Exception {
	String appPath = getTestCaseFileUri("coordinator.xml");
	String appXml = "<coordinator-app name=\"NAME\" frequency=\"${coord:minutes(20)}\" "
	+ "start=\"2009-02-01T01:00Z\" end=\"2009-02-01T03:00Z\" timezone=\"UTC\" "
	+ "xmlns=\"uri:oozie:coordinator:0.1\"> <controls> <timeout>10</timeout> <concurrency>1</concurrency> "
	+ "<execution>LIFO</execution> </controls> "
	+ " <action> <workflow> "
	+ "<app-path>hdfs:///tmp/workflows/</app-path> "
	+" </workflow> </action> </coordinator-app>";
	writeToFile(appXml, appPath);
	return appPath;
	}
	}