| /** |
| * 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.action.hadoop; |
| |
| import java.io.File; |
| import java.io.FileInputStream; |
| import java.io.FileNotFoundException; |
| import java.io.FileOutputStream; |
| import java.io.FileWriter; |
| import java.io.IOException; |
| import java.io.InputStream; |
| import java.io.OutputStream; |
| import java.io.OutputStreamWriter; |
| import java.io.StringReader; |
| import java.io.Writer; |
| import java.net.URI; |
| import java.nio.charset.StandardCharsets; |
| import java.util.Arrays; |
| import java.util.List; |
| import java.util.Map; |
| import java.util.Properties; |
| import java.util.Scanner; |
| import java.util.jar.JarOutputStream; |
| import java.util.regex.Pattern; |
| import java.util.zip.ZipEntry; |
| |
| import org.apache.hadoop.conf.Configuration; |
| import org.apache.hadoop.filecache.DistributedCache; |
| import org.apache.hadoop.fs.FSDataInputStream; |
| import org.apache.hadoop.fs.FileStatus; |
| import org.apache.hadoop.fs.FileSystem; |
| import org.apache.hadoop.fs.Path; |
| import org.apache.hadoop.fs.PathFilter; |
| import org.apache.hadoop.io.SequenceFile; |
| import org.apache.hadoop.io.Text; |
| import org.apache.hadoop.mapred.JobClient; |
| import org.apache.hadoop.mapred.JobID; |
| import org.apache.hadoop.mapred.RunningJob; |
| import org.apache.hadoop.mapreduce.TypeConverter; |
| import org.apache.hadoop.streaming.StreamJob; |
| import org.apache.hadoop.yarn.api.records.ApplicationId; |
| import org.apache.hadoop.yarn.api.records.ApplicationReport; |
| import org.apache.hadoop.yarn.client.api.YarnClient; |
| import org.apache.hadoop.yarn.exceptions.YarnException; |
| import org.apache.hadoop.yarn.util.ConverterUtils; |
| import org.apache.oozie.WorkflowActionBean; |
| import org.apache.oozie.WorkflowJobBean; |
| import org.apache.oozie.action.ActionExecutor; |
| import org.apache.oozie.action.ActionExecutorException; |
| import org.apache.oozie.client.OozieClient; |
| import org.apache.oozie.client.WorkflowAction; |
| import org.apache.oozie.client.WorkflowAction.Status; |
| import org.apache.oozie.command.wf.StartXCommand; |
| import org.apache.oozie.command.wf.SubmitXCommand; |
| import org.apache.oozie.executor.jpa.WorkflowActionQueryExecutor; |
| import org.apache.oozie.executor.jpa.WorkflowActionQueryExecutor.WorkflowActionQuery; |
| import org.apache.oozie.local.LocalOozie; |
| import org.apache.oozie.service.HadoopAccessorException; |
| import org.apache.oozie.service.HadoopAccessorService; |
| import org.apache.oozie.service.Services; |
| import org.apache.oozie.service.WorkflowAppService; |
| import org.apache.oozie.util.ClassUtils; |
| import org.apache.oozie.util.IOUtils; |
| import org.apache.oozie.util.PropertiesUtils; |
| import org.apache.oozie.util.XConfiguration; |
| import org.apache.oozie.util.XLog; |
| import org.apache.oozie.util.XmlUtils; |
| import org.jdom.Element; |
| |
| public class TestMapReduceActionExecutor extends ActionExecutorTestCase { |
| private static final XLog LOG = XLog.getLog(TestMapReduceActionExecutor.class); |
| |
| private static final String PIPES = "pipes"; |
| private static final String MAP_REDUCE = "map-reduce"; |
| |
| @Override |
| protected void setSystemProps() throws Exception { |
| super.setSystemProps(); |
| setSystemProperty("oozie.service.ActionService.executor.classes", |
| String.join(",", MapReduceActionExecutor.class.getName(), JavaActionExecutor.class.getName())); |
| setSystemProperty("oozie.credentials.credentialclasses", "cred=org.apache.oozie.action.hadoop.CredentialForTest"); |
| } |
| |
| public Element createUberJarActionXML(String uberJarPath, String additional) throws Exception{ |
| return XmlUtils.parseXml("<map-reduce>" + "<job-tracker>" + getJobTrackerUri() + "</job-tracker>" |
| + "<name-node>" + getNameNodeUri() + "</name-node>" + additional + "<configuration>" |
| + "<property><name>oozie.mapreduce.uber.jar</name><value>" + uberJarPath + "</value></property>" |
| + "</configuration>" + "</map-reduce>"); |
| } |
| |
| public void testConfigDefaultPropsToAction() throws Exception { |
| String actionXml = "<map-reduce>" |
| + " <prepare>" |
| + " <delete path=\"${nameNode}/user/${wf:user()}/mr/${outputDir}\"/>" |
| + " </prepare>" |
| + " <configuration>" |
| + " <property><name>bb</name><value>BB</value></property>" |
| + " <property><name>cc</name><value>from_action</value></property>" |
| + " </configuration>" |
| + " </map-reduce>"; |
| String wfXml = "<workflow-app xmlns=\"uri:oozie:workflow:0.5\" name=\"map-reduce-wf\">" |
| + "<global>" |
| + "<job-tracker>${jobTracker}</job-tracker>" |
| + "<name-node>${nameNode}</name-node>" |
| + "<configuration><property><name>aa</name><value>AA</value></property></configuration>" |
| + "</global>" |
| + " <start to=\"mr-node\"/>" |
| + " <action name=\"mr-node\">" |
| + actionXml |
| + " <ok to=\"end\"/>" |
| + " <error to=\"fail\"/>" |
| + "</action>" |
| + "<kill name=\"fail\">" |
| + " <message>Map/Reduce failed, error message[${wf:errorMessage(wf:lastErrorNode())}]</message>" |
| + "</kill>" |
| + "<end name=\"end\"/>" |
| + "</workflow-app>"; |
| |
| Writer writer = new OutputStreamWriter(new FileOutputStream(getTestCaseDir() + "/workflow.xml"), |
| StandardCharsets.UTF_8); |
| IOUtils.copyCharStream(new StringReader(wfXml), writer); |
| |
| Configuration conf = new XConfiguration(); |
| conf.set("nameNode", getNameNodeUri()); |
| conf.set("jobTracker", getJobTrackerUri()); |
| conf.set(OozieClient.USER_NAME, getTestUser()); |
| conf.set(OozieClient.APP_PATH, new File(getTestCaseDir(), "workflow.xml").toURI().toString()); |
| conf.set(OozieClient.LOG_TOKEN, "t"); |
| |
| OutputStream os = new FileOutputStream(getTestCaseDir() + "/config-default.xml"); |
| XConfiguration defaultConf = new XConfiguration(); |
| defaultConf.set("outputDir", "default-output-dir"); |
| defaultConf.set("mapred.mapper.class", "MM"); |
| defaultConf.set("mapred.reducer.class", "RR"); |
| defaultConf.set("cc", "from_default"); |
| defaultConf.writeXml(os); |
| os.close(); |
| |
| String wfId = new SubmitXCommand(conf).call(); |
| new StartXCommand(wfId).call(); |
| waitForWorkflowAction(wfId + "@mr-node"); |
| |
| WorkflowActionBean mrAction = WorkflowActionQueryExecutor.getInstance().get(WorkflowActionQuery.GET_ACTION, |
| wfId + "@mr-node"); |
| // check NN and JT settings |
| Element eAction = XmlUtils.parseXml(mrAction.getConf()); |
| Element eConf = eAction.getChild("name-node", eAction.getNamespace()); |
| assertEquals(getNameNodeUri(), eConf.getText()); |
| eConf = eAction.getChild("job-tracker", eAction.getNamespace()); |
| assertEquals(getJobTrackerUri(), eConf.getText()); |
| |
| // check other m-r settings |
| eConf = eAction.getChild("configuration", eAction.getNamespace()); |
| Configuration actionConf = new XConfiguration(new StringReader(XmlUtils.prettyPrint(eConf).toString())); |
| assertEquals("default-output-dir", actionConf.get("outputDir")); |
| assertEquals("MM", actionConf.get("mapred.mapper.class")); |
| assertEquals("RR", actionConf.get("mapred.reducer.class")); |
| // check that default did not overwrite same property explicit in action conf |
| assertEquals("from_action", actionConf.get("cc")); |
| // check that original conf and from global was not deleted |
| assertEquals("AA", actionConf.get("aa")); |
| assertEquals("BB", actionConf.get("bb")); |
| |
| //test no infinite recursion by param referring to itself e.g. path = ${path}/sub-path |
| actionXml = "<map-reduce>" |
| + " <prepare>" |
| + " <delete path=\"${nameNode}/user/${wf:user()}/mr/${outputDir}\"/>" |
| + " </prepare>" |
| + " <configuration>" |
| + " <property><name>cc</name><value>${cc}/action_cc</value></property>" |
| + " </configuration>" |
| + " </map-reduce>"; |
| |
| wfXml = "<workflow-app xmlns=\"uri:oozie:workflow:0.5\" name=\"map-reduce-wf\">" |
| + "<global>" |
| + "<job-tracker>${jobTracker}</job-tracker>" |
| + "<name-node>${nameNode}</name-node>" |
| + "<configuration><property><name>outputDir</name><value>global-output-dir</value></property></configuration>" |
| + "</global>" |
| + " <start to=\"mr-node\"/>" |
| + " <action name=\"mr-node\">" |
| + actionXml |
| + " <ok to=\"end\"/>" |
| + " <error to=\"fail\"/>" |
| + "</action>" |
| + "<kill name=\"fail\">" |
| + " <message>Map/Reduce failed, error message[${wf:errorMessage(wf:lastErrorNode())}]</message>" |
| + "</kill>" |
| + "<end name=\"end\"/>" |
| + "</workflow-app>"; |
| |
| writer = new OutputStreamWriter(new FileOutputStream(getTestCaseDir() + "/workflow.xml"), |
| StandardCharsets.UTF_8); |
| IOUtils.copyCharStream(new StringReader(wfXml), writer); |
| |
| wfId = new SubmitXCommand(conf).call(); |
| new StartXCommand(wfId).call(); |
| waitForWorkflowAction(wfId + "@mr-node"); |
| |
| mrAction = WorkflowActionQueryExecutor.getInstance().get(WorkflowActionQuery.GET_ACTION, |
| wfId + "@mr-node"); |
| |
| // check param |
| eAction = XmlUtils.parseXml(mrAction.getConf()); |
| eConf = eAction.getChild("configuration", eAction.getNamespace()); |
| actionConf = new XConfiguration(new StringReader(XmlUtils.prettyPrint(eConf).toString())); |
| // action param referring to same param name given in defaults cc = ${cc}/action_cc |
| assertEquals("from_default/action_cc", actionConf.get("cc")); |
| // check global is retained and has precedence over config-default |
| eConf = eAction.getChild("name-node", eAction.getNamespace()); |
| assertEquals(getNameNodeUri(), eConf.getText()); |
| assertEquals("global-output-dir", actionConf.get("outputDir")); |
| } |
| |
| public void testGlobalOverrideJobXml() throws Exception { |
| FileSystem fs = getFileSystem(); |
| |
| String jobXml1 = createJobXml("aa", "from_jobXml1", "bb", "from_jobXml1", "jobXml1", fs); |
| String jobXml2 = createJobXml("bb", "from_jobXml2", "cc", "from_jobXml2", "jobXml2", fs); |
| String jobXml3 = createJobXml("cc", "from_jobXml3", "dd", "from_jobXml3", "jobXml3", fs); |
| String jobXml4 = createJobXml("dd", "from_jobXml4", "ee", "from_jobXml4", "jobXml4", fs); |
| |
| String actionXml = "<map-reduce>" |
| + " <prepare>" |
| + " <delete path=\"${nameNode}/user/${wf:user()}/mr/${outputDir}\"/>" |
| + " </prepare>" |
| + " <job-xml>" + jobXml3 + "</job-xml>" |
| + " <job-xml>" + jobXml4 + "</job-xml>" |
| + " <configuration>" |
| + " <property><name>ee</name><value>from_action_config</value></property>" |
| + " </configuration>" |
| + " </map-reduce>"; |
| String wfXml = "<workflow-app xmlns=\"uri:oozie:workflow:0.5\" name=\"map-reduce-wf\">" |
| + "<global>" |
| + "<job-tracker>${jobTracker}</job-tracker>" |
| + "<name-node>${nameNode}</name-node>" |
| + "<job-xml>" + jobXml1 + "</job-xml>" |
| + "<job-xml>" + jobXml2 + "</job-xml>" |
| + "<configuration>" |
| + "<property><name>aa</name><value>from_global_config</value></property>" |
| + "<property><name>ee</name><value>from_global_config</value></property>" |
| + "</configuration>" |
| + "</global>" |
| + " <start to=\"mr-node\"/>" |
| + " <action name=\"mr-node\">" |
| + actionXml |
| + " <ok to=\"end\"/>" |
| + " <error to=\"fail\"/>" |
| + "</action>" |
| + "<kill name=\"fail\">" |
| + " <message>Map/Reduce failed, error message[${wf:errorMessage(wf:lastErrorNode())}]</message>" |
| + "</kill>" |
| + "<end name=\"end\"/>" |
| + "</workflow-app>"; |
| |
| Writer writer = new FileWriter(getTestCaseDir() + "/workflow.xml"); |
| IOUtils.copyCharStream(new StringReader(wfXml), writer); |
| |
| Configuration conf = new XConfiguration(); |
| conf.set("nameNode", getNameNodeUri()); |
| conf.set("jobTracker", getJobTrackerUri()); |
| conf.set(OozieClient.USER_NAME, getTestUser()); |
| conf.set(OozieClient.APP_PATH, "file://" + getTestCaseDir() + File.separator + "workflow.xml"); |
| |
| OutputStream os = new FileOutputStream(getTestCaseDir() + "/config-default.xml"); |
| XConfiguration defaultConf = new XConfiguration(); |
| defaultConf.set("aa", "from_config_default"); |
| defaultConf.set("ff", "from_config_default"); |
| defaultConf.writeXml(os); |
| os.close(); |
| defaultConf.set(WorkflowAppService.HADOOP_USER, getTestUser()); |
| |
| String wfId = new SubmitXCommand(conf).call(); |
| new StartXCommand(wfId).call(); |
| waitForWorkflowAction(wfId + "@mr-node"); |
| WorkflowActionBean mrAction = WorkflowActionQueryExecutor.getInstance().get(WorkflowActionQuery.GET_ACTION, |
| wfId + "@mr-node"); |
| |
| JavaActionExecutor ae = new JavaActionExecutor(); |
| WorkflowJobBean wf = createBaseWorkflow(defaultConf, "mr-action", wfXml); |
| Context context = new Context(wf, mrAction); |
| Element mrActionXml = XmlUtils.parseXml(mrAction.getConf()); |
| Configuration actionConfig = ae.setupActionConf(conf, context, mrActionXml, getAppPath()); |
| |
| // Check attribute values |
| assertEquals("from_global_config", actionConfig.get("aa")); |
| assertEquals("from_jobXml2", actionConfig.get("bb")); |
| assertEquals("from_jobXml3", actionConfig.get("cc")); |
| assertEquals("from_jobXml4", actionConfig.get("dd")); |
| assertEquals("from_action_config", actionConfig.get("ee")); |
| assertEquals("from_config_default", actionConfig.get("ff")); |
| } |
| |
| protected String createJobXml(String key1, String value1, String key2, String value2, String filename, FileSystem fs) |
| throws Exception { |
| String content = "<configuration>" |
| + "<property><name>" + key1 + "</name><value>" + value1 + "</value></property>" |
| + "<property><name>" + key2 + "</name><value>" + value2 + "</value></property>" |
| + "</configuration>"; |
| |
| Path path = new Path(getAppPath(), filename); |
| Writer writer = new OutputStreamWriter(fs.create(path, true)); |
| writer.write(content); |
| writer.close(); |
| |
| return path.toString(); |
| } |
| |
| public void testSetupMethods() throws Exception { |
| MapReduceActionExecutor ae = new MapReduceActionExecutor(); |
| List<Class<?>> classes = Arrays.<Class<?>>asList(StreamingMain.class); |
| assertEquals(classes, ae.getLauncherClasses()); |
| |
| Element actionXml = XmlUtils.parseXml("<map-reduce>" + "<job-tracker>" + getJobTrackerUri() + "</job-tracker>" |
| + "<name-node>" + getNameNodeUri() + "</name-node>" + "<configuration>" |
| + "<property><name>mapred.input.dir</name><value>IN</value></property>" |
| + "<property><name>mapred.output.dir</name><value>OUT</value></property>" + "</configuration>" |
| + "</map-reduce>"); |
| |
| XConfiguration protoConf = new XConfiguration(); |
| protoConf.set(WorkflowAppService.HADOOP_USER, getTestUser()); |
| |
| WorkflowJobBean wf = createBaseWorkflow(protoConf, "mr-action"); |
| WorkflowActionBean action = (WorkflowActionBean) wf.getActions().get(0); |
| action.setType(ae.getType()); |
| |
| Context context = new Context(wf, action); |
| |
| Configuration conf = ae.createBaseHadoopConf(context, actionXml); |
| ae.setupActionConf(conf, context, actionXml, getFsTestCaseDir()); |
| assertEquals("IN", conf.get("mapred.input.dir")); |
| Configuration launcherJobConf = ae.createLauncherConf(getFileSystem(), context, action, actionXml, conf); |
| assertEquals(false, launcherJobConf.getBoolean("mapreduce.job.complete.cancel.delegation.tokens", true)); |
| assertEquals(true, conf.getBoolean("mapreduce.job.complete.cancel.delegation.tokens", false)); |
| |
| // Enable uber jars to test that MapReduceActionExecutor picks up the oozie.mapreduce.uber.jar property correctly |
| Services serv = Services.get(); |
| boolean originalUberJarDisabled = serv.getConf().getBoolean("oozie.action.mapreduce.uber.jar.enable", false); |
| serv.getConf().setBoolean("oozie.action.mapreduce.uber.jar.enable", true); |
| |
| actionXml = createUberJarActionXML(getNameNodeUri() + "/app/job.jar", ""); |
| conf = ae.createBaseHadoopConf(context, actionXml); |
| ae.setupActionConf(conf, context, actionXml, getFsTestCaseDir()); |
| // absolute path with namenode |
| assertEquals(getNameNodeUri() + "/app/job.jar", conf.get(MapReduceMain.OOZIE_MAPREDUCE_UBER_JAR)); |
| |
| actionXml = createUberJarActionXML("/app/job.jar", ""); |
| conf = ae.createBaseHadoopConf(context, actionXml); |
| ae.setupActionConf(conf, context, actionXml, getFsTestCaseDir()); |
| // absolute path without namenode |
| assertEquals(getNameNodeUri() + "/app/job.jar", conf.get(MapReduceMain.OOZIE_MAPREDUCE_UBER_JAR)); |
| |
| actionXml = createUberJarActionXML("job.jar", ""); |
| conf = ae.createBaseHadoopConf(context, actionXml); |
| ae.setupActionConf(conf, context, actionXml, getFsTestCaseDir()); |
| assertEquals(getFsTestCaseDir() + "/job.jar", conf.get(MapReduceMain.OOZIE_MAPREDUCE_UBER_JAR)); // relative path |
| |
| actionXml = createUberJarActionXML("job.jar", "<streaming></streaming>"); |
| conf = ae.createBaseHadoopConf(context, actionXml); |
| ae.setupActionConf(conf, context, actionXml, getFsTestCaseDir()); |
| // ignored for streaming |
| assertEquals("", conf.get(MapReduceMain.OOZIE_MAPREDUCE_UBER_JAR)); |
| |
| actionXml = createUberJarActionXML("job.jar", "<pipes></pipes>"); |
| conf = ae.createBaseHadoopConf(context, actionXml); |
| ae.setupActionConf(conf, context, actionXml, getFsTestCaseDir()); |
| assertEquals("", conf.get("oozie.mapreduce.uber.jar")); // ignored for pipes |
| |
| actionXml = XmlUtils.parseXml("<map-reduce>" + "<job-tracker>" + getJobTrackerUri() + "</job-tracker>" |
| + "<name-node>" + getNameNodeUri() + "</name-node>" + "</map-reduce>"); |
| conf = ae.createBaseHadoopConf(context, actionXml); |
| ae.setupActionConf(conf, context, actionXml, getFsTestCaseDir()); |
| assertNull(conf.get("oozie.mapreduce.uber.jar")); // doesn't resolve if not set |
| |
| // Disable uber jars to test that MapReduceActionExecutor won't allow the oozie.mapreduce.uber.jar property |
| serv.getConf().setBoolean("oozie.action.mapreduce.uber.jar.enable", false); |
| try { |
| actionXml = createUberJarActionXML(getNameNodeUri() + "/app/job.jar", ""); |
| conf = ae.createBaseHadoopConf(context, actionXml); |
| ae.setupActionConf(conf, context, actionXml, getFsTestCaseDir()); |
| fail("ActionExecutorException expected because uber jars are disabled"); |
| } catch (ActionExecutorException aee) { |
| assertEquals("MR003", aee.getErrorCode()); |
| assertEquals(ActionExecutorException.ErrorType.ERROR, aee.getErrorType()); |
| assertTrue(aee.getMessage().contains("oozie.action.mapreduce.uber.jar.enable")); |
| assertTrue(aee.getMessage().contains("oozie.mapreduce.uber.jar")); |
| } |
| serv.getConf().setBoolean("oozie.action.mapreduce.uber.jar.enable", originalUberJarDisabled); |
| |
| actionXml = XmlUtils.parseXml("<map-reduce>" + "<job-tracker>" + getJobTrackerUri() + "</job-tracker>" |
| + "<name-node>" + getNameNodeUri() + "</name-node>" + "<streaming>" + "<mapper>M</mapper>" |
| + "<reducer>R</reducer>" + "<record-reader>RR</record-reader>" |
| + "<record-reader-mapping>RRM1=1</record-reader-mapping>" |
| + "<record-reader-mapping>RRM2=2</record-reader-mapping>" + "<env>e=E</env>" + "<env>ee=EE</env>" |
| + "</streaming>" + "<configuration>" |
| + "<property><name>mapred.input.dir</name><value>IN</value></property>" |
| + "<property><name>mapred.output.dir</name><value>OUT</value></property>" + "</configuration>" |
| + "</map-reduce>"); |
| |
| conf = ae.createBaseHadoopConf(context, actionXml); |
| ae.setupActionConf(conf, context, actionXml, getFsTestCaseDir()); |
| assertEquals("M", conf.get("oozie.streaming.mapper")); |
| assertEquals("R", conf.get("oozie.streaming.reducer")); |
| assertEquals("RR", conf.get("oozie.streaming.record-reader")); |
| assertEquals("2", conf.get("oozie.streaming.record-reader-mapping.size")); |
| assertEquals("2", conf.get("oozie.streaming.env.size")); |
| |
| actionXml = XmlUtils.parseXml("<map-reduce>" + "<job-tracker>" + getJobTrackerUri() + "</job-tracker>" |
| + "<name-node>" + getNameNodeUri() + "</name-node>" + "<pipes>" + "<map>M</map>" + "<reduce>R</reduce>" |
| + "<inputformat>IF</inputformat>" + "<partitioner>P</partitioner>" + "<writer>W</writer>" |
| + "<program>PP</program>" + "</pipes>" + "<configuration>" |
| + "<property><name>mapred.input.dir</name><value>IN</value></property>" |
| + "<property><name>mapred.output.dir</name><value>OUT</value></property>" + "</configuration>" |
| + "</map-reduce>"); |
| |
| conf = ae.createBaseHadoopConf(context, actionXml); |
| ae.setupActionConf(conf, context, actionXml, getFsTestCaseDir()); |
| assertEquals("M", conf.get("oozie.pipes.map")); |
| assertEquals("R", conf.get("oozie.pipes.reduce")); |
| assertEquals("IF", conf.get("oozie.pipes.inputformat")); |
| assertEquals("P", conf.get("oozie.pipes.partitioner")); |
| assertEquals("W", conf.get("oozie.pipes.writer")); |
| assertEquals(getFsTestCaseDir()+"/PP", conf.get("oozie.pipes.program")); |
| } |
| |
| protected Context createContext(String name, String actionXml) throws Exception { |
| JavaActionExecutor ae = new JavaActionExecutor(); |
| |
| Path appJarPath = new Path("lib/test.jar"); |
| File jarFile = IOUtils.createJar(new File(getTestCaseDir()), "test.jar", MapperReducerForTest.class); |
| InputStream is = new FileInputStream(jarFile); |
| OutputStream os = getFileSystem().create(new Path(getAppPath(), "lib/test.jar")); |
| IOUtils.copyStream(is, os); |
| |
| XConfiguration protoConf = new XConfiguration(); |
| protoConf.set(WorkflowAppService.HADOOP_USER, getTestUser()); |
| |
| protoConf.setStrings(WorkflowAppService.APP_LIB_PATH_LIST, appJarPath.toString()); |
| |
| WorkflowJobBean wf = createBaseWorkflow(protoConf, "mr-action"); |
| WorkflowActionBean action = (WorkflowActionBean) wf.getActions().get(0); |
| action.setName(name); |
| action.setType(ae.getType()); |
| action.setConf(actionXml); |
| |
| return new Context(wf, action); |
| } |
| |
| protected Context createContextWithCredentials(String name, String actionXml) throws Exception { |
| JavaActionExecutor ae = new JavaActionExecutor(); |
| |
| Path appJarPath = new Path("lib/test.jar"); |
| File jarFile = IOUtils.createJar(new File(getTestCaseDir()), "test.jar", MapperReducerForTest.class); |
| InputStream is = new FileInputStream(jarFile); |
| OutputStream os = getFileSystem().create(new Path(getAppPath(), "lib/test.jar")); |
| IOUtils.copyStream(is, os); |
| |
| XConfiguration protoConf = new XConfiguration(); |
| protoConf.set(WorkflowAppService.HADOOP_USER, getTestUser()); |
| protoConf.setStrings(WorkflowAppService.APP_LIB_PATH_LIST, appJarPath.toString()); |
| |
| |
| WorkflowJobBean wf = createBaseWorkflowWithCredentials(protoConf, "mr-action"); |
| WorkflowActionBean action = (WorkflowActionBean) wf.getActions().get(0); |
| action.setName(name); |
| action.setType(ae.getType()); |
| action.setConf(actionXml); |
| action.setCred("testcred"); |
| |
| return new Context(wf, action); |
| } |
| |
| protected String submitAction(Context context) throws Exception { |
| MapReduceActionExecutor ae = new MapReduceActionExecutor(); |
| |
| WorkflowAction action = context.getAction(); |
| |
| ae.prepareActionDir(getFileSystem(), context); |
| ae.submitLauncher(getFileSystem(), context, action); |
| |
| return context.getAction().getExternalId(); |
| } |
| |
| private String _testSubmit(String name, String actionXml) throws Exception { |
| |
| Context context = createContext(name, actionXml); |
| final String launcherId = submitAction(context); |
| waitUntilYarnAppDoneAndAssertSuccess(launcherId); |
| |
| Map<String, String> actionData = LauncherHelper.getActionData(getFileSystem(), context.getActionDir(), |
| context.getProtoActionConf()); |
| assertTrue(LauncherHelper.hasIdSwap(actionData)); |
| |
| MapReduceActionExecutor ae = new MapReduceActionExecutor(); |
| ae.check(context, context.getAction()); |
| assertTrue(launcherId.equals(context.getAction().getExternalId())); |
| |
| String externalChildIDs = context.getAction().getExternalChildIDs(); |
| waitUntilYarnAppDoneAndAssertSuccess(externalChildIDs); |
| ae.check(context, context.getAction()); |
| |
| assertEquals(JavaActionExecutor.SUCCEEDED, context.getAction().getExternalStatus()); |
| assertNull(context.getAction().getData()); |
| |
| ae.end(context, context.getAction()); |
| assertEquals(WorkflowAction.Status.OK, context.getAction().getStatus()); |
| |
| //hadoop.counters will always be set in case of MR action. |
| assertNotNull(context.getVar("hadoop.counters")); |
| String counters = context.getVar("hadoop.counters"); |
| assertTrue(counters.contains("Counter")); |
| assertTrue(counters.contains("\"MAP_OUTPUT_RECORDS\":2")); |
| |
| //External Child IDs used to be null, but after 4.0, become Non-Null in case of MR action. |
| assertNotNull(context.getExternalChildIDs()); |
| |
| return externalChildIDs; |
| } |
| |
| private void _testSubmitError(String actionXml, String errorMessage) throws Exception { |
| Context context = createContext(MAP_REDUCE, actionXml); |
| final String launcherId = submitAction(context); |
| waitUntilYarnAppDoneAndAssertSuccess(launcherId); |
| |
| MapReduceActionExecutor ae = new MapReduceActionExecutor(); |
| ae.check(context, context.getAction()); |
| |
| assertEquals(JavaActionExecutor.FAILED_KILLED, context.getAction().getExternalStatus()); |
| |
| ae.end(context, context.getAction()); |
| assertEquals(WorkflowAction.Status.ERROR, context.getAction().getStatus()); |
| assertTrue(context.getAction().getErrorMessage().contains("already exists")); |
| } |
| |
| private void _testSubmitWithCredentials(String name, String actionXml) throws Exception { |
| |
| Context context = createContextWithCredentials(MAP_REDUCE, actionXml); |
| final String launcherId = submitAction(context); |
| waitUntilYarnAppDoneAndAssertSuccess(launcherId); |
| |
| Map<String, String> actionData = LauncherHelper.getActionData(getFileSystem(), context.getActionDir(), |
| context.getProtoActionConf()); |
| assertTrue(LauncherHelper.hasIdSwap(actionData)); |
| |
| MapReduceActionExecutor ae = new MapReduceActionExecutor(); |
| ae.check(context, context.getAction()); |
| assertTrue(launcherId.equals(context.getAction().getExternalId())); |
| |
| String externalChildIDs = context.getAction().getExternalChildIDs(); |
| waitUntilYarnAppDoneAndAssertSuccess(externalChildIDs); |
| ae.check(context, context.getAction()); |
| |
| assertEquals(JavaActionExecutor.SUCCEEDED, context.getAction().getExternalStatus()); |
| assertNull(context.getAction().getData()); |
| |
| ae.end(context, context.getAction()); |
| assertEquals(WorkflowAction.Status.OK, context.getAction().getStatus()); |
| Configuration conf = ae.createBaseHadoopConf(context, XmlUtils.parseXml(actionXml)); |
| String user = conf.get("user.name"); |
| JobClient jobClient = getHadoopAccessorService().createJobClient(user, conf); |
| org.apache.hadoop.mapreduce.JobID jobID = TypeConverter.fromYarn( |
| ConverterUtils.toApplicationId(externalChildIDs)); |
| final RunningJob mrJob = jobClient.getJob(JobID.downgrade(jobID)); |
| assertTrue(MapperReducerCredentialsForTest.hasCredentials(mrJob)); |
| } |
| |
| protected XConfiguration getSleepMapReduceConfig(String inputDir, String outputDir) { |
| XConfiguration conf = getMapReduceConfig(inputDir, outputDir); |
| conf.set("mapred.mapper.class", BlockingMapper.class.getName()); |
| return conf; |
| } |
| |
| protected XConfiguration getMapReduceConfig(String inputDir, String outputDir) { |
| XConfiguration conf = new XConfiguration(); |
| conf.set("mapred.mapper.class", MapperReducerForTest.class.getName()); |
| conf.set("mapred.reducer.class", MapperReducerForTest.class.getName()); |
| conf.set("mapred.input.dir", inputDir); |
| conf.set("mapred.output.dir", outputDir); |
| return conf; |
| } |
| |
| protected XConfiguration getMapReduceCredentialsConfig(String inputDir, String outputDir) { |
| XConfiguration conf = new XConfiguration(); |
| conf.set("mapred.mapper.class", MapperReducerCredentialsForTest.class.getName()); |
| conf.set("mapred.reducer.class", MapperReducerForTest.class.getName()); |
| conf.set("mapred.input.dir", inputDir); |
| conf.set("mapred.output.dir", outputDir); |
| return conf; |
| } |
| |
| protected XConfiguration getMapReduceUberJarConfig(String inputDir, String outputDir) throws Exception{ |
| XConfiguration conf = new XConfiguration(); |
| conf.set("mapred.mapper.class", MapperReducerUberJarForTest.class.getName()); |
| conf.set("mapred.reducer.class", MapperReducerUberJarForTest.class.getName()); |
| conf.set("mapred.input.dir", inputDir); |
| conf.set("mapred.output.dir", outputDir); |
| conf.set("oozie.mapreduce.uber.jar", createAndUploadUberJar().toUri().toString()); |
| return conf; |
| } |
| |
| public void testMapReduce() throws Exception { |
| FileSystem fs = getFileSystem(); |
| |
| Path inputDir = new Path(getFsTestCaseDir(), "input"); |
| Path outputDir = new Path(getFsTestCaseDir(), "output"); |
| |
| writeDummyInput(fs, inputDir); |
| |
| String actionXml = "<map-reduce>" + "<job-tracker>" + getJobTrackerUri() + "</job-tracker>" + "<name-node>" |
| + getNameNodeUri() + "</name-node>" |
| + getMapReduceConfig(inputDir.toString(), outputDir.toString()).toXmlString(false) + "</map-reduce>"; |
| _testSubmit(MAP_REDUCE, actionXml); |
| } |
| |
| public void testMapReduceActionError() throws Exception { |
| FileSystem fs = getFileSystem(); |
| |
| Path inputDir = new Path(getFsTestCaseDir(), "input"); |
| Path outputDir = new Path(getFsTestCaseDir(), "output1"); |
| |
| Writer w = new OutputStreamWriter(fs.create(new Path(inputDir, "data.txt")), StandardCharsets.UTF_8); |
| w.write("dummy\n"); |
| w.write("dummy\n"); |
| writeDummyInput(fs, outputDir); |
| |
| String actionXml = "<map-reduce>" + |
| "<job-tracker>" + getJobTrackerUri() + "</job-tracker>" + |
| "<name-node>" + getNameNodeUri() + "</name-node>" + |
| "<configuration>" + |
| "<property><name>mapred.mapper.class</name><value>" + MapperReducerForTest.class.getName() + |
| "</value></property>" + |
| "<property><name>mapred.reducer.class</name><value>" + MapperReducerForTest.class.getName() + |
| "</value></property>" + |
| "<property><name>mapred.input.dir</name><value>" + inputDir + "</value></property>" + |
| "<property><name>mapred.output.dir</name><value>" + outputDir + "</value></property>" + |
| "</configuration>" + |
| "</map-reduce>"; |
| |
| _testSubmitError(actionXml, "already exists"); |
| } |
| |
| public void testMapReduceWithConfigClass() throws Exception { |
| FileSystem fs = getFileSystem(); |
| |
| Path inputDir = new Path(getFsTestCaseDir(), "input"); |
| Path outputDir = new Path(getFsTestCaseDir(), "output"); |
| |
| writeDummyInput(fs, inputDir); |
| |
| Path jobXml = new Path(getFsTestCaseDir(), "action.xml"); |
| XConfiguration conf = getMapReduceConfig(inputDir.toString(), outputDir.toString()); |
| conf.set(MapperReducerForTest.JOB_XML_OUTPUT_LOCATION, jobXml.toUri().toString()); |
| conf.set("B", "b"); |
| String actionXml = "<map-reduce>" + "<job-tracker>" + getJobTrackerUri() + "</job-tracker>" + "<name-node>" |
| + getNameNodeUri() + "</name-node>" |
| + conf.toXmlString(false) |
| + "<config-class>" + OozieActionConfiguratorForTest.class.getName() + "</config-class>" + "</map-reduce>"; |
| |
| _testSubmit(MAP_REDUCE, actionXml); |
| Configuration conf2 = new Configuration(false); |
| conf2.addResource(fs.open(jobXml)); |
| |
| assertEquals("a", conf2.get("A")); |
| assertEquals("c", conf2.get("B")); |
| } |
| |
| public void testMapReduceWithConfigClassNotFound() throws Exception { |
| FileSystem fs = getFileSystem(); |
| |
| Path inputDir = new Path(getFsTestCaseDir(), "input"); |
| Path outputDir = new Path(getFsTestCaseDir(), "output"); |
| |
| writeDummyInput(fs, inputDir); |
| |
| String actionXml = "<map-reduce>" + "<job-tracker>" + getJobTrackerUri() + "</job-tracker>" + "<name-node>" |
| + getNameNodeUri() + "</name-node>" |
| + getMapReduceConfig(inputDir.toString(), outputDir.toString()).toXmlString(false) |
| + "<config-class>org.apache.oozie.does.not.exist</config-class>" + "</map-reduce>"; |
| |
| Context context = createContext(MAP_REDUCE, actionXml); |
| final String launcherId = submitAction(context); |
| waitUntilYarnAppDoneAndAssertSuccess(launcherId); |
| |
| final Map<String, String> actionData = LauncherHelper.getActionData(fs, context.getActionDir(), |
| context.getProtoActionConf()); |
| Properties errorProps = PropertiesUtils.stringToProperties(actionData.get(LauncherAMUtils.ACTION_DATA_ERROR_PROPS)); |
| assertEquals("An Exception occurred while instantiating the action config class", |
| errorProps.getProperty("exception.message")); |
| assertTrue(errorProps.getProperty("exception.stacktrace").startsWith(OozieActionConfiguratorException.class.getName())); |
| } |
| |
| public void testMapReduceWithConfigClassThrowException() throws Exception { |
| FileSystem fs = getFileSystem(); |
| |
| Path inputDir = new Path(getFsTestCaseDir(), "input"); |
| Path outputDir = new Path(getFsTestCaseDir(), "output"); |
| |
| writeDummyInput(fs, inputDir); |
| |
| XConfiguration conf = getMapReduceConfig(inputDir.toString(), outputDir.toString()); |
| conf.setBoolean("oozie.test.throw.exception", true); // causes OozieActionConfiguratorForTest to throw an exception |
| String actionXml = "<map-reduce>" + "<job-tracker>" + getJobTrackerUri() + "</job-tracker>" + "<name-node>" |
| + getNameNodeUri() + "</name-node>" |
| + conf.toXmlString(false) |
| + "<config-class>" + OozieActionConfiguratorForTest.class.getName() + "</config-class>" + "</map-reduce>"; |
| |
| Context context = createContext(MAP_REDUCE, actionXml); |
| final String launcherId = submitAction(context); |
| waitUntilYarnAppDoneAndAssertSuccess(launcherId); |
| |
| final Map<String, String> actionData = LauncherHelper.getActionData(fs, context.getActionDir(), |
| context.getProtoActionConf()); |
| Properties errorProps = PropertiesUtils.stringToProperties(actionData.get(LauncherAMUtils.ACTION_DATA_ERROR_PROPS)); |
| assertEquals("doh", errorProps.getProperty("exception.message")); |
| assertTrue(errorProps.getProperty("exception.stacktrace").startsWith(OozieActionConfiguratorException.class.getName())); |
| } |
| |
| public void testMapReduceActionKill() throws Exception { |
| FileSystem fs = getFileSystem(); |
| |
| Path inputDir = new Path(getFsTestCaseDir(), "input"); |
| Path outputDir = new Path(getFsTestCaseDir(), "output"); |
| |
| writeDummyInput(fs, inputDir); |
| |
| String actionXml = "<map-reduce>" + "<job-tracker>" + getJobTrackerUri() + "</job-tracker>" + "<name-node>" |
| + getNameNodeUri() + "</name-node>" |
| + getSleepMapReduceConfig(inputDir.toString(), outputDir.toString()).toXmlString(false) + "</map-reduce>"; |
| |
| Context context = createContext(MAP_REDUCE, actionXml); |
| final String launcherId = submitAction(context); |
| // wait until LauncherAM terminates - the MR job keeps running the background |
| waitUntilYarnAppDoneAndAssertSuccess(launcherId); |
| |
| MapReduceActionExecutor mae = new MapReduceActionExecutor(); |
| mae.check(context, context.getAction()); // must be called so that externalChildIDs are read from HDFS |
| |
| mae.kill(context, context.getAction()); |
| |
| waitUntilYarnAppKilledAndAssertSuccess(context.getAction().getExternalChildIDs()); |
| } |
| |
| public void testMapReduceWithCredentials() throws Exception { |
| FileSystem fs = getFileSystem(); |
| |
| Path inputDir = new Path(getFsTestCaseDir(), "input"); |
| Path outputDir = new Path(getFsTestCaseDir(), "output"); |
| |
| writeDummyInput(fs, inputDir); |
| |
| String actionXml = "<map-reduce>" + "<job-tracker>" + getJobTrackerUri() + "</job-tracker>" + "<name-node>" |
| + getNameNodeUri() + "</name-node>" |
| + getMapReduceCredentialsConfig(inputDir.toString(), outputDir.toString()).toXmlString(false) |
| + "</map-reduce>"; |
| _testSubmitWithCredentials(MAP_REDUCE, actionXml); |
| } |
| |
| protected Path createAndUploadUberJar() throws Exception { |
| Path localJobJarPath = makeUberJarWithLib(getTestCaseDir()); |
| Path remoteJobJarPath = new Path(getAppPath(), localJobJarPath.getName()); |
| getFileSystem().moveFromLocalFile(localJobJarPath, remoteJobJarPath); |
| File localJobJarFile = new File(localJobJarPath.toUri().toString()); |
| if (localJobJarFile.exists()) { // just to make sure |
| localJobJarFile.delete(); |
| } |
| return remoteJobJarPath; |
| } |
| |
| private Path makeUberJarWithLib(String testDir) throws Exception { |
| Path jobJarPath = new Path(testDir, "uber.jar"); |
| FileOutputStream fos = new FileOutputStream(new File(jobJarPath.toUri().getPath())); |
| JarOutputStream jos = new JarOutputStream(fos); |
| // Have to put in real jar files or it will complain |
| createAndAddJarToJar(jos, new File(new Path(testDir, "lib1.jar").toUri().getPath())); |
| createAndAddJarToJar(jos, new File(new Path(testDir, "lib2.jar").toUri().getPath())); |
| jos.close(); |
| return jobJarPath; |
| } |
| |
| private void createAndAddJarToJar(JarOutputStream jos, File jarFile) throws Exception { |
| FileOutputStream fos2 = new FileOutputStream(jarFile); |
| JarOutputStream jos2 = new JarOutputStream(fos2); |
| // Have to have at least one entry or it will complain |
| ZipEntry ze = new ZipEntry(jarFile.getName() + ".inside"); |
| jos2.putNextEntry(ze); |
| jos2.closeEntry(); |
| jos2.close(); |
| ze = new ZipEntry("lib/" + jarFile.getName()); |
| jos.putNextEntry(ze); |
| FileInputStream in = new FileInputStream(jarFile); |
| byte buf[] = new byte[1024]; |
| int numRead; |
| do { |
| numRead = in.read(buf); |
| if (numRead >= 0) { |
| jos.write(buf, 0, numRead); |
| } |
| } while (numRead != -1); |
| in.close(); |
| jos.closeEntry(); |
| jarFile.delete(); |
| } |
| |
| public void _testMapReduceWithUberJar() throws Exception { |
| FileSystem fs = getFileSystem(); |
| |
| Path inputDir = new Path(getFsTestCaseDir(), "input"); |
| Path outputDir = new Path(getFsTestCaseDir(), "output"); |
| |
| writeDummyInput(fs, inputDir); |
| |
| String actionXml = "<map-reduce>" + "<job-tracker>" + getJobTrackerUri() + "</job-tracker>" + "<name-node>" |
| + getNameNodeUri() + "</name-node>" |
| + getMapReduceUberJarConfig(inputDir.toString(), outputDir.toString()).toXmlString(false) + "</map-reduce>"; |
| String appID = _testSubmit(MAP_REDUCE, actionXml); |
| |
| boolean containsLib1Jar = false; |
| String lib1JarStr = "jobcache/" + appID + "/jars/lib/lib1.jar"; |
| Pattern lib1JarPatYarn = Pattern.compile( |
| ".*appcache/" + appID + "/filecache/.*/uber.jar/lib/lib1.jar.*"); |
| boolean containsLib2Jar = false; |
| String lib2JarStr = "jobcache/" + appID + "/jars/lib/lib1.jar"; |
| Pattern lib2JarPatYarn = Pattern.compile( |
| ".*appcache/" + appID + "/filecache/.*/uber.jar/lib/lib2.jar.*"); |
| |
| FileStatus[] fstats = getFileSystem().listStatus(outputDir); |
| for (FileStatus fstat : fstats) { |
| Path p = fstat.getPath(); |
| if (getFileSystem().isFile(p) && p.getName().startsWith("part-")) { |
| InputStream is = getFileSystem().open(p); |
| Scanner sc = new Scanner(is,StandardCharsets.UTF_8.name()); |
| while (sc.hasNextLine()) { |
| String line = sc.nextLine(); |
| containsLib1Jar = (containsLib1Jar || line.contains(lib1JarStr) || lib1JarPatYarn.matcher(line).matches()); |
| containsLib2Jar = (containsLib2Jar || line.contains(lib2JarStr) || lib2JarPatYarn.matcher(line).matches()); |
| } |
| sc.close(); |
| is.close(); |
| } |
| } |
| |
| assertTrue("lib/lib1.jar should have been unzipped from the uber jar and added to the classpath but was not", |
| containsLib1Jar); |
| assertTrue("lib/lib2.jar should have been unzipped from the uber jar and added to the classpath but was not", |
| containsLib2Jar); |
| } |
| |
| // With the oozie.action.mapreduce.uber.jar.enable property set to false, a workflow with an uber jar should fail |
| public void testMapReduceWithUberJarDisabled() throws Exception { |
| Services serv = Services.get(); |
| boolean originalUberJarDisabled = serv.getConf().getBoolean("oozie.action.mapreduce.uber.jar.enable", false); |
| try { |
| serv.getConf().setBoolean("oozie.action.mapreduce.uber.jar.enable", false); |
| _testMapReduceWithUberJar(); |
| } catch (ActionExecutorException aee) { |
| assertEquals("MR003", aee.getErrorCode()); |
| assertEquals(ActionExecutorException.ErrorType.ERROR, aee.getErrorType()); |
| assertTrue(aee.getMessage().contains("oozie.action.mapreduce.uber.jar.enable")); |
| assertTrue(aee.getMessage().contains("oozie.mapreduce.uber.jar")); |
| } catch (Exception e) { |
| throw e; |
| } finally { |
| serv.getConf().setBoolean("oozie.action.mapreduce.uber.jar.enable", originalUberJarDisabled); |
| } |
| } |
| |
| public void testJobNameSetForMapReduceChild() throws Exception { |
| Services serv = Services.get(); |
| serv.getConf().setBoolean("oozie.action.mapreduce.uber.jar.enable", true); |
| |
| final FileSystem fs = getFileSystem(); |
| final Path inputDir = new Path(getFsTestCaseDir(), "input"); |
| final Path outputDir = new Path(getFsTestCaseDir(), "output"); |
| writeDummyInput(fs, inputDir); |
| |
| final String actionXml = "<map-reduce>" + "<job-tracker>" + getJobTrackerUri() + "</job-tracker>" + "<name-node>" |
| + getNameNodeUri() + "</name-node>" |
| + getMapReduceUberJarConfig(inputDir.toString(), outputDir.toString()).toXmlString(false) + "</map-reduce>"; |
| |
| final String extId = _testSubmit(MAP_REDUCE, actionXml); |
| final ApplicationId appId = ConverterUtils.toApplicationId(extId); |
| final Configuration conf = getHadoopAccessorService().createConfiguration(getJobTrackerUri()); |
| final String name = getHadoopAccessorService().createYarnClient(getTestUser(), conf).getApplicationReport(appId).getName(); |
| assertTrue(name.contains("oozie:action")); |
| } |
| |
| private void writeDummyInput(FileSystem fs, Path inputDir) throws IOException { |
| Writer w = new OutputStreamWriter(fs.create(new Path(inputDir, "data.txt")), StandardCharsets.UTF_8); |
| w.write("dummy\n"); |
| w.write("dummy\n"); |
| w.close(); |
| } |
| |
| public void testMapReduceWithUberJarEnabled() throws Exception { |
| Services serv = Services.get(); |
| boolean originalUberJarDisabled = serv.getConf().getBoolean("oozie.action.mapreduce.uber.jar.enable", false); |
| try { |
| serv.getConf().setBoolean("oozie.action.mapreduce.uber.jar.enable", true); |
| _testMapReduceWithUberJar(); |
| } catch (Exception e) { |
| throw e; |
| } finally { |
| serv.getConf().setBoolean("oozie.action.mapreduce.uber.jar.enable", originalUberJarDisabled); |
| } |
| } |
| |
| protected XConfiguration getStreamingConfig(String inputDir, String outputDir) { |
| XConfiguration conf = new XConfiguration(); |
| conf.set("mapred.input.dir", inputDir); |
| conf.set("mapred.output.dir", outputDir); |
| return conf; |
| } |
| |
| private void runStreamingWordCountJob(Path inputDir, Path outputDir, XConfiguration streamingConf) throws Exception { |
| FileSystem fs = getFileSystem(); |
| Path streamingJar = new Path(getFsTestCaseDir(), "jar/hadoop-streaming.jar"); |
| |
| InputStream is = new FileInputStream(ClassUtils.findContainingJar(StreamJob.class)); |
| OutputStream os = fs.create(new Path(getAppPath(), streamingJar)); |
| IOUtils.copyStream(is, os); |
| |
| writeDummyInput(fs, inputDir); |
| |
| String actionXml = "<map-reduce>" + "<job-tracker>" + getJobTrackerUri() + "</job-tracker>" + "<name-node>" |
| + getNameNodeUri() + "</name-node>" + " <streaming>" + " <mapper>cat</mapper>" |
| + " <reducer>wc</reducer>" + " </streaming>" |
| + streamingConf.toXmlString(false) + "<file>" |
| + streamingJar + "</file>" + "</map-reduce>"; |
| _testSubmit("streaming", actionXml); |
| } |
| |
| public void testStreaming() throws Exception { |
| FileSystem fs = getFileSystem(); |
| Path inputDir = new Path(getFsTestCaseDir(), "input"); |
| Path outputDir = new Path(getFsTestCaseDir(), "output"); |
| final XConfiguration streamingConf = getStreamingConfig(inputDir.toString(), outputDir.toString()); |
| |
| runStreamingWordCountJob(inputDir, outputDir, streamingConf); |
| |
| final FSDataInputStream dis = fs.open(getOutputFile(outputDir, fs)); |
| final List<String> lines = org.apache.commons.io.IOUtils.readLines(dis); |
| dis.close(); |
| assertEquals(1, lines.size()); |
| // Not sure why it is 14 instead of 12. \n twice ?? |
| assertEquals("2 2 14", lines.get(0).trim()); |
| } |
| |
| public void testStreamingConfOverride() throws Exception { |
| FileSystem fs = getFileSystem(); |
| Path inputDir = new Path(getFsTestCaseDir(), "input"); |
| Path outputDir = new Path(getFsTestCaseDir(), "output"); |
| final XConfiguration streamingConf = getStreamingConfig(inputDir.toString(), outputDir.toString()); |
| streamingConf.set("mapred.output.format.class", "org.apache.hadoop.mapred.SequenceFileOutputFormat"); |
| |
| runStreamingWordCountJob(inputDir, outputDir, streamingConf); |
| |
| SequenceFile.Reader seqFile = new SequenceFile.Reader(fs, getOutputFile(outputDir, fs), getFileSystem().getConf()); |
| Text key = new Text(), value = new Text(); |
| if (seqFile.next(key, value)) { |
| assertEquals("2 2 14", key.toString().trim()); |
| assertEquals("", value.toString()); |
| } |
| assertFalse(seqFile.next(key, value)); |
| seqFile.close(); |
| } |
| |
| private Path getOutputFile(Path outputDir, FileSystem fs) throws FileNotFoundException, IOException { |
| final FileStatus[] files = fs.listStatus(outputDir, new PathFilter() { |
| |
| @Override |
| public boolean accept(Path path) { |
| return path.getName().startsWith("part"); |
| } |
| }); |
| return files[0].getPath(); //part-[m/r]-00000 |
| } |
| |
| protected XConfiguration getPipesConfig(String inputDir, String outputDir) { |
| XConfiguration conf = new XConfiguration(); |
| conf.setBoolean("hadoop.pipes.java.recordreader", true); |
| conf.setBoolean("hadoop.pipes.java.recordwriter", true); |
| conf.set("mapred.input.dir", inputDir); |
| conf.set("mapred.output.dir", outputDir); |
| return conf; |
| } |
| |
| private XConfiguration getOozieActionExternalStatsWriteProperty(String inputDir, String outputDir, |
| String oozieProperty) { |
| XConfiguration conf = new XConfiguration(); |
| conf.set("mapred.input.dir", inputDir); |
| conf.set("mapred.output.dir", outputDir); |
| conf.set("oozie.action.external.stats.write", oozieProperty); |
| return conf; |
| } |
| |
| public void testPipes() throws Exception { |
| Path programPath = new Path(getFsTestCaseDir(), "wordcount-simple"); |
| |
| FileSystem fs = getFileSystem(); |
| |
| InputStream is = Thread.currentThread().getContextClassLoader().getResourceAsStream("wordcount-simple"); |
| if (is != null) { |
| OutputStream os = fs.create(programPath); |
| IOUtils.copyStream(is, os); |
| |
| Path inputDir = new Path(getFsTestCaseDir(), "input"); |
| Path outputDir = new Path(getFsTestCaseDir(), "output"); |
| |
| writeDummyInput(fs, inputDir); |
| |
| String actionXml = "<map-reduce>" + "<job-tracker>" + getJobTrackerUri() + "</job-tracker>" + "<name-node>" |
| + getNameNodeUri() + "</name-node>" + " <pipes>" + " <program>" + programPath |
| + "#wordcount-simple" + "</program>" + " </pipes>" |
| + getPipesConfig(inputDir.toString(), outputDir.toString()).toXmlString(false) + "<file>" |
| + programPath + "</file>" + "</map-reduce>"; |
| _testSubmit(PIPES, actionXml); |
| } |
| else { |
| System.out.println( |
| "SKIPPING TEST: TestMapReduceActionExecutor.testPipes(), " + |
| "binary 'wordcount-simple' not available in the classpath"); |
| } |
| } |
| |
| // Test to assert that executionStats is set when user has specified stats |
| // write property as true. |
| public void testSetExecutionStats_when_user_has_specified_stats_write_TRUE() throws Exception { |
| FileSystem fs = getFileSystem(); |
| |
| Path inputDir = new Path(getFsTestCaseDir(), "input"); |
| Path outputDir = new Path(getFsTestCaseDir(), "output"); |
| |
| writeDummyInput(fs, inputDir); |
| |
| // set user stats write property as true explicitly in the |
| // configuration. |
| String actionXml = "<map-reduce>" |
| + "<job-tracker>" |
| + getJobTrackerUri() |
| + "</job-tracker>" |
| + "<name-node>" |
| + getNameNodeUri() |
| + "</name-node>" |
| + getOozieActionExternalStatsWriteProperty(inputDir.toString(), outputDir.toString(), "true") |
| .toXmlString(false) + "</map-reduce>"; |
| |
| Context context = createContext(MAP_REDUCE, actionXml); |
| final String launcherId = submitAction(context); |
| waitUntilYarnAppDoneAndAssertSuccess(launcherId); |
| |
| MapReduceActionExecutor ae = new MapReduceActionExecutor(); |
| Configuration conf = ae.createBaseHadoopConf(context, XmlUtils.parseXml(actionXml)); |
| |
| Map<String, String> actionData = LauncherHelper.getActionData(getFileSystem(), context.getActionDir(), |
| conf); |
| assertTrue(LauncherHelper.hasIdSwap(actionData)); |
| |
| ae.check(context, context.getAction()); |
| assertTrue(launcherId.equals(context.getAction().getExternalId())); |
| |
| waitUntilYarnAppDoneAndAssertSuccess(context.getAction().getExternalChildIDs()); |
| ae.check(context, context.getAction()); |
| |
| assertEquals(JavaActionExecutor.SUCCEEDED, context.getAction().getExternalStatus()); |
| assertNull(context.getAction().getData()); |
| |
| ae.end(context, context.getAction()); |
| assertEquals(WorkflowAction.Status.OK, context.getAction().getStatus()); |
| |
| // Assert for stats info stored in the context. |
| assertNotNull(context.getExecutionStats()); |
| assertTrue(context.getExecutionStats().contains("ACTION_TYPE")); |
| assertTrue(context.getExecutionStats().contains("Counter")); |
| |
| // External Child IDs used to be null, but after 4.0, become Non-Null in case of MR action. |
| assertNotNull(context.getExternalChildIDs()); |
| |
| // hadoop.counters will always be set in case of MR action. |
| assertNotNull(context.getVar("hadoop.counters")); |
| String counters = context.getVar("hadoop.counters"); |
| assertTrue(counters.contains("Counter")); |
| } |
| |
| // Test to assert that executionStats is not set when user has specified |
| // stats write property as false. |
| public void testSetExecutionStats_when_user_has_specified_stats_write_FALSE() throws Exception { |
| FileSystem fs = getFileSystem(); |
| |
| Path inputDir = new Path(getFsTestCaseDir(), "input"); |
| Path outputDir = new Path(getFsTestCaseDir(), "output"); |
| |
| writeDummyInput(fs, inputDir); |
| |
| // set user stats write property as false explicitly in the |
| // configuration. |
| String actionXml = "<map-reduce>" |
| + "<job-tracker>" |
| + getJobTrackerUri() |
| + "</job-tracker>" |
| + "<name-node>" |
| + getNameNodeUri() |
| + "</name-node>" |
| + getOozieActionExternalStatsWriteProperty(inputDir.toString(), outputDir.toString(), "false") |
| .toXmlString(false) + "</map-reduce>"; |
| |
| Context context = createContext(MAP_REDUCE, actionXml); |
| final String launcherId = submitAction(context); |
| waitUntilYarnAppDoneAndAssertSuccess(launcherId); |
| |
| Map<String, String> actionData = LauncherHelper.getActionData(getFileSystem(), context.getActionDir(), |
| context.getProtoActionConf()); |
| assertTrue(LauncherHelper.hasIdSwap(actionData)); |
| |
| MapReduceActionExecutor ae = new MapReduceActionExecutor(); |
| ae.check(context, context.getAction()); |
| assertTrue(launcherId.equals(context.getAction().getExternalId())); |
| |
| waitUntilYarnAppDoneAndAssertSuccess(context.getAction().getExternalChildIDs()); |
| ae.check(context, context.getAction()); |
| |
| assertEquals(JavaActionExecutor.SUCCEEDED, context.getAction().getExternalStatus()); |
| assertNull(context.getAction().getData()); |
| |
| ae.end(context, context.getAction()); |
| assertEquals(WorkflowAction.Status.OK, context.getAction().getStatus()); |
| |
| // Assert for stats info stored in the context. |
| assertNull(context.getExecutionStats()); |
| |
| // External Child IDs used to be null, but after 4.0, become Non-Null in case of MR action. |
| assertNotNull(context.getExternalChildIDs()); |
| |
| // hadoop.counters will always be set in case of MR action. |
| assertNotNull(context.getVar("hadoop.counters")); |
| String counters = context.getVar("hadoop.counters"); |
| assertTrue(counters.contains("Counter")); |
| } |
| |
| public void testEndWithoutConfiguration() throws Exception { |
| FileSystem fs = getFileSystem(); |
| |
| Path inputDir = new Path(getFsTestCaseDir(), "input"); |
| Path outputDir = new Path(getFsTestCaseDir(), "output"); |
| |
| writeDummyInput(fs, inputDir); |
| |
| // set user stats write property as false explicitly in the |
| // configuration. |
| String actionXml = "<map-reduce>" |
| + "<job-tracker>" |
| + getJobTrackerUri() |
| + "</job-tracker>" |
| + "<name-node>" |
| + getNameNodeUri() |
| + "</name-node>" |
| + getOozieActionExternalStatsWriteProperty(inputDir.toString(), outputDir.toString(), "false") |
| .toXmlString(false) + "</map-reduce>"; |
| |
| Context context = createContext(MAP_REDUCE, actionXml); |
| final String launcherId = submitAction(context); |
| waitUntilYarnAppDoneAndAssertSuccess(launcherId); |
| |
| Map<String, String> actionData = LauncherHelper.getActionData(getFileSystem(), context.getActionDir(), |
| context.getProtoActionConf()); |
| assertTrue(LauncherHelper.hasIdSwap(actionData)); |
| |
| MapReduceActionExecutor ae = new MapReduceActionExecutor(); |
| ae.check(context, context.getAction()); |
| assertTrue(launcherId.equals(context.getAction().getExternalId())); |
| |
| waitUntilYarnAppDoneAndAssertSuccess(context.getAction().getExternalChildIDs()); |
| ae.check(context, context.getAction()); |
| |
| assertEquals(JavaActionExecutor.SUCCEEDED, context.getAction().getExternalStatus()); |
| assertNull(context.getAction().getData()); |
| |
| actionXml = "<map-reduce>" |
| + "<job-tracker>" |
| + getJobTrackerUri() |
| + "</job-tracker>" |
| + "<name-node>" |
| + getNameNodeUri() |
| + "</name-node></map-reduce>"; |
| |
| WorkflowActionBean action = (WorkflowActionBean) context.getAction(); |
| action.setConf(actionXml); |
| |
| try { |
| ae.end(context, context.getAction()); |
| } |
| catch(Exception e) |
| { |
| fail("unexpected exception throwing " + e); |
| } |
| } |
| |
| /** |
| * Test "oozie.launcher.mapred.job.name" and "mapred.job.name" can be set in |
| * the action configuration and not overridden by the action executor |
| * |
| * @throws Exception |
| */ |
| public void testSetMapredJobName() throws Exception { |
| final String launcherJobName = "MapReduceLauncherTest"; |
| final String mapredJobName = "MapReduceTest"; |
| |
| FileSystem fs = getFileSystem(); |
| |
| Path inputDir = new Path(getFsTestCaseDir(), "input"); |
| Path outputDir = new Path(getFsTestCaseDir(), "output"); |
| |
| writeDummyInput(fs, inputDir); |
| |
| XConfiguration mrConfig = getMapReduceConfig(inputDir.toString(), |
| outputDir.toString()); |
| mrConfig.set("oozie.launcher.mapred.job.name", launcherJobName); |
| mrConfig.set("mapred.job.name", mapredJobName); |
| |
| StringBuilder sb = new StringBuilder("<map-reduce>") |
| .append("<job-tracker>").append(getJobTrackerUri()) |
| .append("</job-tracker>").append("<name-node>") |
| .append(getNameNodeUri()).append("</name-node>") |
| .append(mrConfig.toXmlString(false)).append("</map-reduce>"); |
| String actionXml = sb.toString(); |
| |
| Context context = createContext(MAP_REDUCE, actionXml); |
| final String launcherId = submitAction(context); |
| waitUntilYarnAppDoneAndAssertSuccess(launcherId); |
| |
| Map<String, String> actionData = LauncherHelper.getActionData(getFileSystem(), context.getActionDir(), |
| context.getProtoActionConf()); |
| assertTrue(LauncherHelper.hasIdSwap(actionData)); |
| |
| MapReduceActionExecutor ae = new MapReduceActionExecutor(); |
| ae.check(context, context.getAction()); |
| assertTrue(launcherId.equals(context.getAction().getExternalId())); |
| |
| String externalChildIDs = context.getAction().getExternalChildIDs(); |
| waitUntilYarnAppDoneAndAssertSuccess(externalChildIDs); |
| ae.check(context, context.getAction()); |
| |
| assertEquals(JavaActionExecutor.SUCCEEDED, context.getAction().getExternalStatus()); |
| assertNull(context.getAction().getData()); |
| |
| ae.end(context, context.getAction()); |
| assertEquals(WorkflowAction.Status.OK, context.getAction().getStatus()); |
| |
| Configuration conf = getHadoopAccessorService().createConfiguration(getJobTrackerUri()); |
| final YarnClient yarnClient = getHadoopAccessorService().createYarnClient(getTestUser(), conf); |
| ApplicationReport report = yarnClient.getApplicationReport(ConverterUtils.toApplicationId(externalChildIDs)); |
| // Assert Mapred job name has been set |
| assertEquals(mapredJobName, report.getName()); |
| |
| // Assert for stats info stored in the context. |
| assertNull(context.getExecutionStats()); |
| |
| // External Child IDs used to be null, but after 4.0, become Non-Null in case of MR action. |
| assertNotNull(context.getExternalChildIDs()); |
| |
| // hadoop.counters will always be set in case of MR action. |
| assertNotNull(context.getVar("hadoop.counters")); |
| String counters = context.getVar("hadoop.counters"); |
| assertTrue(counters.contains("Counter")); |
| } |
| |
| public void testDefaultShareLibName() { |
| MapReduceActionExecutor ae = new MapReduceActionExecutor(); |
| Element e = new Element("mapreduce"); |
| assertNull(ae.getDefaultShareLibName(e)); |
| e.addContent(new Element("streaming")); |
| assertEquals("mapreduce-streaming", ae.getDefaultShareLibName(e)); |
| } |
| |
| /** |
| * https://issues.apache.org/jira/browse/OOZIE-87 |
| * This test covers map-reduce action |
| * @throws Exception |
| */ |
| public void testCommaSeparatedFilesAndArchives() throws Exception { |
| Path root = new Path(getFsTestCaseDir(), "root"); |
| |
| Path jar = new Path("jar.jar"); |
| getFileSystem().create(new Path(getAppPath(), jar)).close(); |
| Path rootJar = new Path(root, "rootJar.jar"); |
| getFileSystem().create(rootJar).close(); |
| |
| Path file = new Path("file"); |
| getFileSystem().create(new Path(getAppPath(), file)).close(); |
| Path rootFile = new Path(root, "rootFile"); |
| getFileSystem().create(rootFile).close(); |
| |
| Path so = new Path("soFile.so"); |
| getFileSystem().create(new Path(getAppPath(), so)).close(); |
| Path rootSo = new Path(root, "rootSoFile.so"); |
| getFileSystem().create(rootSo).close(); |
| |
| Path so1 = new Path("soFile.so.1"); |
| getFileSystem().create(new Path(getAppPath(), so1)).close(); |
| Path rootSo1 = new Path(root, "rootSoFile.so.1"); |
| getFileSystem().create(rootSo1).close(); |
| |
| Path archive = new Path("archive.tar"); |
| getFileSystem().create(new Path(getAppPath(), archive)).close(); |
| Path rootArchive = new Path(root, "rootArchive.tar"); |
| getFileSystem().create(rootArchive).close(); |
| |
| String actionXml = "<map-reduce>" + |
| " <job-tracker>" + getJobTrackerUri() + "</job-tracker>" + |
| " <name-node>" + getNameNodeUri() + "</name-node>" + |
| " <main-class>CLASS</main-class>" + |
| " <file>" + jar.toString() + |
| "," + rootJar.toString() + |
| "," + file.toString() + |
| ", " + rootFile.toString() + // with leading and trailing spaces |
| " ," + so.toString() + |
| "," + rootSo.toString() + |
| "," + so1.toString() + |
| "," + rootSo1.toString() + "</file>\n" + |
| " <archive>" + archive.toString() + ", " |
| + rootArchive.toString() + " </archive>\n" + // with leading and trailing spaces |
| "</map-reduce>"; |
| |
| Element eActionXml = XmlUtils.parseXml(actionXml); |
| |
| Context context = createContext(MAP_REDUCE, actionXml); |
| |
| Path appPath = getAppPath(); |
| |
| MapReduceActionExecutor ae = new MapReduceActionExecutor(); |
| |
| Configuration jobConf = ae.createBaseHadoopConf(context, eActionXml); |
| ae.setupActionConf(jobConf, context, eActionXml, appPath); |
| ae.setLibFilesArchives(context, eActionXml, appPath, jobConf); |
| |
| |
| assertTrue(DistributedCache.getSymlink(jobConf)); |
| |
| Path[] filesInClasspath = DistributedCache.getFileClassPaths(jobConf); |
| for (Path p : new Path[]{new Path(getAppPath(), jar), rootJar}) { |
| boolean found = false; |
| for (Path c : filesInClasspath) { |
| if (!found && p.toUri().getPath().equals(c.toUri().getPath())) { |
| found = true; |
| } |
| } |
| assertTrue("file " + p.toUri().getPath() + " not found in classpath", found); |
| } |
| for (Path p : new Path[]{new Path(getAppPath(), file), rootFile, new Path(getAppPath(), so), rootSo, |
| new Path(getAppPath(), so1), rootSo1}) { |
| boolean found = false; |
| for (Path c : filesInClasspath) { |
| if (!found && p.toUri().getPath().equals(c.toUri().getPath())) { |
| found = true; |
| } |
| } |
| assertFalse("file " + p.toUri().getPath() + " found in classpath", found); |
| } |
| |
| URI[] filesInCache = DistributedCache.getCacheFiles(jobConf); |
| for (Path p : new Path[]{new Path(getAppPath(), jar), rootJar, new Path(getAppPath(), file), rootFile, |
| new Path(getAppPath(), so), rootSo, new Path(getAppPath(), so1), rootSo1}) { |
| boolean found = false; |
| for (URI c : filesInCache) { |
| if (!found && p.toUri().getPath().equals(c.getPath())) { |
| found = true; |
| } |
| } |
| assertTrue("file " + p.toUri().getPath() + " not found in cache", found); |
| } |
| |
| URI[] archivesInCache = DistributedCache.getCacheArchives(jobConf); |
| for (Path p : new Path[]{new Path(getAppPath(), archive), rootArchive}) { |
| boolean found = false; |
| for (URI c : archivesInCache) { |
| if (!found && p.toUri().getPath().equals(c.getPath())) { |
| found = true; |
| } |
| } |
| assertTrue("archive " + p.toUri().getPath() + " not found in cache", found); |
| } |
| } |
| |
| private void waitForWorkflowAction(final String actionId) { |
| waitFor(3 * 60 * 1000, new Predicate() { |
| public boolean evaluate() throws Exception { |
| WorkflowActionBean mrAction = WorkflowActionQueryExecutor.getInstance() |
| .get(WorkflowActionQuery.GET_ACTION, actionId); |
| return mrAction.inTerminalState() || mrAction.getStatus() == Status.RUNNING; |
| } |
| }); |
| } |
| |
| public void testFailingMapReduceJobCausesOozieLauncherAMToFail() throws Exception { |
| final String workflowUri = createWorkflowWithMapReduceAction(); |
| |
| startWorkflowAndFailChildMRJob(workflowUri); |
| } |
| |
| private String createWorkflowWithMapReduceAction() throws IOException { |
| final String workflowUri = getTestCaseFileUri("workflow.xml"); |
| final String appXml = "<workflow-app xmlns=\"uri:oozie:workflow:1.0\" name=\"workflow\">" + |
| " <start to=\"map-reduce\"/>" + |
| " <action name=\"map-reduce\">" + |
| " <map-reduce>" + |
| " <resource-manager>" + getJobTrackerUri() + "</resource-manager>" + |
| " <name-node>" + getNameNodeUri() + "</name-node>" + |
| " <configuration>\n" + |
| " <property>\n" + |
| " <name>mapred.job.queue.name</name>\n" + |
| " <value>default</value>\n" + |
| " </property>\n" + |
| " <property>\n" + |
| " <name>mapred.mapper.class</name>\n" + |
| " <value>org.apache.oozie.action.hadoop.SleepMapperReducerForTest</value>\n" + |
| " </property>\n" + |
| " <property>\n" + |
| " <name>mapred.reducer.class</name>\n" + |
| " <value>org.apache.oozie.action.hadoop.SleepMapperReducerForTest</value>\n" + |
| " </property>\n" + |
| " <property>\n" + |
| " <name>mapred.input.dir</name>\n" + |
| " <value>" + getFsTestCaseDir() + "/input</value>\n" + |
| " </property>\n" + |
| " <property>\n" + |
| " <name>mapred.output.dir</name>\n" + |
| " <value>" + getFsTestCaseDir() + "/output</value>\n" + |
| " </property>\n" + |
| " </configuration>\n" + |
| " </map-reduce>" + |
| " <ok to=\"end\"/>" + |
| " <error to=\"fail\"/>" + |
| " </action>" + |
| " <kill name=\"fail\">" + |
| " <message>Sub workflow failed, error message[${wf:errorMessage(wf:lastErrorNode())}]</message>" + |
| " </kill>" + |
| " <end name=\"end\"/>" + |
| "</workflow-app>"; |
| |
| writeToFile(appXml, workflowUri); |
| |
| return workflowUri; |
| } |
| |
| private void startWorkflowAndFailChildMRJob(final String workflowUri) throws Exception { |
| try { |
| LocalOozie.start(); |
| final OozieClient wfClient = LocalOozie.getClient(); |
| final String workflowId = submitWorkflow(workflowUri, wfClient); |
| final Configuration conf = Services.get().get(HadoopAccessorService.class).createConfiguration(getJobTrackerUri()); |
| |
| final Path inputFolder = createInputFolder(conf); |
| |
| waitForWorkflowToStart(wfClient, workflowId); |
| waitForChildYarnApplication(getHadoopAccessorService().createYarnClient(getTestUser(), conf), workflowId); |
| assertAndWriteNextMRJobId(workflowId, conf, inputFolder); |
| |
| final ApplicationId externalChildJobId = getChildMRJobApplicationId(conf); |
| |
| killYarnApplication(conf, externalChildJobId); |
| waitUntilYarnAppKilledAndAssertSuccess(externalChildJobId.toString()); |
| waitForWorkflowToKill(wfClient, workflowId); |
| } finally { |
| LocalOozie.stop(); |
| } |
| } |
| |
| /** |
| * Get all YARN application IDs, select the one of type {@code MAPREDUCE} that is relevant to {@code workflowId}, |
| * and write to {@code inputFolder/jobID.txt}. |
| * <p> |
| * Simulating functional parts of {@link LauncherMain#writeExternalChildIDs(String, Pattern[], String)} in order |
| * {@link MapReduceActionExecutor#check(ActionExecutor.Context, WorkflowAction)} can find it later on the call chain. |
| * <p> |
| * We need to write out an own sequence file to {@link LauncherMainTester#JOB_ID_FILE_NAME} in order |
| * {@link ActionExecutorTestCase#getChildMRJobApplicationId(Configuration)} can find it. We unfortunately cannot rely on the |
| * original sequence file written by {@link LauncherMain#writeExternalChildIDs(String, Pattern[], String)} because we don't own |
| * a reference to the original {@link ActionExecutor.Context} as in {@link MapReduceActionExecutor}. |
| * @param workflowId the workflow ID |
| * @param conf the {@link Configuration} used for Hadoop Common / YARN API calls |
| * @param inputFolder where to write the output text file |
| * @throws IOException when the output text file cannot be written |
| * @throws YarnException when the list of YARN applications cannot be queried |
| * @throws HadoopAccessorException when {@link YarnClient} cannot be created |
| */ |
| private void assertAndWriteNextMRJobId(final String workflowId, final Configuration conf, final Path inputFolder) |
| throws IOException, YarnException, HadoopAccessorException { |
| final Path wfIDFile = new Path(inputFolder, LauncherMainTester.JOB_ID_FILE_NAME); |
| try (final FileSystem fs = FileSystem.get(conf); |
| final Writer w = new OutputStreamWriter(fs.create(wfIDFile), StandardCharsets.UTF_8)) { |
| final List<ApplicationReport> allApplications = |
| getHadoopAccessorService().createYarnClient(getTestUser(), conf).getApplications(); |
| |
| assertTrue("YARN applications number mismatch", allApplications.size() >= 2); |
| |
| ApplicationReport mapReduce = null; |
| for (final ApplicationReport candidate : allApplications) { |
| if (candidate.getApplicationType().equals(MapReduceActionExecutor.YARN_APPLICATION_TYPE_MAPREDUCE) |
| && candidate.getName().contains(workflowId)) { |
| mapReduce = candidate; |
| } |
| } |
| assertNotNull("MAPREDUCE YARN application not found", mapReduce); |
| |
| final String applicationId = mapReduce.getApplicationId().toString(); |
| final String nextMRJobId = applicationId.replace("application", "job"); |
| |
| LOG.debug("Writing next MapReduce job ID: {0}", nextMRJobId); |
| |
| w.write(nextMRJobId); |
| } |
| } |
| |
| private Path createInputFolder(final Configuration conf) throws IOException { |
| final Path inputDir = new Path(getFsTestCaseDir(), "input"); |
| try (final FileSystem fs = FileSystem.get(conf)) { |
| fs.mkdirs(inputDir); |
| } |
| return inputDir; |
| } |
| |
| private void waitForChildYarnApplication(final YarnClient yarnClient, final String workflowId) { |
| waitFor(JOB_TIMEOUT, new ChildYarnApplicationPresentPredicate(yarnClient, workflowId)); |
| } |
| |
| private class ChildYarnApplicationPresentPredicate implements Predicate { |
| private final YarnClient yarnClient; |
| private final String workflowId; |
| |
| ChildYarnApplicationPresentPredicate(final YarnClient yarnClient, final String workflowId) { |
| this.yarnClient = yarnClient; |
| this.workflowId = workflowId; |
| } |
| |
| @Override |
| public boolean evaluate() throws Exception { |
| if (yarnClient.getApplications().isEmpty()) { |
| return false; |
| } |
| |
| for (final ApplicationReport applicationReport : yarnClient.getApplications()) { |
| final String name = applicationReport.getName(); |
| final String type = applicationReport.getApplicationType(); |
| if (type.equals(MapReduceActionExecutor.YARN_APPLICATION_TYPE_MAPREDUCE) && name.contains(workflowId)) { |
| return true; |
| } |
| } |
| |
| return false; |
| } |
| } |
| } |
