You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@oozie.apache.org by ge...@apache.org on 2016/10/10 11:52:32 UTC
[33/50] [abbrv] oozie git commit: OOZIE-2594 Make MapReduce action
work, small refactors, remove RunningJob from test cases,
test fixes. Follow up: OOZIE-2686
OOZIE-2594 Make MapReduce action work, small refactors, remove RunningJob from test cases, test fixes. Follow up: OOZIE-2686
Change-Id: I797963d65bc248c81c9e9d0b2a48a68dd2bab5cf
Project: http://git-wip-us.apache.org/repos/asf/oozie/repo
Commit: http://git-wip-us.apache.org/repos/asf/oozie/commit/ca7e56fd
Tree: http://git-wip-us.apache.org/repos/asf/oozie/tree/ca7e56fd
Diff: http://git-wip-us.apache.org/repos/asf/oozie/diff/ca7e56fd
Branch: refs/heads/oya
Commit: ca7e56fdccbca80ce2f9b87812c15305ca9d09d0
Parents: 2fddebb
Author: Peter Bacsko <pb...@cloudera.com>
Authored: Mon Sep 12 11:29:14 2016 +0200
Committer: Peter Bacsko <pb...@cloudera.com>
Committed: Mon Sep 26 14:11:28 2016 +0200
----------------------------------------------------------------------
.../oozie/action/hadoop/JavaActionExecutor.java | 117 ++++----
.../action/hadoop/MapReduceActionExecutor.java | 79 +++++-
.../apache/oozie/servlet/CallbackServlet.java | 4 +-
.../action/hadoop/TestDistCpActionExecutor.java | 29 +-
.../action/hadoop/TestJavaActionExecutor.java | 12 +-
.../hadoop/TestLauncherAMCallbackNotifier.java | 148 ++++++----
.../action/hadoop/TestMapReduceActionError.java | 173 ------------
.../oozie/action/hadoop/TestOozieJobInfo.java | 19 +-
.../action/hadoop/TestShellActionExecutor.java | 46 +---
.../command/wf/TestActionCheckXCommand.java | 7 +-
.../command/wf/TestActionStartXCommand.java | 26 +-
.../java/org/apache/oozie/test/XFsTestCase.java | 27 ++
.../java/org/apache/oozie/test/XTestCase.java | 10 +-
.../action/hadoop/TestHiveActionExecutor.java | 39 +--
.../action/hadoop/TestHive2ActionExecutor.java | 40 +--
.../apache/oozie/action/hadoop/LauncherAM.java | 274 +++++++++++--------
.../hadoop/LauncherAMCallbackNotifier.java | 15 +-
.../oozie/action/hadoop/LauncherMain.java | 8 +-
.../oozie/action/hadoop/MapReduceMain.java | 1 -
.../org/apache/oozie/action/hadoop/PigMain.java | 2 +-
.../action/hadoop/TestPigActionExecutor.java | 58 +---
.../apache/oozie/action/hadoop/TestPyspark.java | 19 +-
.../action/hadoop/TestSparkActionExecutor.java | 21 +-
.../action/hadoop/TestSqoopActionExecutor.java | 51 +---
.../hadoop/TestMapReduceActionExecutor.java | 267 +++++++++---------
25 files changed, 645 insertions(+), 847 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/oozie/blob/ca7e56fd/core/src/main/java/org/apache/oozie/action/hadoop/JavaActionExecutor.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/oozie/action/hadoop/JavaActionExecutor.java b/core/src/main/java/org/apache/oozie/action/hadoop/JavaActionExecutor.java
index 6a28406..8637f64 100644
--- a/core/src/main/java/org/apache/oozie/action/hadoop/JavaActionExecutor.java
+++ b/core/src/main/java/org/apache/oozie/action/hadoop/JavaActionExecutor.java
@@ -18,6 +18,28 @@
package org.apache.oozie.action.hadoop;
+import java.io.FileNotFoundException;
+import java.io.IOException;
+import java.io.StringReader;
+import java.net.ConnectException;
+import java.net.URI;
+import java.net.URISyntaxException;
+import java.net.UnknownHostException;
+import java.nio.ByteBuffer;
+import java.text.MessageFormat;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.Map.Entry;
+import java.util.Properties;
+import java.util.Set;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+
import org.apache.commons.io.IOUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.filecache.DistributedCache;
@@ -31,6 +53,7 @@ import org.apache.hadoop.mapred.JobClient;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.JobID;
import org.apache.hadoop.mapred.RunningJob;
+import org.apache.hadoop.mapred.TaskLog;
import org.apache.hadoop.mapreduce.filecache.ClientDistributedCacheManager;
import org.apache.hadoop.mapreduce.v2.util.MRApps;
import org.apache.hadoop.security.token.Token;
@@ -48,6 +71,7 @@ import org.apache.hadoop.yarn.api.records.Resource;
import org.apache.hadoop.yarn.api.records.YarnApplicationState;
import org.apache.hadoop.yarn.client.api.YarnClient;
import org.apache.hadoop.yarn.client.api.YarnClientApplication;
+import org.apache.hadoop.yarn.conf.YarnConfiguration;
import org.apache.hadoop.yarn.util.ConverterUtils;
import org.apache.hadoop.yarn.util.Records;
import org.apache.oozie.WorkflowActionBean;
@@ -78,38 +102,21 @@ import org.jdom.Element;
import org.jdom.JDOMException;
import org.jdom.Namespace;
-import java.io.FileNotFoundException;
-import java.io.IOException;
-import java.io.StringReader;
-import java.net.ConnectException;
-import java.net.URI;
-import java.net.URISyntaxException;
-import java.net.UnknownHostException;
-import java.nio.ByteBuffer;
-import java.text.MessageFormat;
-import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.HashMap;
-import java.util.HashSet;
-import java.util.Iterator;
-import java.util.List;
-import java.util.Map;
-import java.util.Map.Entry;
-import java.util.Properties;
-import java.util.Set;
-import java.util.regex.Matcher;
-import java.util.regex.Pattern;
+import com.google.common.collect.ImmutableList;
public class JavaActionExecutor extends ActionExecutor {
- protected static final String HADOOP_USER = "user.name";
+ public static final String RUNNING = "RUNNING";
+ public static final String SUCCEEDED = "SUCCEEDED";
+ public static final String KILLED = "KILLED";
+ public static final String FAILED = "FAILED";
+ public static final String FAILED_KILLED = "FAILED/KILLED";
public static final String HADOOP_YARN_RM = "yarn.resourcemanager.address";
public static final String HADOOP_NAME_NODE = "fs.default.name";
- private static final String HADOOP_JOB_NAME = "mapred.job.name";
public static final String OOZIE_COMMON_LIBDIR = "oozie";
- private static final Set<String> DISALLOWED_PROPERTIES = new HashSet<String>();
- public final static String MAX_EXTERNAL_STATS_SIZE = "oozie.external.stats.max.size";
+
+ public static final String MAX_EXTERNAL_STATS_SIZE = "oozie.external.stats.max.size";
public static final String ACL_VIEW_JOB = "mapreduce.job.acl-view-job";
public static final String ACL_MODIFY_JOB = "mapreduce.job.acl-modify-job";
public static final String HADOOP_YARN_TIMELINE_SERVICE_ENABLED = "yarn.timeline-service.enabled";
@@ -120,24 +127,27 @@ public class JavaActionExecutor extends ActionExecutor {
public static final String HADOOP_REDUCE_JAVA_OPTS = "mapreduce.reduce.java.opts";
public static final String HADOOP_CHILD_JAVA_ENV = "mapred.child.env";
public static final String HADOOP_MAP_JAVA_ENV = "mapreduce.map.env";
+ public static final String HADOOP_JOB_CLASSLOADER = "mapreduce.job.classloader";
+ public static final String HADOOP_USER_CLASSPATH_FIRST = "mapreduce.user.classpath.first";
+ public static final String OOZIE_CREDENTIALS_SKIP = "oozie.credentials.skip";
public static final String YARN_AM_RESOURCE_MB = "yarn.app.mapreduce.am.resource.mb";
public static final String YARN_AM_COMMAND_OPTS = "yarn.app.mapreduce.am.command-opts";
public static final String YARN_AM_ENV = "yarn.app.mapreduce.am.env";
- private static final String JAVA_MAIN_CLASS_NAME = "org.apache.oozie.action.hadoop.JavaMain";
public static final int YARN_MEMORY_MB_MIN = 512;
+
+ private static final String JAVA_MAIN_CLASS_NAME = "org.apache.oozie.action.hadoop.JavaMain";
+ private static final String HADOOP_JOB_NAME = "mapred.job.name";
+ private static final Set<String> DISALLOWED_PROPERTIES = new HashSet<String>();
+
private static int maxActionOutputLen;
private static int maxExternalStatsSize;
private static int maxFSGlobMax;
- private static final String SUCCEEDED = "SUCCEEDED";
- private static final String KILLED = "KILLED";
- private static final String FAILED = "FAILED";
- private static final String FAILED_KILLED = "FAILED/KILLED";
+
+ protected static final String HADOOP_USER = "user.name";
+
protected XLog LOG = XLog.getLog(getClass());
private static final Pattern heapPattern = Pattern.compile("-Xmx(([0-9]+)[mMgG])");
private static final String JAVA_TMP_DIR_SETTINGS = "-Djava.io.tmpdir=";
- public static final String HADOOP_JOB_CLASSLOADER = "mapreduce.job.classloader";
- public static final String HADOOP_USER_CLASSPATH_FIRST = "mapreduce.user.classpath.first";
- public static final String OOZIE_CREDENTIALS_SKIP = "oozie.credentials.skip";
static {
DISALLOWED_PROPERTIES.add(HADOOP_USER);
@@ -237,6 +247,13 @@ public class JavaActionExecutor extends ActionExecutor {
conf.set(HADOOP_YARN_RM, jobTracker);
conf.set(HADOOP_NAME_NODE, nameNode);
conf.set("mapreduce.fileoutputcommitter.marksuccessfuljobs", "true");
+
+ // FIXME - think about this!
+ Element e = actionXml.getChild("config-class", ns);
+ if (e != null) {
+ conf.set(LauncherMapper.OOZIE_ACTION_CONFIG_CLASS, e.getTextTrim());
+ }
+
return conf;
}
@@ -308,6 +325,7 @@ public class JavaActionExecutor extends ActionExecutor {
}
}
+ // FIXME: is this needed?
private HashMap<String, List<String>> populateEnvMap(String input) {
HashMap<String, List<String>> envMaps = new HashMap<String, List<String>>();
String[] envEntries = input.split(",");
@@ -918,7 +936,7 @@ public class JavaActionExecutor extends ActionExecutor {
}
}
- private void injectCallback(Context context, Configuration conf) {
+ protected void injectCallback(Context context, Configuration conf) {
String callback = context.getCallbackUrl(LauncherAMCallbackNotifier.OOZIE_LAUNCHER_CALLBACK_JOBSTATUS_TOKEN);
conf.set(LauncherAMCallbackNotifier.OOZIE_LAUNCHER_CALLBACK_URL, callback);
}
@@ -1109,6 +1127,7 @@ public class JavaActionExecutor extends ActionExecutor {
private ApplicationSubmissionContext createAppSubmissionContext(ApplicationId appId, JobConf launcherJobConf, String user,
Context context, Configuration actionConf)
throws IOException, HadoopAccessorException, URISyntaxException {
+
// Create launch context for app master
ApplicationSubmissionContext appContext = Records.newRecord(ApplicationSubmissionContext.class);
@@ -1149,9 +1168,10 @@ public class JavaActionExecutor extends ActionExecutor {
Map<String, String> env = new HashMap<String, String>();
// This adds the Hadoop jars to the classpath in the Launcher JVM
ClasspathUtils.setupClasspath(env, launcherJobConf);
- if (false) { // TODO: OYA: config to add MR jars? Probably also needed for MR Action
- ClasspathUtils.addMapReduceToClasspath(env, launcherJobConf);
- }
+
+ // FIXME: move this to specific places where it's actually needed - keeping it here for now
+ ClasspathUtils.addMapReduceToClasspath(env, launcherJobConf);
+
amContainer.setEnvironment(env);
// Set the command
@@ -1160,18 +1180,28 @@ public class JavaActionExecutor extends ActionExecutor {
+ "/bin/java");
// TODO: OYA: remove attach debugger to AM; useful for debugging
// vargs.add("-agentlib:jdwp=transport=dt_socket,server=y,suspend=y,address=5005");
- MRApps.addLog4jSystemProperties("INFO", 1024 * 1024, 0, vargs);
- vargs.add(LauncherAM.class.getName());
+
+ // FIXME: decide what to do with this method call - signature keeps changing
+ // MRApps.addLog4jSystemProperties("INFO", 1024 * 1024, 0, vargs, null);
+
+ vargs.add("-Dlog4j.configuration=container-log4j.properties");
+ vargs.add("-Dlog4j.debug=true");
+ vargs.add("-D" + YarnConfiguration.YARN_APP_CONTAINER_LOG_DIR + "=" + ApplicationConstants.LOG_DIR_EXPANSION_VAR);
+ vargs.add("-D" + YarnConfiguration.YARN_APP_CONTAINER_LOG_SIZE + "=" + 1024 * 1024);
+ vargs.add("-Dhadoop.root.logger=INFO,CLA");
+ vargs.add("-Dhadoop.root.logfile=" + TaskLog.LogName.SYSLOG);
+ vargs.add("-Dsubmitter.user=" + context.getWorkflow().getUser());
+ vargs.add("org.apache.oozie.action.hadoop.LauncherAM"); // note: using string temporarily so we don't have to depend on sharelib-oozie
vargs.add("1>" + ApplicationConstants.LOG_DIR_EXPANSION_VAR +
Path.SEPARATOR + ApplicationConstants.STDOUT);
vargs.add("2>" + ApplicationConstants.LOG_DIR_EXPANSION_VAR +
Path.SEPARATOR + ApplicationConstants.STDERR);
- List<String> vargsFinal = new ArrayList<String>(6);
StringBuilder mergedCommand = new StringBuilder();
for (CharSequence str : vargs) {
mergedCommand.append(str).append(" ");
}
- vargsFinal.add(mergedCommand.toString());
+
+ List<String> vargsFinal = ImmutableList.of(mergedCommand.toString());
LOG.debug("Command to launch container for ApplicationMaster is : "
+ mergedCommand);
amContainer.setCommands(vargsFinal);
@@ -1405,11 +1435,6 @@ public class JavaActionExecutor extends ActionExecutor {
return Services.get().get(HadoopAccessorService.class).createYarnClient(user, jobConf);
}
- protected RunningJob getRunningJob(Context context, WorkflowAction action, JobClient jobClient) throws Exception{
- RunningJob runningJob = jobClient.getJob(JobID.forName(action.getExternalId()));
- return runningJob;
- }
-
/**
* Useful for overriding in actions that do subsequent job runs
* such as the MapReduce Action, where the launcher job is not the
http://git-wip-us.apache.org/repos/asf/oozie/blob/ca7e56fd/core/src/main/java/org/apache/oozie/action/hadoop/MapReduceActionExecutor.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/oozie/action/hadoop/MapReduceActionExecutor.java b/core/src/main/java/org/apache/oozie/action/hadoop/MapReduceActionExecutor.java
index 4553351..019c4d9 100644
--- a/core/src/main/java/org/apache/oozie/action/hadoop/MapReduceActionExecutor.java
+++ b/core/src/main/java/org/apache/oozie/action/hadoop/MapReduceActionExecutor.java
@@ -21,7 +21,9 @@ package org.apache.oozie.action.hadoop;
import java.io.IOException;
import java.io.StringReader;
import java.util.ArrayList;
+import java.util.Collections;
import java.util.List;
+import java.util.Map;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
@@ -312,13 +314,80 @@ public class MapReduceActionExecutor extends JavaActionExecutor {
}
@Override
- protected RunningJob getRunningJob(Context context, WorkflowAction action, JobClient jobClient) throws Exception{
+ protected void injectCallback(Context context, Configuration conf) {
+ // add callback for the MapReduce job
+ String callback = context.getCallbackUrl("$jobStatus");
+ if (conf.get("job.end.notification.url") != null) {
+ LOG.warn("Overriding the action job end notification URI");
+ }
+ conf.set("job.end.notification.url", callback);
+
+ super.injectCallback(context, conf);
+ }
- RunningJob runningJob;
- String jobId = getActualExternalId(action);
+ @Override
+ public void check(Context context, WorkflowAction action) throws ActionExecutorException {
+ Map<String, String> actionData = Collections.emptyMap();
+ JobConf jobConf = null;
- runningJob = jobClient.getJob(JobID.forName(jobId));
+ try {
+ FileSystem actionFs = context.getAppFileSystem();
+ Element actionXml = XmlUtils.parseXml(action.getConf());
+ jobConf = createBaseHadoopConf(context, actionXml);
+ Path actionDir = context.getActionDir();
+ actionData = LauncherMapperHelper.getActionData(actionFs, actionDir, jobConf);
+ } catch (Exception e) {
+ LOG.warn("Exception in check(). Message[{0}]", e.getMessage(), e);
+ throw convertException(e);
+ }
- return runningJob;
+ final String newId = actionData.get(LauncherMapper.ACTION_DATA_NEW_ID);
+
+ // check the Hadoop job if newID is defined (which should be the case here) - otherwise perform the normal check()
+ if (newId != null) {
+ boolean jobCompleted;
+ JobClient jobClient = null;
+ boolean exception = false;
+
+ try {
+ jobClient = createJobClient(context, jobConf);
+ RunningJob runningJob = jobClient.getJob(JobID.forName(newId));
+
+ if (runningJob == null) {
+ context.setExternalStatus(FAILED);
+ throw new ActionExecutorException(ActionExecutorException.ErrorType.FAILED, "JA017",
+ "Unknown hadoop job [{0}] associated with action [{1}]. Failing this action!", newId,
+ action.getId());
+ }
+
+ jobCompleted = runningJob.isComplete();
+ } catch (Exception e) {
+ LOG.warn("Exception in check(). Message[{0}]", e.getMessage(), e);
+ exception = true;
+ throw convertException(e);
+ } finally {
+ if (jobClient != null) {
+ try {
+ jobClient.close();
+ } catch (Exception e) {
+ if (exception) {
+ LOG.error("JobClient error (not re-throwing due to a previous error): ", e);
+ } else {
+ throw convertException(e);
+ }
+ }
+ }
+ }
+
+ // run original check() if the MR action is completed or there are errors - otherwise mark it as RUNNING
+ if (jobCompleted || (!jobCompleted && actionData.containsKey(LauncherMapper.ACTION_DATA_ERROR_PROPS))) {
+ super.check(context, action);
+ } else {
+ context.setExternalStatus(RUNNING);
+ context.setExternalChildIDs(newId);
+ }
+ } else {
+ super.check(context, action);
+ }
}
}
http://git-wip-us.apache.org/repos/asf/oozie/blob/ca7e56fd/core/src/main/java/org/apache/oozie/servlet/CallbackServlet.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/oozie/servlet/CallbackServlet.java b/core/src/main/java/org/apache/oozie/servlet/CallbackServlet.java
index 6123021..86ba467 100644
--- a/core/src/main/java/org/apache/oozie/servlet/CallbackServlet.java
+++ b/core/src/main/java/org/apache/oozie/servlet/CallbackServlet.java
@@ -40,10 +40,12 @@ import org.apache.oozie.util.PropertiesUtils;
import org.apache.oozie.util.XLog;
public class CallbackServlet extends JsonRestServlet {
+ private static final long serialVersionUID = 6439106936153152786L;
+
private static final String INSTRUMENTATION_NAME = "callback";
private static final ResourceInfo RESOURCE_INFO =
- new ResourceInfo("", Arrays.asList("POST", "GET"), Collections.EMPTY_LIST);
+ new ResourceInfo("", Arrays.asList("POST", "GET"), Collections.<ParameterInfo>emptyList());
public final static String CONF_MAX_DATA_LEN = "oozie.servlet.CallbackServlet.max.data.len";
http://git-wip-us.apache.org/repos/asf/oozie/blob/ca7e56fd/core/src/test/java/org/apache/oozie/action/hadoop/TestDistCpActionExecutor.java
----------------------------------------------------------------------
diff --git a/core/src/test/java/org/apache/oozie/action/hadoop/TestDistCpActionExecutor.java b/core/src/test/java/org/apache/oozie/action/hadoop/TestDistCpActionExecutor.java
index e8a140f..c1f0e6f 100644
--- a/core/src/test/java/org/apache/oozie/action/hadoop/TestDistCpActionExecutor.java
+++ b/core/src/test/java/org/apache/oozie/action/hadoop/TestDistCpActionExecutor.java
@@ -25,18 +25,10 @@ import java.io.OutputStream;
import java.util.Arrays;
import org.apache.hadoop.fs.Path;
-import org.apache.hadoop.mapred.JobClient;
-import org.apache.hadoop.mapred.JobConf;
-import org.apache.hadoop.mapred.JobID;
-import org.apache.hadoop.mapred.RunningJob;
import org.apache.oozie.WorkflowActionBean;
import org.apache.oozie.WorkflowJobBean;
-import org.apache.oozie.action.hadoop.ActionExecutorTestCase.Context;
import org.apache.oozie.client.WorkflowAction;
-import org.apache.oozie.service.HadoopAccessorService;
-import org.apache.oozie.service.Services;
import org.apache.oozie.service.WorkflowAppService;
-import org.apache.oozie.test.XTestCase.Predicate;
import org.apache.oozie.util.IOUtils;
import org.apache.oozie.util.XConfiguration;
@@ -64,13 +56,8 @@ public class TestDistCpActionExecutor extends ActionExecutorTestCase{
"<arg>" + outputPath + "</arg>" +
"</distcp>";
Context context = createContext(actionXml);
- final RunningJob runningJob = submitAction(context);
- waitFor(60 * 1000, new Predicate() {
- public boolean evaluate() throws Exception {
- return runningJob.isComplete();
- }
- });
- assertTrue(runningJob.isSuccessful());
+ final String launcherId = submitAction(context);
+ waitUntilYarnAppDoneAndAssertSuccess(launcherId);
waitFor(60 * 1000, new Predicate() {
public boolean evaluate() throws Exception {
@@ -139,7 +126,7 @@ public class TestDistCpActionExecutor extends ActionExecutorTestCase{
return new Context(wf, action);
}
- protected RunningJob submitAction(Context context) throws Exception {
+ protected String submitAction(Context context) throws Exception {
DistcpActionExecutor ae = new DistcpActionExecutor();
WorkflowAction action = context.getAction();
@@ -154,14 +141,8 @@ public class TestDistCpActionExecutor extends ActionExecutorTestCase{
assertNotNull(jobTracker);
assertNotNull(consoleUrl);
- JobConf jobConf = Services.get().get(HadoopAccessorService.class).createJobConf(jobTracker);
- jobConf.set("mapred.job.tracker", jobTracker);
-
- JobClient jobClient =
- Services.get().get(HadoopAccessorService.class).createJobClient(getTestUser(), jobConf);
- final RunningJob runningJob = jobClient.getJob(JobID.forName(jobId));
- assertNotNull(runningJob);
- return runningJob;
+ ae.submitLauncher(getFileSystem(), context, context.getAction());
+ return context.getAction().getExternalId();
}
}
http://git-wip-us.apache.org/repos/asf/oozie/blob/ca7e56fd/core/src/test/java/org/apache/oozie/action/hadoop/TestJavaActionExecutor.java
----------------------------------------------------------------------
diff --git a/core/src/test/java/org/apache/oozie/action/hadoop/TestJavaActionExecutor.java b/core/src/test/java/org/apache/oozie/action/hadoop/TestJavaActionExecutor.java
index 123eba5..8adc606 100644
--- a/core/src/test/java/org/apache/oozie/action/hadoop/TestJavaActionExecutor.java
+++ b/core/src/test/java/org/apache/oozie/action/hadoop/TestJavaActionExecutor.java
@@ -21,7 +21,6 @@ package org.apache.oozie.action.hadoop;
import java.io.File;
import java.io.FileInputStream;
import java.io.FileOutputStream;
-import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.io.OutputStreamWriter;
@@ -36,7 +35,6 @@ import java.util.HashMap;
import java.util.Map;
import java.util.Properties;
-import org.apache.commons.lang3.mutable.MutableObject;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.filecache.DistributedCache;
import org.apache.hadoop.fs.FSDataOutputStream;
@@ -44,14 +42,9 @@ import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.JobConf;
-import org.apache.hadoop.mapred.RunningJob;
import org.apache.hadoop.security.token.Token;
import org.apache.hadoop.security.token.TokenIdentifier;
-import org.apache.hadoop.yarn.api.records.ApplicationId;
import org.apache.hadoop.yarn.api.records.YarnApplicationState;
-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;
@@ -60,7 +53,6 @@ import org.apache.oozie.client.OozieClient;
import org.apache.oozie.client.WorkflowAction;
import org.apache.oozie.client.WorkflowJob;
import org.apache.oozie.service.ConfigurationService;
-import org.apache.oozie.service.HadoopAccessorException;
import org.apache.oozie.service.HadoopAccessorService;
import org.apache.oozie.service.LiteWorkflowStoreService;
import org.apache.oozie.service.Services;
@@ -582,9 +574,7 @@ public class TestJavaActionExecutor extends ActionExecutorTestCase {
assertEquals(launcherId, runningJob2);
assertEquals(launcherId, context.getAction().getExternalId());
- waitUntilYarnAppCompletes(runningJob2);
- //FIXME?????
- waitUntilYarnAppState(launcherId, YarnApplicationState.FINISHED);
+ waitUntilYarnAppDoneAndAssertSuccess(launcherId);
ActionExecutor ae = new JavaActionExecutor();
ae.check(context, context.getAction());
assertEquals("SUCCEEDED", context.getAction().getExternalStatus());
http://git-wip-us.apache.org/repos/asf/oozie/blob/ca7e56fd/core/src/test/java/org/apache/oozie/action/hadoop/TestLauncherAMCallbackNotifier.java
----------------------------------------------------------------------
diff --git a/core/src/test/java/org/apache/oozie/action/hadoop/TestLauncherAMCallbackNotifier.java b/core/src/test/java/org/apache/oozie/action/hadoop/TestLauncherAMCallbackNotifier.java
index d0b4d5b..9ba04da 100644
--- a/core/src/test/java/org/apache/oozie/action/hadoop/TestLauncherAMCallbackNotifier.java
+++ b/core/src/test/java/org/apache/oozie/action/hadoop/TestLauncherAMCallbackNotifier.java
@@ -33,6 +33,22 @@ import java.util.Map;
// A lot of this adapted from org.apache.hadoop.mapreduce.v2.app.TestJobEndNotifier and org.apache.hadoop.mapred.TestJobEndNotifier
public class TestLauncherAMCallbackNotifier extends XTestCase {
+ private EmbeddedServletContainer container;
+
+ @Override
+ public void setUp() throws Exception {
+ super.setUp();
+ QueryServlet.lastQueryString = null;
+ }
+
+ @Override
+ public void tearDown() throws Exception {
+ if (container != null) {
+ container.stop();
+ }
+
+ super.tearDown();
+ }
public void testConfiguration() throws Exception {
Configuration conf = new Configuration(false);
@@ -91,8 +107,9 @@ public class TestLauncherAMCallbackNotifier extends XTestCase {
conf.set(LauncherAMCallbackNotifier.OOZIE_LAUNCHER_CALLBACK_RETRY_INTERVAL, "5000");
LauncherAMCallbackNotifier cnSpy = Mockito.spy(new LauncherAMCallbackNotifier(conf));
+
long start = System.currentTimeMillis();
- cnSpy.notifyURL(FinalApplicationStatus.SUCCEEDED);
+ cnSpy.notifyURL(FinalApplicationStatus.SUCCEEDED, false);
long end = System.currentTimeMillis();
Mockito.verify(cnSpy, Mockito.times(1)).notifyURLOnce();
Assert.assertTrue("Should have taken more than 5 seconds but it only took " + (end - start), end - start >= 5000);
@@ -103,68 +120,93 @@ public class TestLauncherAMCallbackNotifier extends XTestCase {
cnSpy = Mockito.spy(new LauncherAMCallbackNotifier(conf));
start = System.currentTimeMillis();
- cnSpy.notifyURL(FinalApplicationStatus.SUCCEEDED);
+ cnSpy.notifyURL(FinalApplicationStatus.SUCCEEDED, false);
end = System.currentTimeMillis();
Mockito.verify(cnSpy, Mockito.times(3)).notifyURLOnce();
Assert.assertTrue("Should have taken more than 9 seconds but it only took " + (end - start), end - start >= 9000);
}
public void testNotifyTimeout() throws Exception {
- EmbeddedServletContainer container = null;
- try {
- container = new EmbeddedServletContainer("blah");
- Map<String, String> params = new HashMap<String, String>();
- params.put(HangServlet.SLEEP_TIME_MS, "1000000");
- container.addServletEndpoint("/hang/*", HangServlet.class, params);
- container.start();
-
- Configuration conf = new Configuration(false);
- conf.set(LauncherAMCallbackNotifier.OOZIE_LAUNCHER_CALLBACK_RETRY_ATTEMPTS, "0");
- conf.set(LauncherAMCallbackNotifier.OOZIE_LAUNCHER_CALLBACK_MAX_ATTEMPTS, "1");
- conf.set(LauncherAMCallbackNotifier.OOZIE_LAUNCHER_CALLBACK_URL, container.getServletURL("/hang/*"));
- conf.set(LauncherAMCallbackNotifier.OOZIE_LAUNCHER_CALLBACK_RETRY_INTERVAL, "5000");
-
- LauncherAMCallbackNotifier cnSpy = Mockito.spy(new LauncherAMCallbackNotifier(conf));
- long start = System.currentTimeMillis();
- cnSpy.notifyURL(FinalApplicationStatus.SUCCEEDED);
- long end = System.currentTimeMillis();
- Mockito.verify(cnSpy, Mockito.times(1)).notifyURLOnce();
- Assert.assertTrue("Should have taken more than 5 seconds but it only took " + (end - start), end - start >= 5000);
- } finally {
- if (container != null) {
- container.stop();
- }
- }
+ Map<String, String> params = new HashMap<String, String>();
+ params.put(HangServlet.SLEEP_TIME_MS, "1000000");
+ Configuration conf = setupEmbeddedContainer(HangServlet.class, "/hang/*", "/hang/*", params);
+
+ LauncherAMCallbackNotifier cnSpy = Mockito.spy(new LauncherAMCallbackNotifier(conf));
+ long start = System.currentTimeMillis();
+ cnSpy.notifyURL(FinalApplicationStatus.SUCCEEDED, false);
+ long end = System.currentTimeMillis();
+ Mockito.verify(cnSpy, Mockito.times(1)).notifyURLOnce();
+ Assert.assertTrue("Should have taken more than 5 seconds but it only took " + (end - start), end - start >= 5000);
}
public void testNotify() throws Exception {
- EmbeddedServletContainer container = null;
- try {
- container = new EmbeddedServletContainer("blah");
- container.addServletEndpoint("/count/*", QueryServlet.class);
- container.start();
-
- Configuration conf = new Configuration(false);
- conf.set(LauncherAMCallbackNotifier.OOZIE_LAUNCHER_CALLBACK_RETRY_ATTEMPTS, "0");
- conf.set(LauncherAMCallbackNotifier.OOZIE_LAUNCHER_CALLBACK_MAX_ATTEMPTS, "1");
- conf.set(LauncherAMCallbackNotifier.OOZIE_LAUNCHER_CALLBACK_URL, container.getServletURL("/count/?status=$jobStatus"));
- conf.set(LauncherAMCallbackNotifier.OOZIE_LAUNCHER_CALLBACK_RETRY_INTERVAL, "5000");
-
- LauncherAMCallbackNotifier cn = new LauncherAMCallbackNotifier(conf);
- QueryServlet.lastQueryString = null;
- assertNull(QueryServlet.lastQueryString);
- cn.notifyURL(FinalApplicationStatus.SUCCEEDED);
- waitFor(5000, new Predicate() {
- @Override
- public boolean evaluate() throws Exception {
- return "status=SUCCEEDED".equals(QueryServlet.lastQueryString);
- }
- });
- assertEquals("status=SUCCEEDED", QueryServlet.lastQueryString);
- } finally {
- if (container != null) {
- container.stop();
+ Configuration conf = setupEmbeddedContainer(QueryServlet.class, "/count/*", "/count/?status=$jobStatus", null);
+
+ LauncherAMCallbackNotifier cn = new LauncherAMCallbackNotifier(conf);
+
+ assertNull(QueryServlet.lastQueryString);
+ cn.notifyURL(FinalApplicationStatus.SUCCEEDED, false);
+ waitForCallbackAndCheckResult(FinalApplicationStatus.SUCCEEDED.toString());
+ }
+
+ public void testNotifyBackgroundActionWhenSubmitSucceeds() throws Exception {
+ Configuration conf = setupEmbeddedContainer(QueryServlet.class, "/count/*", "/count/?status=$jobStatus", null);
+
+ LauncherAMCallbackNotifier cn = new LauncherAMCallbackNotifier(conf);
+
+ assertNull(QueryServlet.lastQueryString);
+ cn.notifyURL(FinalApplicationStatus.SUCCEEDED, true);
+ waitForCallbackAndCheckResult("RUNNING");
+ }
+
+ public void testNotifyBackgroundActionWhenSubmitFailsWithKilled() throws Exception {
+ Configuration conf = setupEmbeddedContainer(QueryServlet.class, "/count/*", "/count/?status=$jobStatus", null);
+
+ LauncherAMCallbackNotifier cn = new LauncherAMCallbackNotifier(conf);
+
+ assertNull(QueryServlet.lastQueryString);
+ cn.notifyURL(FinalApplicationStatus.KILLED, true);
+ waitForCallbackAndCheckResult(FinalApplicationStatus.KILLED.toString());
+ }
+
+ public void testNotifyBackgroundActionWhenSubmitFailsWithFailed() throws Exception {
+ Configuration conf = setupEmbeddedContainer(QueryServlet.class, "/count/*", "/count/?status=$jobStatus", null);
+
+ LauncherAMCallbackNotifier cn = new LauncherAMCallbackNotifier(conf);
+
+ assertNull(QueryServlet.lastQueryString);
+ cn.notifyURL(FinalApplicationStatus.FAILED, true);
+ waitForCallbackAndCheckResult(FinalApplicationStatus.FAILED.toString());
+ }
+
+ private Configuration setupEmbeddedContainer(Class<?> servletClass, String servletEndPoint, String servletUrl, Map<String, String> params) throws Exception {
+ container = new EmbeddedServletContainer("test");
+ if (servletEndPoint != null) {
+ if (params != null) {
+ container.addServletEndpoint(servletEndPoint, servletClass, params);
+ } else {
+ container.addServletEndpoint(servletEndPoint, servletClass);
}
}
+ container.start();
+
+ Configuration conf = new Configuration(false);
+ conf.set(LauncherAMCallbackNotifier.OOZIE_LAUNCHER_CALLBACK_RETRY_ATTEMPTS, "0");
+ conf.set(LauncherAMCallbackNotifier.OOZIE_LAUNCHER_CALLBACK_MAX_ATTEMPTS, "1");
+ conf.set(LauncherAMCallbackNotifier.OOZIE_LAUNCHER_CALLBACK_URL, container.getServletURL(servletUrl));
+ conf.set(LauncherAMCallbackNotifier.OOZIE_LAUNCHER_CALLBACK_RETRY_INTERVAL, "5000");
+
+ return conf;
+ }
+
+ private void waitForCallbackAndCheckResult(final String expectedResult) {
+ waitFor(5000, new Predicate() {
+ @Override
+ public boolean evaluate() throws Exception {
+ return ("status=" + expectedResult).equals(QueryServlet.lastQueryString);
+ }
+ });
+
+ assertEquals("status=" + expectedResult, QueryServlet.lastQueryString);
}
}
http://git-wip-us.apache.org/repos/asf/oozie/blob/ca7e56fd/core/src/test/java/org/apache/oozie/action/hadoop/TestMapReduceActionError.java
----------------------------------------------------------------------
diff --git a/core/src/test/java/org/apache/oozie/action/hadoop/TestMapReduceActionError.java b/core/src/test/java/org/apache/oozie/action/hadoop/TestMapReduceActionError.java
deleted file mode 100644
index 4cda615..0000000
--- a/core/src/test/java/org/apache/oozie/action/hadoop/TestMapReduceActionError.java
+++ /dev/null
@@ -1,173 +0,0 @@
-/**
- * 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 org.apache.hadoop.fs.Path;
-import org.apache.hadoop.fs.FileSystem;
-import org.apache.hadoop.mapred.JobConf;
-import org.apache.hadoop.mapred.JobClient;
-import org.apache.hadoop.mapred.RunningJob;
-import org.apache.hadoop.mapred.JobID;
-import org.apache.oozie.WorkflowActionBean;
-import org.apache.oozie.WorkflowJobBean;
-import org.apache.oozie.client.WorkflowAction;
-import org.apache.oozie.service.HadoopAccessorService;
-import org.apache.oozie.service.Services;
-import org.apache.oozie.service.WorkflowAppService;
-import org.apache.oozie.util.XConfiguration;
-import org.apache.oozie.util.XmlUtils;
-import org.apache.oozie.util.IOUtils;
-import org.jdom.Element;
-
-import java.io.File;
-import java.io.OutputStream;
-import java.io.InputStream;
-import java.io.FileInputStream;
-import java.io.StringReader;
-import java.io.Writer;
-import java.io.OutputStreamWriter;
-
-public class TestMapReduceActionError extends ActionExecutorTestCase {
-
- @Override
- protected void setSystemProps() throws Exception {
- super.setSystemProps();
- setSystemProperty("oozie.service.ActionService.executor.classes", MapReduceActionExecutor.class.getName());
- }
-
- private Context createContext(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.setType(ae.getType());
- action.setConf(actionXml);
-
- return new Context(wf, action);
- }
-
- private RunningJob submitAction(Context context) throws Exception {
- MapReduceActionExecutor ae = new MapReduceActionExecutor();
-
- WorkflowAction action = context.getAction();
-
- ae.prepareActionDir(getFileSystem(), context);
- ae.submitLauncher(getFileSystem(), context, action);
-
- String jobId = action.getExternalId();
- String jobTracker = action.getTrackerUri();
- String consoleUrl = action.getConsoleUrl();
- assertNotNull(jobId);
- assertNotNull(jobTracker);
- assertNotNull(consoleUrl);
-
- Element e = XmlUtils.parseXml(action.getConf());
- XConfiguration conf =
- new XConfiguration(new StringReader(XmlUtils.prettyPrint(e.getChild("configuration")).toString()));
- conf.set("mapred.job.tracker", e.getChildTextTrim("job-tracker"));
- conf.set("fs.default.name", e.getChildTextTrim("name-node"));
- conf.set("user.name", context.getProtoActionConf().get("user.name"));
- conf.set("group.name", getTestGroup());
-
- conf.set("mapreduce.framework.name", "yarn");
- JobConf jobConf = Services.get().get(HadoopAccessorService.class).createJobConf(jobTracker);
- XConfiguration.copy(conf, jobConf);
- String user = jobConf.get("user.name");
- String group = jobConf.get("group.name");
- JobClient jobClient = Services.get().get(HadoopAccessorService.class).createJobClient(user, jobConf);
- final RunningJob runningJob = jobClient.getJob(JobID.forName(jobId));
- assertNotNull(runningJob);
- return runningJob;
- }
-
- private void _testSubmit(String actionXml) throws Exception {
-
- Context context = createContext(actionXml);
- final RunningJob launcherJob = submitAction(context);
- String launcherId = context.getAction().getExternalId();
- waitFor(60 * 1000, new Predicate() {
- public boolean evaluate() throws Exception {
- return launcherJob.isComplete();
- }
- });
-
- MapReduceActionExecutor ae = new MapReduceActionExecutor();
- ae.check(context, context.getAction());
-
- JobConf conf = ae.createBaseHadoopConf(context, XmlUtils.parseXml(actionXml));
- String user = conf.get("user.name");
- String group = conf.get("group.name");
- JobClient jobClient = Services.get().get(HadoopAccessorService.class).createJobClient(user, conf);
- final RunningJob mrJob = jobClient.getJob(JobID.forName(context.getAction().getExternalId()));
-
- waitFor(60 * 1000, new Predicate() {
- public boolean evaluate() throws Exception {
- return mrJob.isComplete();
- }
- });
- ae.check(context, context.getAction());
-
- assertEquals("FAILED/KILLED", context.getAction().getExternalStatus());
-
- ae.end(context, context.getAction());
- assertEquals(WorkflowAction.Status.ERROR, context.getAction().getStatus());
- assertTrue(context.getAction().getErrorMessage().contains("already exists"));
- }
-
- public void testMapReduce() 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")));
- w.write("dummy\n");
- w.write("dummy\n");
- Writer ow = new OutputStreamWriter(fs.create(new Path(outputDir, "data.txt")));
- ow.write("dummy\n");
- ow.write("dummy\n");
- ow.close();
-
- 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>";
- _testSubmit(actionXml);
- }
-
-}
http://git-wip-us.apache.org/repos/asf/oozie/blob/ca7e56fd/core/src/test/java/org/apache/oozie/action/hadoop/TestOozieJobInfo.java
----------------------------------------------------------------------
diff --git a/core/src/test/java/org/apache/oozie/action/hadoop/TestOozieJobInfo.java b/core/src/test/java/org/apache/oozie/action/hadoop/TestOozieJobInfo.java
index 5bc7d00..9efacdd 100644
--- a/core/src/test/java/org/apache/oozie/action/hadoop/TestOozieJobInfo.java
+++ b/core/src/test/java/org/apache/oozie/action/hadoop/TestOozieJobInfo.java
@@ -33,19 +33,13 @@ import java.util.regex.Matcher;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
-import org.apache.hadoop.mapred.JobClient;
import org.apache.hadoop.mapred.JobConf;
-import org.apache.hadoop.mapred.JobID;
-import org.apache.hadoop.mapred.RunningJob;
import org.apache.oozie.BundleActionBean;
import org.apache.oozie.BundleJobBean;
import org.apache.oozie.CoordinatorJobBean;
import org.apache.oozie.ErrorCode;
import org.apache.oozie.WorkflowActionBean;
import org.apache.oozie.WorkflowJobBean;
-import org.apache.oozie.action.hadoop.MapReduceActionExecutor;
-import org.apache.oozie.action.hadoop.MapperReducerForTest;
-import org.apache.oozie.action.hadoop.OozieJobInfo;
import org.apache.oozie.client.Job;
import org.apache.oozie.client.OozieClient;
import org.apache.oozie.command.CommandException;
@@ -53,19 +47,16 @@ import org.apache.oozie.command.bundle.BundleStartXCommand;
import org.apache.oozie.command.bundle.BundleSubmitXCommand;
import org.apache.oozie.command.wf.ActionXCommand;
import org.apache.oozie.command.wf.ActionXCommand.ActionExecutorContext;
-import org.apache.oozie.command.wf.JobXCommand;
import org.apache.oozie.executor.jpa.BundleActionQueryExecutor;
+import org.apache.oozie.executor.jpa.BundleActionQueryExecutor.BundleActionQuery;
import org.apache.oozie.executor.jpa.BundleJobGetJPAExecutor;
import org.apache.oozie.executor.jpa.CoordJobGetJPAExecutor;
import org.apache.oozie.executor.jpa.WorkflowActionsGetForJobJPAExecutor;
import org.apache.oozie.executor.jpa.WorkflowJobGetJPAExecutor;
import org.apache.oozie.executor.jpa.WorkflowJobsGetFromCoordParentIdJPAExecutor;
-import org.apache.oozie.executor.jpa.BundleActionQueryExecutor.BundleActionQuery;
-import org.apache.oozie.service.HadoopAccessorService;
import org.apache.oozie.service.JPAService;
import org.apache.oozie.service.Services;
import org.apache.oozie.service.UUIDService;
-import org.apache.oozie.service.ConfigurationService;
import org.apache.oozie.service.UUIDService.ApplicationType;
import org.apache.oozie.test.XDataTestCase;
import org.apache.oozie.util.IOUtils;
@@ -165,12 +156,9 @@ public class TestOozieJobInfo extends XDataTestCase {
MapReduceActionExecutor actionExecutor = new MapReduceActionExecutor();
JobConf conf = actionExecutor.createBaseHadoopConf(context, XmlUtils.parseXml(actionList.get(1).getConf()));
String user = conf.get("user.name");
- JobClient jobClient = Services.get().get(HadoopAccessorService.class).createJobClient(user, conf);
- String launcherId = actionList.get(1).getExternalId();
- final RunningJob launcherJob = jobClient.getJob(JobID.forName(launcherId));
- FileSystem fs = context.getAppFileSystem();
- Configuration jobXmlConf = new XConfiguration(fs.open(new Path(launcherJob.getJobFile())));
+ FileSystem fs = getFileSystem();
+ Configuration jobXmlConf = new XConfiguration(fs.open(getPathToWorkflowResource(user, wfbean, services, context, LauncherAM.LAUNCHER_JOB_CONF_XML)));
String jobInfo = jobXmlConf.get(OozieJobInfo.JOB_INFO_KEY);
// BUNDLE_ID;BUNDLE_NAME;COORDINATOR_NAME;COORDINATOR_NOMINAL_TIME;
@@ -186,7 +174,6 @@ public class TestOozieJobInfo extends XDataTestCase {
assertTrue(jobInfo.contains(",testing=test,"));
assertTrue(jobInfo.contains(",coord.nominal.time="));
assertTrue(jobInfo.contains("launcher=true"));
-
}
protected void setCoordConf(Configuration jobConf) throws IOException {
http://git-wip-us.apache.org/repos/asf/oozie/blob/ca7e56fd/core/src/test/java/org/apache/oozie/action/hadoop/TestShellActionExecutor.java
----------------------------------------------------------------------
diff --git a/core/src/test/java/org/apache/oozie/action/hadoop/TestShellActionExecutor.java b/core/src/test/java/org/apache/oozie/action/hadoop/TestShellActionExecutor.java
index 9468fad..3354b3a 100644
--- a/core/src/test/java/org/apache/oozie/action/hadoop/TestShellActionExecutor.java
+++ b/core/src/test/java/org/apache/oozie/action/hadoop/TestShellActionExecutor.java
@@ -229,14 +229,8 @@ public class TestShellActionExecutor extends ActionExecutorTestCase {
Context context = createContext(actionXml);
// Submit the action
- final RunningJob launcherJob = submitAction(context);
- waitFor(180 * 1000, new Predicate() { // Wait for the external job to
- // finish
- public boolean evaluate() throws Exception {
- return launcherJob.isComplete();
- }
- });
-
+ final String launcherId = submitAction(context);
+ waitUntilYarnAppDoneAndAssertSuccess(launcherId);
ShellActionExecutor ae = new ShellActionExecutor();
WorkflowAction action = context.getAction();
ae.check(context, action);
@@ -258,19 +252,10 @@ public class TestShellActionExecutor extends ActionExecutorTestCase {
private WorkflowAction _testSubmit(String actionXml, boolean checkForSuccess, String capture_output) throws Exception {
Context context = createContext(actionXml);
- final RunningJob launcherJob = submitAction(context);// Submit the
+ final String launcherId = submitAction(context);// Submit the
// action
- String launcherId = context.getAction().getExternalId(); // Get LM id
- waitFor(180 * 1000, new Predicate() { // Wait for the external job to
- // finish
- public boolean evaluate() throws Exception {
- return launcherJob.isComplete();
- }
- });
- // Thread.sleep(2000);
- assertTrue(launcherJob.isSuccessful());
-
- sleep(2000);// Wait more to make sure no ID swap happens
+ waitUntilYarnAppDoneAndAssertSuccess(launcherId);
+
Configuration conf = new XConfiguration();
conf.set("user.name", getTestUser());
Map<String, String> actionData = LauncherMapperHelper.getActionData(getFileSystem(), context.getActionDir(),
@@ -334,14 +319,13 @@ public class TestShellActionExecutor extends ActionExecutorTestCase {
* @return The RunningJob of the Launcher Mapper
* @throws Exception
*/
- private RunningJob submitAction(Context context) throws Exception {
+ private String submitAction(Context context) throws Exception {
ShellActionExecutor ae = new ShellActionExecutor();
WorkflowAction action = context.getAction();
ae.prepareActionDir(getFileSystem(), context);
- ae.submitLauncher(getFileSystem(), context, action); // Submit the
- // Launcher Mapper
+ ae.submitLauncher(getFileSystem(), context, action); // Submit the action
String jobId = action.getExternalId();
String jobTracker = action.getTrackerUri();
@@ -351,20 +335,6 @@ public class TestShellActionExecutor extends ActionExecutorTestCase {
assertNotNull(jobTracker);
assertNotNull(consoleUrl);
- Element e = XmlUtils.parseXml(action.getConf());
- XConfiguration conf = new XConfiguration();
- conf.set("mapred.job.tracker", e.getChildTextTrim("job-tracker"));
- conf.set("fs.default.name", e.getChildTextTrim("name-node"));
- conf.set("user.name", context.getProtoActionConf().get("user.name"));
- conf.set("group.name", getTestGroup());
-
- JobConf jobConf = Services.get().get(HadoopAccessorService.class).createJobConf(jobTracker);
- XConfiguration.copy(conf, jobConf);
- String user = jobConf.get("user.name");
- String group = jobConf.get("group.name");
- JobClient jobClient = Services.get().get(HadoopAccessorService.class).createJobClient(user, jobConf);
- final RunningJob runningJob = jobClient.getJob(JobID.forName(jobId));
- assertNotNull(runningJob);
- return runningJob;
+ return jobId;
}
}
http://git-wip-us.apache.org/repos/asf/oozie/blob/ca7e56fd/core/src/test/java/org/apache/oozie/command/wf/TestActionCheckXCommand.java
----------------------------------------------------------------------
diff --git a/core/src/test/java/org/apache/oozie/command/wf/TestActionCheckXCommand.java b/core/src/test/java/org/apache/oozie/command/wf/TestActionCheckXCommand.java
index 0e1d0fd..d1f458b 100644
--- a/core/src/test/java/org/apache/oozie/command/wf/TestActionCheckXCommand.java
+++ b/core/src/test/java/org/apache/oozie/command/wf/TestActionCheckXCommand.java
@@ -29,8 +29,6 @@ import org.apache.hadoop.mapred.JobClient;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.JobID;
import org.apache.hadoop.mapred.RunningJob;
-import org.apache.hadoop.yarn.api.records.YarnApplicationState;
-import org.apache.oozie.ForTestingActionExecutor;
import org.apache.oozie.WorkflowActionBean;
import org.apache.oozie.WorkflowJobBean;
import org.apache.oozie.action.ActionExecutor;
@@ -46,7 +44,6 @@ import org.apache.oozie.executor.jpa.JPAExecutorException;
import org.apache.oozie.executor.jpa.WorkflowActionGetJPAExecutor;
import org.apache.oozie.executor.jpa.WorkflowActionInsertJPAExecutor;
import org.apache.oozie.executor.jpa.WorkflowJobGetJPAExecutor;
-import org.apache.oozie.service.ActionCheckerService;
import org.apache.oozie.service.ActionService;
import org.apache.oozie.service.ConfigurationService;
import org.apache.oozie.service.HadoopAccessorService;
@@ -266,9 +263,7 @@ public class TestActionCheckXCommand extends XDataTestCase {
String launcherId = action.getExternalId();
- waitUntilYarnAppCompletes(launcherId);
- YarnApplicationState appState = getYarnApplicationState(launcherId);
- assertEquals("YarnApplicationState", YarnApplicationState.FINISHED, appState);
+ waitUntilYarnAppDoneAndAssertSuccess(launcherId);
Map<String, String> actionData = LauncherMapperHelper.getActionData(getFileSystem(), context.getActionDir(),
conf);
http://git-wip-us.apache.org/repos/asf/oozie/blob/ca7e56fd/core/src/test/java/org/apache/oozie/command/wf/TestActionStartXCommand.java
----------------------------------------------------------------------
diff --git a/core/src/test/java/org/apache/oozie/command/wf/TestActionStartXCommand.java b/core/src/test/java/org/apache/oozie/command/wf/TestActionStartXCommand.java
index b7489e9..28e4755 100644
--- a/core/src/test/java/org/apache/oozie/command/wf/TestActionStartXCommand.java
+++ b/core/src/test/java/org/apache/oozie/command/wf/TestActionStartXCommand.java
@@ -30,8 +30,6 @@ import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.permission.FsPermission;
import org.apache.hadoop.mapred.JobClient;
import org.apache.hadoop.mapred.JobConf;
-import org.apache.hadoop.mapred.JobID;
-import org.apache.hadoop.mapred.RunningJob;
import org.apache.oozie.WorkflowActionBean;
import org.apache.oozie.WorkflowJobBean;
import org.apache.oozie.action.hadoop.LauncherMapperHelper;
@@ -163,19 +161,10 @@ public class TestActionStartXCommand extends XDataTestCase {
ActionExecutorContext context = new ActionXCommand.ActionExecutorContext(job, action, false, false);
MapReduceActionExecutor actionExecutor = new MapReduceActionExecutor();
JobConf conf = actionExecutor.createBaseHadoopConf(context, XmlUtils.parseXml(action.getConf()));
- String user = conf.get("user.name");
- JobClient jobClient = Services.get().get(HadoopAccessorService.class).createJobClient(user, conf);
String launcherId = action.getExternalId();
- final RunningJob launcherJob = jobClient.getJob(JobID.forName(launcherId));
-
- waitFor(120 * 1000, new Predicate() {
- public boolean evaluate() throws Exception {
- return launcherJob.isComplete();
- }
- });
- assertTrue(launcherJob.isSuccessful());
+ waitUntilYarnAppDoneAndAssertSuccess(launcherId);
Map<String, String> actionData = LauncherMapperHelper.getActionData(getFileSystem(), context.getActionDir(),
conf);
assertTrue(LauncherMapperHelper.hasIdSwap(actionData));
@@ -240,21 +229,10 @@ public class TestActionStartXCommand extends XDataTestCase {
MapReduceActionExecutor actionExecutor = new MapReduceActionExecutor();
JobConf conf = actionExecutor.createBaseHadoopConf(context, XmlUtils.parseXml(action.getConf()));
String user = conf.get("user.name");
- JobClient jobClient = Services.get().get(HadoopAccessorService.class).createJobClient(user, conf);
String launcherId = action.getExternalId();
- // retrieve launcher job
- final RunningJob launcherJob = jobClient.getJob(JobID.forName(launcherId));
-
- // time out after 120 seconds unless launcher job succeeds
- waitFor(240 * 1000, new Predicate() {
- public boolean evaluate() throws Exception {
- return launcherJob.isComplete();
- }
- });
- // check if launcher job succeeds
- assertTrue(launcherJob.isSuccessful());
+ waitUntilYarnAppDoneAndAssertSuccess(launcherId);
Map<String, String> actionData = LauncherMapperHelper.getActionData(getFileSystem(), context.getActionDir(),
conf);
assertTrue(LauncherMapperHelper.hasIdSwap(actionData));
http://git-wip-us.apache.org/repos/asf/oozie/blob/ca7e56fd/core/src/test/java/org/apache/oozie/test/XFsTestCase.java
----------------------------------------------------------------------
diff --git a/core/src/test/java/org/apache/oozie/test/XFsTestCase.java b/core/src/test/java/org/apache/oozie/test/XFsTestCase.java
index 1d399e4..20529e8 100644
--- a/core/src/test/java/org/apache/oozie/test/XFsTestCase.java
+++ b/core/src/test/java/org/apache/oozie/test/XFsTestCase.java
@@ -28,11 +28,16 @@ import org.apache.hadoop.mapred.JobClient;
import org.apache.hadoop.mapred.JobConf;
import org.apache.oozie.util.XConfiguration;
import org.apache.oozie.util.XLog;
+import org.apache.oozie.client.WorkflowAction;
+import org.apache.oozie.client.WorkflowJob;
+import org.apache.oozie.command.wf.ActionXCommand.ActionExecutorContext;
import org.apache.oozie.service.HadoopAccessorException;
import org.apache.oozie.service.HadoopAccessorService;
+import org.apache.oozie.service.Services;
import java.io.IOException;
import java.net.URI;
+import java.net.URISyntaxException;
/**
* Base JUnit <code>TestCase</code> subclass used by all Oozie testcases that need Hadoop FS access. <p/> As part of its
@@ -175,4 +180,26 @@ public abstract class XFsTestCase extends XTestCase {
return has.createJobClient(getTestUser(), conf);
}
+ /**
+ * Returns a Path object to a filesystem resource which belongs to a specific workflow on HDFS
+ * Example: /user/test/oozie-abcd/0000003-160913132555310-oozie-abcd-W/hadoop--map-reduce/launcher.xml
+ *
+ * @param userName current username
+ * @param action workflow Action object
+ * @param services Oozie Services class
+ * @param context Executor context
+ * @param fileName the filename
+ * @return the Path object which represents a file on HDFS
+ * @throws Exception
+ */
+ protected Path getPathToWorkflowResource(String userName, WorkflowJob job, Services services, ActionExecutorContext context, String fileName) throws Exception {
+ return new Path(
+ "/user" +
+ "/" + userName +
+ "/" + services.getSystemId() +
+ "/" + job.getId() +
+ "/" + context.getActionDir().getName(),
+ fileName
+ );
+ }
}
http://git-wip-us.apache.org/repos/asf/oozie/blob/ca7e56fd/core/src/test/java/org/apache/oozie/test/XTestCase.java
----------------------------------------------------------------------
diff --git a/core/src/test/java/org/apache/oozie/test/XTestCase.java b/core/src/test/java/org/apache/oozie/test/XTestCase.java
index 7d8c48f..1299fa3 100644
--- a/core/src/test/java/org/apache/oozie/test/XTestCase.java
+++ b/core/src/test/java/org/apache/oozie/test/XTestCase.java
@@ -1192,11 +1192,12 @@ public abstract class XTestCase extends TestCase {
return services;
}
- protected void waitUntilYarnAppState(String externalId, final YarnApplicationState... acceptedStates)
+ protected YarnApplicationState waitUntilYarnAppState(String externalId, final YarnApplicationState... acceptedStates)
throws HadoopAccessorException, IOException, YarnException {
final ApplicationId appId = ConverterUtils.toApplicationId(externalId);
final Set<YarnApplicationState> states = Sets.immutableEnumSet(Lists.newArrayList(acceptedStates));
final MutableBoolean endStateOK = new MutableBoolean(false);
+ final MutableObject<YarnApplicationState> finalState = new MutableObject<YarnApplicationState>();
JobConf jobConf = Services.get().get(HadoopAccessorService.class).createJobConf(getJobTrackerUri());
// This is needed here because we need a mutable final YarnClient
@@ -1207,6 +1208,7 @@ public abstract class XTestCase extends TestCase {
@Override
public boolean evaluate() throws Exception {
YarnApplicationState state = yarnClientMO.getValue().getApplicationReport(appId).getYarnApplicationState();
+ finalState.setValue(state);
if (states.contains(state)) {
endStateOK.setValue(true);
@@ -1223,10 +1225,12 @@ public abstract class XTestCase extends TestCase {
}
assertTrue(endStateOK.isTrue());
+ return finalState.getValue();
}
- protected void waitUntilYarnAppCompletes(String externalId) throws HadoopAccessorException, IOException, YarnException {
- waitUntilYarnAppState(externalId, YarnApplicationState.FAILED, YarnApplicationState.KILLED, YarnApplicationState.FINISHED);
+ protected void waitUntilYarnAppDoneAndAssertSuccess(String externalId) throws HadoopAccessorException, IOException, YarnException {
+ YarnApplicationState state = waitUntilYarnAppState(externalId, YarnApplicationState.FAILED, YarnApplicationState.KILLED, YarnApplicationState.FINISHED);
+ assertEquals("YARN App state", YarnApplicationState.FINISHED, state);
}
protected YarnApplicationState getYarnApplicationState(String externalId) throws HadoopAccessorException, IOException, YarnException {
http://git-wip-us.apache.org/repos/asf/oozie/blob/ca7e56fd/sharelib/hive/src/test/java/org/apache/oozie/action/hadoop/TestHiveActionExecutor.java
----------------------------------------------------------------------
diff --git a/sharelib/hive/src/test/java/org/apache/oozie/action/hadoop/TestHiveActionExecutor.java b/sharelib/hive/src/test/java/org/apache/oozie/action/hadoop/TestHiveActionExecutor.java
index b966d4b..ec53ba9 100644
--- a/sharelib/hive/src/test/java/org/apache/oozie/action/hadoop/TestHiveActionExecutor.java
+++ b/sharelib/hive/src/test/java/org/apache/oozie/action/hadoop/TestHiveActionExecutor.java
@@ -164,14 +164,8 @@ public class TestHiveActionExecutor extends ActionExecutorTestCase {
dataWriter.close();
Context context = createContext(getActionScriptXml());
Namespace ns = Namespace.getNamespace("uri:oozie:hive-action:0.2");
- final RunningJob launcherJob = submitAction(context, ns);
- String launcherId = context.getAction().getExternalId();
- waitFor(200 * 1000, new Predicate() {
- public boolean evaluate() throws Exception {
- return launcherJob.isComplete();
- }
- });
- assertTrue(launcherJob.isSuccessful());
+ final String launcherId = submitAction(context, ns);
+ waitUntilYarnAppDoneAndAssertSuccess(launcherId);
Configuration conf = new XConfiguration();
conf.set("user.name", getTestUser());
Map<String, String> actionData = LauncherMapperHelper.getActionData(getFileSystem(), context.getActionDir(),
@@ -198,14 +192,8 @@ public class TestHiveActionExecutor extends ActionExecutorTestCase {
{
Context context = createContext(getActionQueryXml(hiveScript));
Namespace ns = Namespace.getNamespace("uri:oozie:hive-action:0.6");
- final RunningJob launcherJob = submitAction(context, ns);
- String launcherId = context.getAction().getExternalId();
- waitFor(200 * 1000, new Predicate() {
- public boolean evaluate() throws Exception {
- return launcherJob.isComplete();
- }
- });
- assertTrue(launcherJob.isSuccessful());
+ final String launcherId = submitAction(context, ns);
+ waitUntilYarnAppDoneAndAssertSuccess(launcherId);
Configuration conf = new XConfiguration();
conf.set("user.name", getTestUser());
Map<String, String> actionData = LauncherMapperHelper.getActionData(getFileSystem(), context.getActionDir(),
@@ -231,7 +219,7 @@ public class TestHiveActionExecutor extends ActionExecutorTestCase {
}
}
- private RunningJob submitAction(Context context, Namespace ns) throws Exception {
+ private String submitAction(Context context, Namespace ns) throws Exception {
HiveActionExecutor ae = new HiveActionExecutor();
WorkflowAction action = context.getAction();
@@ -245,22 +233,9 @@ public class TestHiveActionExecutor extends ActionExecutorTestCase {
assertNotNull(jobId);
assertNotNull(jobTracker);
assertNotNull(consoleUrl);
- Element e = XmlUtils.parseXml(action.getConf());
- XConfiguration conf =
- new XConfiguration(new StringReader(XmlUtils.prettyPrint(e.getChild("configuration", ns)).toString()));
- conf.set("mapred.job.tracker", e.getChildTextTrim("job-tracker", ns));
- conf.set("fs.default.name", e.getChildTextTrim("name-node", ns));
- conf.set("user.name", context.getProtoActionConf().get("user.name"));
- conf.set("group.name", getTestGroup());
- JobConf jobConf = Services.get().get(HadoopAccessorService.class).createJobConf(jobTracker);
- XConfiguration.copy(conf, jobConf);
- String user = jobConf.get("user.name");
- String group = jobConf.get("group.name");
- JobClient jobClient = Services.get().get(HadoopAccessorService.class).createJobClient(user, jobConf);
- final RunningJob runningJob = jobClient.getJob(JobID.forName(jobId));
- assertNotNull(runningJob);
- return runningJob;
+
+ return jobId;
}
private String copyJar(String targetFile, Class<?> anyContainedClass)
http://git-wip-us.apache.org/repos/asf/oozie/blob/ca7e56fd/sharelib/hive2/src/test/java/org/apache/oozie/action/hadoop/TestHive2ActionExecutor.java
----------------------------------------------------------------------
diff --git a/sharelib/hive2/src/test/java/org/apache/oozie/action/hadoop/TestHive2ActionExecutor.java b/sharelib/hive2/src/test/java/org/apache/oozie/action/hadoop/TestHive2ActionExecutor.java
index 5963e42..5e71f12 100644
--- a/sharelib/hive2/src/test/java/org/apache/oozie/action/hadoop/TestHive2ActionExecutor.java
+++ b/sharelib/hive2/src/test/java/org/apache/oozie/action/hadoop/TestHive2ActionExecutor.java
@@ -205,16 +205,9 @@ public class TestHive2ActionExecutor extends ActionExecutorTestCase {
dataWriter.write(SAMPLE_DATA_TEXT);
dataWriter.close();
Context context = createContext(getQueryActionXml(query));
- final RunningJob launcherJob = submitAction(context,
+ final String launcherId = submitAction(context,
Namespace.getNamespace("uri:oozie:hive2-action:0.2"));
- String launcherId = context.getAction().getExternalId();
- waitFor(200 * 1000, new Predicate() {
- @Override
- public boolean evaluate() throws Exception {
- return launcherJob.isComplete();
- }
- });
- assertTrue(launcherJob.isSuccessful());
+ waitUntilYarnAppDoneAndAssertSuccess(launcherId);
Configuration conf = new XConfiguration();
conf.set("user.name", getTestUser());
Map<String, String> actionData = LauncherMapperHelper.getActionData(getFileSystem(), context.getActionDir(),
@@ -244,16 +237,9 @@ public class TestHive2ActionExecutor extends ActionExecutorTestCase {
dataWriter.write(SAMPLE_DATA_TEXT);
dataWriter.close();
Context context = createContext(getScriptActionXml());
- final RunningJob launcherJob = submitAction(context,
+ final String launcherId = submitAction(context,
Namespace.getNamespace("uri:oozie:hive2-action:0.1"));
- String launcherId = context.getAction().getExternalId();
- waitFor(200 * 1000, new Predicate() {
- @Override
- public boolean evaluate() throws Exception {
- return launcherJob.isComplete();
- }
- });
- assertTrue(launcherJob.isSuccessful());
+ waitUntilYarnAppDoneAndAssertSuccess(launcherId);
Configuration conf = new XConfiguration();
conf.set("user.name", getTestUser());
Map<String, String> actionData = LauncherMapperHelper.getActionData(getFileSystem(), context.getActionDir(),
@@ -275,7 +261,7 @@ public class TestHive2ActionExecutor extends ActionExecutorTestCase {
}
}
- private RunningJob submitAction(Context context, Namespace ns) throws Exception {
+ private String submitAction(Context context, Namespace ns) throws Exception {
Hive2ActionExecutor ae = new Hive2ActionExecutor();
WorkflowAction action = context.getAction();
@@ -289,21 +275,7 @@ public class TestHive2ActionExecutor extends ActionExecutorTestCase {
assertNotNull(jobId);
assertNotNull(jobTracker);
assertNotNull(consoleUrl);
- Element e = XmlUtils.parseXml(action.getConf());
- XConfiguration conf =
- new XConfiguration(new StringReader(XmlUtils.prettyPrint(e.getChild("configuration", ns)).toString()));
- conf.set("mapred.job.tracker", e.getChildTextTrim("job-tracker", ns));
- conf.set("fs.default.name", e.getChildTextTrim("name-node", ns));
- conf.set("user.name", context.getProtoActionConf().get("user.name"));
- conf.set("group.name", getTestGroup());
-
- JobConf jobConf = Services.get().get(HadoopAccessorService.class).createJobConf(jobTracker);
- XConfiguration.copy(conf, jobConf);
- String user = jobConf.get("user.name");
- JobClient jobClient = Services.get().get(HadoopAccessorService.class).createJobClient(user, jobConf);
- final RunningJob runningJob = jobClient.getJob(JobID.forName(jobId));
- assertNotNull(runningJob);
- return runningJob;
+ return jobId;
}
private Context createContext(String actionXml) throws Exception {
http://git-wip-us.apache.org/repos/asf/oozie/blob/ca7e56fd/sharelib/oozie/src/main/java/org/apache/oozie/action/hadoop/LauncherAM.java
----------------------------------------------------------------------
diff --git a/sharelib/oozie/src/main/java/org/apache/oozie/action/hadoop/LauncherAM.java b/sharelib/oozie/src/main/java/org/apache/oozie/action/hadoop/LauncherAM.java
index a1998e2..43ce520 100644
--- a/sharelib/oozie/src/main/java/org/apache/oozie/action/hadoop/LauncherAM.java
+++ b/sharelib/oozie/src/main/java/org/apache/oozie/action/hadoop/LauncherAM.java
@@ -17,22 +17,6 @@
*/
package org.apache.oozie.action.hadoop;
-import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.fs.FileSystem;
-import org.apache.hadoop.fs.Path;
-import org.apache.hadoop.io.SequenceFile;
-import org.apache.hadoop.io.Text;
-import org.apache.hadoop.yarn.api.records.Container;
-import org.apache.hadoop.yarn.api.records.ContainerStatus;
-import org.apache.hadoop.yarn.api.records.FinalApplicationStatus;
-import org.apache.hadoop.yarn.api.records.NodeReport;
-import org.apache.hadoop.yarn.client.api.AMRMClient;
-import org.apache.hadoop.yarn.client.api.async.AMRMClientAsync;
-import org.apache.hadoop.yarn.exceptions.YarnException;
-import org.codehaus.jackson.map.Module.SetupContext;
-import org.xml.sax.SAXException;
-
-import javax.xml.parsers.ParserConfigurationException;
import java.io.BufferedReader;
import java.io.File;
import java.io.FileReader;
@@ -43,13 +27,29 @@ import java.io.StringWriter;
import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.security.Permission;
-import java.util.Date;
+import java.security.PrivilegedAction;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.Set;
import java.util.StringTokenizer;
+import java.util.concurrent.atomic.AtomicReference;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.io.SequenceFile;
+import org.apache.hadoop.io.Text;
+import org.apache.hadoop.security.UserGroupInformation;
+import org.apache.hadoop.yarn.api.records.Container;
+import org.apache.hadoop.yarn.api.records.ContainerStatus;
+import org.apache.hadoop.yarn.api.records.FinalApplicationStatus;
+import org.apache.hadoop.yarn.api.records.NodeReport;
+import org.apache.hadoop.yarn.client.api.AMRMClient;
+import org.apache.hadoop.yarn.client.api.async.AMRMClientAsync;
+import org.apache.hadoop.yarn.exceptions.YarnException;
+
+import com.google.common.base.Preconditions;
public class LauncherAM {
@@ -125,8 +125,14 @@ public class LauncherAM {
// TODO: OYA: rethink all print messages and formatting
public static void main(String[] AMargs) throws Exception {
- ErrorHolder eHolder = new ErrorHolder();
+ final ErrorHolder eHolder = new ErrorHolder();
FinalApplicationStatus finalStatus = FinalApplicationStatus.FAILED;
+ String submitterUser = System.getProperty("submitter.user", "").trim();
+ Preconditions.checkArgument(!submitterUser.isEmpty(), "Submitter user is undefined");
+ System.out.println("Submitter user is: " + submitterUser);
+ UserGroupInformation ugi = UserGroupInformation.createRemoteUser(submitterUser);
+ boolean backgroundAction = false;
+
try {
try {
launcherJobConf = readLauncherConf();
@@ -143,7 +149,7 @@ public class LauncherAM {
try {
System.out.println("\nStarting the execution of prepare actions");
- executePrepare();
+ executePrepare(ugi);
System.out.println("Completed the execution of prepare actions successfully");
} catch (Exception ex) {
eHolder.setErrorMessage("Prepare execution in the Launcher AM has failed");
@@ -151,7 +157,7 @@ public class LauncherAM {
throw ex;
}
- String[] mainArgs = getMainArguments(launcherJobConf);
+ final String[] mainArgs = getMainArguments(launcherJobConf);
// TODO: OYA: should we allow turning this off?
// TODO: OYA: what should default be?
@@ -161,7 +167,8 @@ public class LauncherAM {
setupMainConfiguration();
- finalStatus = runActionMain(mainArgs, eHolder);
+ finalStatus = runActionMain(mainArgs, eHolder, ugi);
+
if (finalStatus == FinalApplicationStatus.SUCCEEDED) {
handleActionData();
if (actionData.get(ACTION_DATA_OUTPUT_PROPS) != null) {
@@ -180,6 +187,7 @@ public class LauncherAM {
System.out.println(actionData.get(ACTION_DATA_NEW_ID));
System.out.println("=======================");
System.out.println();
+ backgroundAction = true;
}
}
} catch (Exception e) {
@@ -193,13 +201,13 @@ public class LauncherAM {
if (finalStatus != FinalApplicationStatus.SUCCEEDED) {
failLauncher(eHolder);
}
- uploadActionDataToHDFS();
+ uploadActionDataToHDFS(ugi);
} finally {
try {
unregisterWithRM(finalStatus, eHolder.getErrorMessage());
} finally {
LauncherAMCallbackNotifier cn = new LauncherAMCallbackNotifier(launcherJobConf);
- cn.notifyURL(finalStatus);
+ cn.notifyURL(finalStatus, backgroundAction);
}
}
}
@@ -240,16 +248,31 @@ public class LauncherAM {
}
// Method to execute the prepare actions
- private static void executePrepare() throws IOException, LauncherException, ParserConfigurationException, SAXException {
- String prepareXML = launcherJobConf.get(ACTION_PREPARE_XML);
- if (prepareXML != null) {
- if (prepareXML.length() != 0) {
- Configuration actionConf = new Configuration(launcherJobConf);
- actionConf.addResource(ACTION_CONF_XML);
- PrepareActionsDriver.doOperations(prepareXML, actionConf);
- } else {
- System.out.println("There are no prepare actions to execute.");
+ private static void executePrepare(UserGroupInformation ugi) throws Exception {
+ Exception e = ugi.doAs(new PrivilegedAction<Exception>() {
+ @Override
+ public Exception run() {
+ try {
+ String prepareXML = launcherJobConf.get(ACTION_PREPARE_XML);
+ if (prepareXML != null) {
+ if (prepareXML.length() != 0) {
+ Configuration actionConf = new Configuration(launcherJobConf);
+ actionConf.addResource(ACTION_CONF_XML);
+ PrepareActionsDriver.doOperations(prepareXML, actionConf);
+ } else {
+ System.out.println("There are no prepare actions to execute.");
+ }
+ }
+ } catch (Exception e) {
+ e.printStackTrace();
+ return e;
+ }
+ return null;
}
+ });
+
+ if (e != null) {
+ throw e;
}
}
@@ -282,65 +305,74 @@ public class LauncherAM {
// }
}
- private static FinalApplicationStatus runActionMain(String[] mainArgs, ErrorHolder eHolder) {
- FinalApplicationStatus finalStatus = FinalApplicationStatus.FAILED;
- LauncherSecurityManager secMan = new LauncherSecurityManager();
- try {
- Class<?> klass = launcherJobConf.getClass(CONF_OOZIE_ACTION_MAIN_CLASS, Object.class);
- System.out.println("Launcher class: " + klass.toString());
- System.out.flush();
- Method mainMethod = klass.getMethod("main", String[].class);
- // Enable LauncherSecurityManager to catch System.exit calls
- secMan.set();
- mainMethod.invoke(null, (Object) mainArgs);
-
- System.out.println();
- System.out.println("<<< Invocation of Main class completed <<<");
- System.out.println();
- finalStatus = FinalApplicationStatus.SUCCEEDED;
- } catch (InvocationTargetException ex) {
- ex.printStackTrace(System.out);
- // Get what actually caused the exception
- Throwable cause = ex.getCause();
- // If we got a JavaMainException from JavaMain, then we need to unwrap it
- if (JavaMainException.class.isInstance(cause)) {
- cause = cause.getCause();
- }
- if (LauncherMainException.class.isInstance(cause)) {
- String mainClass = launcherJobConf.get(CONF_OOZIE_ACTION_MAIN_CLASS);
- eHolder.setErrorMessage("Main Class [" + mainClass + "], exit code [" +
- ((LauncherMainException) ex.getCause()).getErrorCode() + "]");
- } else if (SecurityException.class.isInstance(cause)) {
- if (secMan.getExitInvoked()) {
- System.out.println("Intercepting System.exit(" + secMan.getExitCode()
- + ")");
- System.err.println("Intercepting System.exit(" + secMan.getExitCode()
- + ")");
- // if 0 main() method finished successfully
- // ignoring
- eHolder.setErrorCode(secMan.getExitCode());
- if (eHolder.getErrorCode() != 0) {
+ private static FinalApplicationStatus runActionMain(final String[] mainArgs, final ErrorHolder eHolder, UserGroupInformation ugi) {
+ final AtomicReference<FinalApplicationStatus> finalStatus = new AtomicReference<FinalApplicationStatus>(FinalApplicationStatus.FAILED);
+
+ ugi.doAs(new PrivilegedAction<Void>() {
+ @Override
+ public Void run() {
+ LauncherSecurityManager secMan = new LauncherSecurityManager();
+ try {
+ Class<?> klass = launcherJobConf.getClass(CONF_OOZIE_ACTION_MAIN_CLASS, Object.class);
+ System.out.println("Launcher class: " + klass.toString());
+ System.out.flush();
+ Method mainMethod = klass.getMethod("main", String[].class);
+ // Enable LauncherSecurityManager to catch System.exit calls
+ secMan.set();
+ mainMethod.invoke(null, (Object) mainArgs);
+
+ System.out.println();
+ System.out.println("<<< Invocation of Main class completed <<<");
+ System.out.println();
+ finalStatus.set(FinalApplicationStatus.SUCCEEDED);
+ } catch (InvocationTargetException ex) {
+ ex.printStackTrace(System.out);
+ // Get what actually caused the exception
+ Throwable cause = ex.getCause();
+ // If we got a JavaMainException from JavaMain, then we need to unwrap it
+ if (JavaMainException.class.isInstance(cause)) {
+ cause = cause.getCause();
+ }
+ if (LauncherMainException.class.isInstance(cause)) {
String mainClass = launcherJobConf.get(CONF_OOZIE_ACTION_MAIN_CLASS);
- eHolder.setErrorMessage("Main Class [" + mainClass + "], exit code [" + eHolder.getErrorCode() + "]");
+ eHolder.setErrorMessage("Main Class [" + mainClass + "], exit code [" +
+ ((LauncherMainException) ex.getCause()).getErrorCode() + "]");
+ } else if (SecurityException.class.isInstance(cause)) {
+ if (secMan.getExitInvoked()) {
+ System.out.println("Intercepting System.exit(" + secMan.getExitCode()
+ + ")");
+ System.err.println("Intercepting System.exit(" + secMan.getExitCode()
+ + ")");
+ // if 0 main() method finished successfully
+ // ignoring
+ eHolder.setErrorCode(secMan.getExitCode());
+ if (eHolder.getErrorCode() != 0) {
+ String mainClass = launcherJobConf.get(CONF_OOZIE_ACTION_MAIN_CLASS);
+ eHolder.setErrorMessage("Main Class [" + mainClass + "], exit code [" + eHolder.getErrorCode() + "]");
+ } else {
+ finalStatus.set(FinalApplicationStatus.SUCCEEDED);
+ }
+ }
} else {
- finalStatus = FinalApplicationStatus.SUCCEEDED;
+ eHolder.setErrorMessage(cause.getMessage());
+ eHolder.setErrorCause(cause);
}
+ } catch (Throwable t) {
+ t.printStackTrace();
+ eHolder.setErrorMessage(t.getMessage());
+ eHolder.setErrorCause(t);
+ } finally {
+ System.out.flush();
+ System.err.flush();
+ // Disable LauncherSecurityManager
+ secMan.unset();
}
- } else {
- eHolder.setErrorMessage(cause.getMessage());
- eHolder.setErrorCause(cause);
+
+ return null;
}
- } catch (Throwable t) {
- t.printStackTrace(System.out);
- eHolder.setErrorMessage(t.getMessage());
- eHolder.setErrorCause(t);
- } finally {
- System.out.flush();
- System.err.flush();
- // Disable LauncherSecurityManager
- secMan.unset();
- }
- return finalStatus;
+ });
+
+ return finalStatus.get();
}
private static void handleActionData() throws IOException {
@@ -410,40 +442,52 @@ public class LauncherAM {
return sb.toString();
}
- private static void uploadActionDataToHDFS() throws IOException {
- Path finalPath = new Path(actionDir, ACTION_DATA_SEQUENCE_FILE);
- // unused ??
- FileSystem fs = FileSystem.get(finalPath.toUri(), launcherJobConf);
- // upload into sequence file
- System.out.println("Oozie Launcher, uploading action data to HDFS sequence file: "
- + new Path(actionDir, ACTION_DATA_SEQUENCE_FILE).toUri());
+ private static void uploadActionDataToHDFS(UserGroupInformation ugi) throws IOException {
+ IOException ioe = ugi.doAs(new PrivilegedAction<IOException>() {
+ @Override
+ public IOException run() {
+ Path finalPath = new Path(actionDir, ACTION_DATA_SEQUENCE_FILE);
+ // upload into sequence file
+ System.out.println("Oozie Launcher, uploading action data to HDFS sequence file: "
+ + new Path(actionDir, ACTION_DATA_SEQUENCE_FILE).toUri());
- SequenceFile.Writer wr = null;
- try {
- wr = SequenceFile.createWriter(launcherJobConf,
- SequenceFile.Writer.file(finalPath),
- SequenceFile.Writer.keyClass(Text.class),
- SequenceFile.Writer.valueClass(Text.class));
- if (wr != null) {
- Set<String> keys = actionData.keySet();
- for (String propsKey : keys) {
- wr.append(new Text(propsKey), new Text(actionData.get(propsKey)));
+ SequenceFile.Writer wr = null;
+ try {
+ wr = SequenceFile.createWriter(launcherJobConf,
+ SequenceFile.Writer.file(finalPath),
+ SequenceFile.Writer.keyClass(Text.class),
+ SequenceFile.Writer.valueClass(Text.class));
+ if (wr != null) {
+ Set<String> keys = actionData.keySet();
+ for (String propsKey : keys) {
+ wr.append(new Text(propsKey), new Text(actionData.get(propsKey)));
+ }
+ } else {
+ throw new IOException("SequenceFile.Writer is null for " + finalPath);
+ }
+ } catch (IOException e) {
+ e.printStackTrace();
+ return e;
+ } finally {
+ if (wr != null) {
+ try {
+ wr.close();
+ } catch (IOException e) {
+ e.printStackTrace();
+ return e;
+ }
+ }
}
+
+ return null;
}
- else {
- throw new IOException("SequenceFile.Writer is null for " + finalPath);
- }
- }
- catch(IOException e) {
- e.printStackTrace();
- throw e;
- }
- finally {
- if (wr != null) {
- wr.close();
- }
+ });
+
+ if (ioe != null) {
+ throw ioe;
}
}
+
private static void failLauncher(int errorCode, String message, Throwable ex) {
ErrorHolder eHolder = new ErrorHolder();
eHolder.setErrorCode(errorCode);