\n"
+ + " | \n Multiple_line_value\n | \n \n"
+ + " This is first line.\n \n \n"
+ + " This is second line.\n \n";
+ assertTrue(output.contains(expectedSinglelineData) && output.contains(expectedMultilineData));
+ }
+}
\ No newline at end of file
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/ContainerExecutor.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/ContainerExecutor.java
index 9cffde1a65d..31fe49c86f3 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/ContainerExecutor.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/ContainerExecutor.java
@@ -37,6 +37,7 @@ import org.apache.hadoop.util.Shell.ShellCommandExecutor;
import org.apache.hadoop.yarn.api.records.ContainerId;
import org.apache.hadoop.yarn.server.nodemanager.containermanager.container.Container;
import org.apache.hadoop.yarn.server.nodemanager.util.ProcessIdFileReader;
+import org.apache.hadoop.util.Shell;
public abstract class ContainerExecutor implements Configurable {
@@ -182,6 +183,33 @@ public abstract class ContainerExecutor implements Configurable {
readLock.unlock();
}
}
+
+ /** Return a command to execute the given command in OS shell.
+ * On Windows, the passed in groupId can be used to launch
+ * and associate the given groupId in a process group. On
+ * non-Windows, groupId is ignored. */
+ protected static String[] getRunCommand(String command,
+ String groupId) {
+ if (Shell.WINDOWS) {
+ return new String[] { Shell.WINUTILS, "task", "create", groupId,
+ "cmd /c " + command };
+ } else {
+ return new String[] { "bash", "-c", command };
+ }
+ }
+
+ /** Return a command for determining if process with specified pid is alive. */
+ protected static String[] getCheckProcessIsAliveCommand(String pid) {
+ return Shell.WINDOWS ?
+ new String[] { Shell.WINUTILS, "task", "isAlive", pid } :
+ new String[] { "kill", "-0", pid };
+ }
+
+ /** Return a command to send a signal to a given pid */
+ protected static String[] getSignalKillCommand(int code, String pid) {
+ return Shell.WINDOWS ? new String[] { Shell.WINUTILS, "task", "kill", pid } :
+ new String[] { "kill", "-" + code, pid };
+ }
/**
* Is the container still active?
@@ -253,6 +281,9 @@ public abstract class ContainerExecutor implements Configurable {
public static final boolean isSetsidAvailable = isSetsidSupported();
private static boolean isSetsidSupported() {
+ if (Shell.WINDOWS) {
+ return true;
+ }
ShellCommandExecutor shexec = null;
boolean setsidSupported = true;
try {
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/DefaultContainerExecutor.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/DefaultContainerExecutor.java
index 27730f421b6..cc3fc76697e 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/DefaultContainerExecutor.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/DefaultContainerExecutor.java
@@ -37,6 +37,8 @@ import org.apache.hadoop.fs.FileContext;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.UnsupportedFileSystemException;
import org.apache.hadoop.fs.permission.FsPermission;
+import org.apache.hadoop.io.IOUtils;
+import org.apache.hadoop.util.Shell;
import org.apache.hadoop.util.Shell.ExitCodeException;
import org.apache.hadoop.util.Shell.ShellCommandExecutor;
import org.apache.hadoop.yarn.api.records.ContainerId;
@@ -53,10 +55,9 @@ public class DefaultContainerExecutor extends ContainerExecutor {
private static final Log LOG = LogFactory
.getLog(DefaultContainerExecutor.class);
- private final FileContext lfs;
+ private static final int WIN_MAX_PATH = 260;
- private static final String WRAPPER_LAUNCH_SCRIPT =
- "default_container_executor.sh";
+ private final FileContext lfs;
public DefaultContainerExecutor() {
try {
@@ -145,15 +146,24 @@ public class DefaultContainerExecutor extends ContainerExecutor {
lfs.util().copy(nmPrivateTokensPath, tokenDst);
// Create new local launch wrapper script
- Path wrapperScriptDst = new Path(containerWorkDir, WRAPPER_LAUNCH_SCRIPT);
- DataOutputStream wrapperScriptOutStream =
- lfs.create(wrapperScriptDst,
- EnumSet.of(CREATE, OVERWRITE));
+ LocalWrapperScriptBuilder sb = Shell.WINDOWS ?
+ new WindowsLocalWrapperScriptBuilder(containerIdStr, containerWorkDir) :
+ new UnixLocalWrapperScriptBuilder(containerWorkDir);
+
+ // Fail fast if attempting to launch the wrapper script would fail due to
+ // Windows path length limitation.
+ if (Shell.WINDOWS &&
+ sb.getWrapperScriptPath().toString().length() > WIN_MAX_PATH) {
+ throw new IOException(String.format(
+ "Cannot launch container using script at path %s, because it exceeds " +
+ "the maximum supported path length of %d characters. Consider " +
+ "configuring shorter directories in %s.", sb.getWrapperScriptPath(),
+ WIN_MAX_PATH, YarnConfiguration.NM_LOCAL_DIRS));
+ }
Path pidFile = getPidFilePath(containerId);
if (pidFile != null) {
- writeLocalWrapperScript(wrapperScriptOutStream, launchDst.toUri()
- .getPath().toString(), pidFile.toString());
+ sb.writeLocalWrapperScript(launchDst, pidFile);
} else {
LOG.info("Container " + containerIdStr
+ " was marked as inactive. Returning terminated error");
@@ -166,12 +176,13 @@ public class DefaultContainerExecutor extends ContainerExecutor {
try {
lfs.setPermission(launchDst,
ContainerExecutor.TASK_LAUNCH_SCRIPT_PERMISSION);
- lfs.setPermission(wrapperScriptDst,
+ lfs.setPermission(sb.getWrapperScriptPath(),
ContainerExecutor.TASK_LAUNCH_SCRIPT_PERMISSION);
// Setup command to run
- String[] command = {"bash",
- wrapperScriptDst.toUri().getPath().toString()};
+ String[] command = getRunCommand(sb.getWrapperScriptPath().toString(),
+ containerIdStr);
+
LOG.info("launchContainer: " + Arrays.toString(command));
shExec = new ShellCommandExecutor(
command,
@@ -202,28 +213,85 @@ public class DefaultContainerExecutor extends ContainerExecutor {
return 0;
}
- private void writeLocalWrapperScript(DataOutputStream out,
- String launchScriptDst, String pidFilePath) throws IOException {
- // We need to do a move as writing to a file is not atomic
- // Process reading a file being written to may get garbled data
- // hence write pid to tmp file first followed by a mv
- StringBuilder sb = new StringBuilder("#!/bin/bash\n\n");
- sb.append("echo $$ > " + pidFilePath + ".tmp\n");
- sb.append("/bin/mv -f " + pidFilePath + ".tmp " + pidFilePath + "\n");
- sb.append(ContainerExecutor.isSetsidAvailable? "exec setsid" : "exec");
- sb.append(" /bin/bash ");
- sb.append("\"");
- sb.append(launchScriptDst);
- sb.append("\"\n");
- PrintStream pout = null;
- try {
- pout = new PrintStream(out);
- pout.append(sb);
- } finally {
- if (out != null) {
- out.close();
+ private abstract class LocalWrapperScriptBuilder {
+
+ private final Path wrapperScriptPath;
+
+ public Path getWrapperScriptPath() {
+ return wrapperScriptPath;
+ }
+
+ public void writeLocalWrapperScript(Path launchDst, Path pidFile) throws IOException {
+ DataOutputStream out = null;
+ PrintStream pout = null;
+
+ try {
+ out = lfs.create(wrapperScriptPath, EnumSet.of(CREATE, OVERWRITE));
+ pout = new PrintStream(out);
+ writeLocalWrapperScript(launchDst, pidFile, pout);
+ } finally {
+ IOUtils.cleanup(LOG, pout, out);
}
}
+
+ protected abstract void writeLocalWrapperScript(Path launchDst, Path pidFile,
+ PrintStream pout);
+
+ protected LocalWrapperScriptBuilder(Path wrapperScriptPath) {
+ this.wrapperScriptPath = wrapperScriptPath;
+ }
+ }
+
+ private final class UnixLocalWrapperScriptBuilder
+ extends LocalWrapperScriptBuilder {
+
+ public UnixLocalWrapperScriptBuilder(Path containerWorkDir) {
+ super(new Path(containerWorkDir, "default_container_executor.sh"));
+ }
+
+ @Override
+ public void writeLocalWrapperScript(Path launchDst, Path pidFile,
+ PrintStream pout) {
+
+ // We need to do a move as writing to a file is not atomic
+ // Process reading a file being written to may get garbled data
+ // hence write pid to tmp file first followed by a mv
+ pout.println("#!/bin/bash");
+ pout.println();
+ pout.println("echo $$ > " + pidFile.toString() + ".tmp");
+ pout.println("/bin/mv -f " + pidFile.toString() + ".tmp " + pidFile);
+ String exec = ContainerExecutor.isSetsidAvailable? "exec setsid" : "exec";
+ pout.println(exec + " /bin/bash -c \"" +
+ launchDst.toUri().getPath().toString() + "\"");
+ }
+ }
+
+ private final class WindowsLocalWrapperScriptBuilder
+ extends LocalWrapperScriptBuilder {
+
+ private final String containerIdStr;
+
+ public WindowsLocalWrapperScriptBuilder(String containerIdStr,
+ Path containerWorkDir) {
+
+ super(new Path(containerWorkDir, "default_container_executor.cmd"));
+ this.containerIdStr = containerIdStr;
+ }
+
+ @Override
+ public void writeLocalWrapperScript(Path launchDst, Path pidFile,
+ PrintStream pout) {
+
+ // On Windows, the pid is the container ID, so that it can also serve as
+ // the name of the job object created by winutils for task management.
+ // Write to temp file followed by atomic move.
+ String normalizedPidFile = new File(pidFile.toString()).getPath();
+ pout.println("@echo " + containerIdStr + " > " + normalizedPidFile +
+ ".tmp");
+ pout.println("@move /Y " + normalizedPidFile + ".tmp " +
+ normalizedPidFile);
+ pout.println("@call " + launchDst.toString());
+ }
}
@Override
@@ -234,17 +302,13 @@ public class DefaultContainerExecutor extends ContainerExecutor {
: pid;
LOG.debug("Sending signal " + signal.getValue() + " to pid " + sigpid
+ " as user " + user);
- try {
- sendSignal(sigpid, Signal.NULL);
- } catch (ExitCodeException e) {
+ if (!containerIsAlive(sigpid)) {
return false;
}
try {
- sendSignal(sigpid, signal);
+ killContainer(sigpid, signal);
} catch (IOException e) {
- try {
- sendSignal(sigpid, Signal.NULL);
- } catch (IOException ignore) {
+ if (!containerIsAlive(sigpid)) {
return false;
}
throw e;
@@ -252,6 +316,24 @@ public class DefaultContainerExecutor extends ContainerExecutor {
return true;
}
+ /**
+ * Returns true if the process with the specified pid is alive.
+ *
+ * @param pid String pid
+ * @return boolean true if the process is alive
+ */
+ private boolean containerIsAlive(String pid) throws IOException {
+ try {
+ new ShellCommandExecutor(getCheckProcessIsAliveCommand(pid)).execute();
+ // successful execution means process is alive
+ return true;
+ }
+ catch (ExitCodeException e) {
+ // failure (non-zero exit code) means process is not alive
+ return false;
+ }
+ }
+
/**
* Send a specified signal to the specified pid
*
@@ -259,11 +341,9 @@ public class DefaultContainerExecutor extends ContainerExecutor {
* @param signal signal to send
* (for logging).
*/
- protected void sendSignal(String pid, Signal signal) throws IOException {
- ShellCommandExecutor shexec = null;
- String[] arg = { "kill", "-" + signal.getValue(), pid };
- shexec = new ShellCommandExecutor(arg);
- shexec.execute();
+ private void killContainer(String pid, Signal signal) throws IOException {
+ new ShellCommandExecutor(getSignalKillCommand(signal.getValue(), pid))
+ .execute();
}
@Override
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/LocalDirsHandlerService.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/LocalDirsHandlerService.java
index 96a58dda191..517b365d060 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/LocalDirsHandlerService.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/LocalDirsHandlerService.java
@@ -20,7 +20,6 @@ package org.apache.hadoop.yarn.server.nodemanager;
import java.io.IOException;
import java.net.URI;
-import java.net.URISyntaxException;
import java.util.ArrayList;
import java.util.List;
import java.util.Timer;
@@ -305,7 +304,7 @@ public class LocalDirsHandlerService extends AbstractService {
ArrayList validPaths = new ArrayList();
for (int i = 0; i < paths.length; ++i) {
try {
- URI uriPath = new URI(paths[i]);
+ URI uriPath = (new Path(paths[i])).toUri();
if (uriPath.getScheme() == null
|| uriPath.getScheme().equals(FILE_SCHEME)) {
validPaths.add(uriPath.getPath());
@@ -316,7 +315,7 @@ public class LocalDirsHandlerService extends AbstractService {
+ " is not a valid path. Path should be with " + FILE_SCHEME
+ " scheme or without scheme");
}
- } catch (URISyntaxException e) {
+ } catch (IllegalArgumentException e) {
LOG.warn(e.getMessage());
throw new YarnException(paths[i]
+ " is not a valid path. Path should be with " + FILE_SCHEME
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/application/ApplicationEventType.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/application/ApplicationEventType.java
index 24c9a132b0a..b4ba76ae0d5 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/application/ApplicationEventType.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/application/ApplicationEventType.java
@@ -34,5 +34,6 @@ public enum ApplicationEventType {
// Source: Log Handler
APPLICATION_LOG_HANDLING_INITED,
- APPLICATION_LOG_HANDLING_FINISHED
+ APPLICATION_LOG_HANDLING_FINISHED,
+ APPLICATION_LOG_HANDLING_FAILED
}
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/application/ApplicationImpl.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/application/ApplicationImpl.java
index 85f7da1d08c..e48e2fae55a 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/application/ApplicationImpl.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/application/ApplicationImpl.java
@@ -149,6 +149,9 @@ public class ApplicationImpl implements Application {
.addTransition(ApplicationState.INITING, ApplicationState.INITING,
ApplicationEventType.APPLICATION_LOG_HANDLING_INITED,
new AppLogInitDoneTransition())
+ .addTransition(ApplicationState.INITING, ApplicationState.INITING,
+ ApplicationEventType.APPLICATION_LOG_HANDLING_FAILED,
+ new AppLogInitFailTransition())
.addTransition(ApplicationState.INITING, ApplicationState.RUNNING,
ApplicationEventType.APPLICATION_INITED,
new AppInitDoneTransition())
@@ -237,6 +240,26 @@ public class ApplicationImpl implements Application {
}
}
+ /**
+ * Handles the APPLICATION_LOG_HANDLING_FAILED event that occurs after
+ * {@link LogAggregationService} has failed to initialize the log
+ * aggregation service
+ *
+ * In particular, this requests that the {@link ResourceLocalizationService}
+ * localize the application-scoped resources.
+ */
+ @SuppressWarnings("unchecked")
+ static class AppLogInitFailTransition implements
+ SingleArcTransition {
+ @Override
+ public void transition(ApplicationImpl app, ApplicationEvent event) {
+ LOG.warn("Log Aggregation service failed to initialize, there will " +
+ "be no logs for this application");
+ app.dispatcher.getEventHandler().handle(
+ new ApplicationLocalizationEvent(
+ LocalizationEventType.INIT_APPLICATION_RESOURCES, app));
+ }
+ }
/**
* Handles INIT_CONTAINER events which request that we launch a new
* container. When we're still in the INITTING state, we simply
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/launcher/ContainerLaunch.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/launcher/ContainerLaunch.java
index b06788341fe..e1d66bde40a 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/launcher/ContainerLaunch.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/launcher/ContainerLaunch.java
@@ -23,6 +23,7 @@ import static org.apache.hadoop.fs.CreateFlag.OVERWRITE;
import java.io.DataOutputStream;
import java.io.IOException;
+import java.io.File;
import java.io.OutputStream;
import java.io.PrintStream;
import java.util.ArrayList;
@@ -37,6 +38,7 @@ import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileContext;
+import org.apache.hadoop.fs.FileUtil;
import org.apache.hadoop.fs.LocalDirAllocator;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IOUtils;
@@ -69,7 +71,8 @@ public class ContainerLaunch implements Callable {
private static final Log LOG = LogFactory.getLog(ContainerLaunch.class);
- public static final String CONTAINER_SCRIPT = "launch_container.sh";
+ public static final String CONTAINER_SCRIPT = Shell.WINDOWS ?
+ "launch_container.cmd" : "launch_container.sh";
public static final String FINAL_CONTAINER_TOKENS_FILE = "container_tokens";
private static final String PID_FILE_NAME_FMT = "%s.pid";
@@ -130,7 +133,7 @@ public class ContainerLaunch implements Callable {
for (String str : command) {
// TODO: Should we instead work via symlinks without this grammar?
newCmds.add(str.replace(ApplicationConstants.LOG_DIR_EXPANSION_VAR,
- containerLogDir.toUri().getPath()));
+ containerLogDir.toString()));
}
launchContext.setCommands(newCmds);
@@ -141,7 +144,7 @@ public class ContainerLaunch implements Callable {
entry.setValue(
value.replace(
ApplicationConstants.LOG_DIR_EXPANSION_VAR,
- containerLogDir.toUri().getPath())
+ containerLogDir.toString())
);
}
// /////////////////////////// End of variable expansion
@@ -411,28 +414,17 @@ public class ContainerLaunch implements Callable {
+ appIdStr;
}
- private static class ShellScriptBuilder {
-
- private final StringBuilder sb;
-
- public ShellScriptBuilder() {
- this(new StringBuilder("#!/bin/bash\n\n"));
- }
-
- protected ShellScriptBuilder(StringBuilder sb) {
- this.sb = sb;
- }
-
- public ShellScriptBuilder env(String key, String value) {
- line("export ", key, "=\"", value, "\"");
- return this;
- }
-
- public ShellScriptBuilder symlink(Path src, String dst) throws IOException {
- return symlink(src, new Path(dst));
- }
-
- public ShellScriptBuilder symlink(Path src, Path dst) throws IOException {
+ private static abstract class ShellScriptBuilder {
+
+ private static final String LINE_SEPARATOR =
+ System.getProperty("line.separator");
+ private final StringBuilder sb = new StringBuilder();
+
+ public abstract void command(List command);
+
+ public abstract void env(String key, String value);
+
+ public final void symlink(Path src, Path dst) throws IOException {
if (!src.isAbsolute()) {
throw new IOException("Source must be absolute");
}
@@ -440,28 +432,89 @@ public class ContainerLaunch implements Callable {
throw new IOException("Destination must be relative");
}
if (dst.toUri().getPath().indexOf('/') != -1) {
- line("mkdir -p ", dst.getParent().toString());
+ mkdir(dst.getParent());
}
- line("ln -sf \"", src.toUri().getPath(), "\" \"", dst.toString(), "\"");
- return this;
+ link(src, dst);
}
-
- public void write(PrintStream out) throws IOException {
- out.append(sb);
- }
-
- public void line(String... command) {
- for (String s : command) {
- sb.append(s);
- }
- sb.append("\n");
- }
-
+
@Override
public String toString() {
return sb.toString();
}
+ public final void write(PrintStream out) throws IOException {
+ out.append(sb);
+ }
+
+ protected final void line(String... command) {
+ for (String s : command) {
+ sb.append(s);
+ }
+ sb.append(LINE_SEPARATOR);
+ }
+
+ protected abstract void link(Path src, Path dst) throws IOException;
+
+ protected abstract void mkdir(Path path);
+ }
+
+ private static final class UnixShellScriptBuilder extends ShellScriptBuilder {
+
+ public UnixShellScriptBuilder(){
+ line("#!/bin/bash");
+ line();
+ }
+
+ @Override
+ public void command(List command) {
+ line("exec /bin/bash -c \"", StringUtils.join(" ", command), "\"");
+ }
+
+ @Override
+ public void env(String key, String value) {
+ line("export ", key, "=\"", value, "\"");
+ }
+
+ @Override
+ protected void link(Path src, Path dst) throws IOException {
+ line("ln -sf \"", src.toUri().getPath(), "\" \"", dst.toString(), "\"");
+ }
+
+ @Override
+ protected void mkdir(Path path) {
+ line("mkdir -p ", path.toString());
+ }
+ }
+
+ private static final class WindowsShellScriptBuilder
+ extends ShellScriptBuilder {
+
+ public WindowsShellScriptBuilder() {
+ line("@setlocal");
+ line();
+ }
+
+ @Override
+ public void command(List command) {
+ line("@call ", StringUtils.join(" ", command));
+ }
+
+ @Override
+ public void env(String key, String value) {
+ line("@set ", key, "=", value);
+ }
+
+ @Override
+ protected void link(Path src, Path dst) throws IOException {
+ line(String.format("@%s symlink \"%s\" \"%s\"", Shell.WINUTILS,
+ new File(dst.toString()).getPath(),
+ new File(src.toUri().getPath()).getPath()));
+ }
+
+ @Override
+ protected void mkdir(Path path) {
+ line("@if not exist ", path.toString(), " mkdir ", path.toString());
+ }
}
private static void putEnvIfNotNull(
@@ -479,7 +532,7 @@ public class ContainerLaunch implements Callable {
}
public void sanitizeEnv(Map environment,
- Path pwd, List appDirs) {
+ Path pwd, List appDirs) throws IOException {
/**
* Non-modifiable environment variables
*/
@@ -513,6 +566,14 @@ public class ContainerLaunch implements Callable {
environment.put("JVM_PID", "$$");
}
+ // TODO: Remove Windows check and use this approach on all platforms after
+ // additional testing. See YARN-358.
+ if (Shell.WINDOWS) {
+ String inputClassPath = environment.get(Environment.CLASSPATH.name());
+ environment.put(Environment.CLASSPATH.name(),
+ FileUtil.createJarWithClassPath(inputClassPath, pwd));
+ }
+
/**
* Modifiable environment variables
*/
@@ -537,7 +598,8 @@ public class ContainerLaunch implements Callable {
Map environment, Map> resources,
List command)
throws IOException {
- ShellScriptBuilder sb = new ShellScriptBuilder();
+ ShellScriptBuilder sb = Shell.WINDOWS ? new WindowsShellScriptBuilder() :
+ new UnixShellScriptBuilder();
if (environment != null) {
for (Map.Entry env : environment.entrySet()) {
sb.env(env.getKey().toString(), env.getValue().toString());
@@ -546,21 +608,13 @@ public class ContainerLaunch implements Callable {
if (resources != null) {
for (Map.Entry> entry : resources.entrySet()) {
for (String linkName : entry.getValue()) {
- sb.symlink(entry.getKey(), linkName);
+ sb.symlink(entry.getKey(), new Path(linkName));
}
}
}
- ArrayList cmd = new ArrayList(2 * command.size() + 5);
- cmd.add("exec /bin/bash ");
- cmd.add("-c ");
- cmd.add("\"");
- for (String cs : command) {
- cmd.add(cs.toString());
- cmd.add(" ");
- }
- cmd.add("\"");
- sb.line(cmd.toArray(new String[cmd.size()]));
+ sb.command(command);
+
PrintStream pout = null;
try {
pout = new PrintStream(out);
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/localizer/ResourceLocalizationService.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/localizer/ResourceLocalizationService.java
index 1a880fe8e6d..9ca812e1ddd 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/localizer/ResourceLocalizationService.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/localizer/ResourceLocalizationService.java
@@ -659,25 +659,23 @@ public class ResourceLocalizationService extends CompositeService
new ContainerResourceFailedEvent(
assoc.getContext().getContainerId(),
assoc.getResource().getRequest(), e.getCause()));
+ List reqs;
synchronized (attempts) {
LocalResourceRequest req = assoc.getResource().getRequest();
- List reqs = attempts.get(req);
+ reqs = attempts.get(req);
if (null == reqs) {
LOG.error("Missing pending list for " + req);
return;
}
- if (reqs.isEmpty()) {
- attempts.remove(req);
- }
- /*
- * Do not retry for now. Once failed is failed!
- * LocalizerResourceRequestEvent request = reqs.remove(0);
-
- pending.put(queue.submit(new FSDownload(
- lfs, null, conf, publicDirs,
- request.getResource().getRequest(), new Random())),
- request);
- */ }
+ attempts.remove(req);
+ }
+ // let the other containers know about the localization failure
+ for (LocalizerResourceRequestEvent reqEvent : reqs) {
+ dispatcher.getEventHandler().handle(
+ new ContainerResourceFailedEvent(
+ reqEvent.getContext().getContainerId(),
+ reqEvent.getResource().getRequest(), e.getCause()));
+ }
} catch (CancellationException e) {
// ignore; shutting down
}
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/logaggregation/LogAggregationService.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/logaggregation/LogAggregationService.java
index 69c981e198e..88a01eb3854 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/logaggregation/LogAggregationService.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/logaggregation/LogAggregationService.java
@@ -300,8 +300,9 @@ public class LogAggregationService extends AbstractService implements
eventResponse = new ApplicationEvent(appId,
ApplicationEventType.APPLICATION_LOG_HANDLING_INITED);
} catch (YarnException e) {
- eventResponse = new ApplicationFinishEvent(appId,
- "Application failed to init aggregation: " + e.getMessage());
+ LOG.warn("Application failed to init aggregation: " + e.getMessage());
+ eventResponse = new ApplicationEvent(appId,
+ ApplicationEventType.APPLICATION_LOG_HANDLING_FAILED);
}
this.dispatcher.getEventHandler().handle(eventResponse);
}
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/util/ProcessIdFileReader.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/util/ProcessIdFileReader.java
index 279aa29858c..0c7c2505237 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/util/ProcessIdFileReader.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/util/ProcessIdFileReader.java
@@ -25,6 +25,8 @@ import java.io.IOException;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.util.Shell;
+import org.apache.hadoop.yarn.util.ConverterUtils;
/**
* Helper functionality to read the pid from a file.
@@ -62,14 +64,28 @@ public class ProcessIdFileReader {
}
String temp = line.trim();
if (!temp.isEmpty()) {
- try {
- Long pid = Long.valueOf(temp);
- if (pid > 0) {
+ if (Shell.WINDOWS) {
+ // On Windows, pid is expected to be a container ID, so find first
+ // line that parses successfully as a container ID.
+ try {
+ ConverterUtils.toContainerId(temp);
processId = temp;
break;
+ } catch (Exception e) {
+ // do nothing
+ }
+ }
+ else {
+ // Otherwise, find first line containing a numeric pid.
+ try {
+ Long pid = Long.valueOf(temp);
+ if (pid > 0) {
+ processId = temp;
+ break;
+ }
+ } catch (Exception e) {
+ // do nothing
}
- } catch (Exception e) {
- // do nothing
}
}
}
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/test/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/localizer/TestResourceLocalizationService.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/test/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/localizer/TestResourceLocalizationService.java
index b448dea5755..7ca2c91e3c7 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/test/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/localizer/TestResourceLocalizationService.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/test/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/localizer/TestResourceLocalizationService.java
@@ -27,13 +27,16 @@ import static org.mockito.Matchers.argThat;
import static org.mockito.Matchers.eq;
import static org.mockito.Matchers.isA;
import static org.mockito.Matchers.isNull;
+import static org.mockito.Mockito.doAnswer;
import static org.mockito.Mockito.doNothing;
import static org.mockito.Mockito.doReturn;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.timeout;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
+import java.io.IOException;
import java.net.InetSocketAddress;
import java.net.URI;
import java.util.ArrayList;
@@ -46,6 +49,8 @@ import java.util.List;
import java.util.Map;
import java.util.Random;
import java.util.Set;
+import java.util.concurrent.BrokenBarrierException;
+import java.util.concurrent.CyclicBarrier;
import junit.framework.Assert;
@@ -89,6 +94,7 @@ import org.apache.hadoop.yarn.server.nodemanager.containermanager.application.Ap
import org.apache.hadoop.yarn.server.nodemanager.containermanager.container.Container;
import org.apache.hadoop.yarn.server.nodemanager.containermanager.container.ContainerEvent;
import org.apache.hadoop.yarn.server.nodemanager.containermanager.container.ContainerEventType;
+import org.apache.hadoop.yarn.server.nodemanager.containermanager.container.ContainerResourceFailedEvent;
import org.apache.hadoop.yarn.server.nodemanager.containermanager.localizer.ResourceLocalizationService.LocalizerTracker;
import org.apache.hadoop.yarn.server.nodemanager.containermanager.localizer.event.ApplicationLocalizationEvent;
import org.apache.hadoop.yarn.server.nodemanager.containermanager.localizer.event.ContainerLocalizationCleanupEvent;
@@ -102,6 +108,8 @@ import org.junit.BeforeClass;
import org.junit.Test;
import org.mockito.ArgumentCaptor;
import org.mockito.ArgumentMatcher;
+import org.mockito.invocation.InvocationOnMock;
+import org.mockito.stubbing.Answer;
public class TestResourceLocalizationService {
@@ -512,6 +520,111 @@ public class TestResourceLocalizationService {
}
}
+ @Test(timeout=20000)
+ @SuppressWarnings("unchecked") // mocked generics
+ public void testFailedPublicResource() throws Exception {
+ Configuration conf = new YarnConfiguration();
+ AbstractFileSystem spylfs =
+ spy(FileContext.getLocalFSFileContext().getDefaultFileSystem());
+ final FileContext lfs = FileContext.getFileContext(spylfs, conf);
+ doNothing().when(spylfs).mkdir(
+ isA(Path.class), isA(FsPermission.class), anyBoolean());
+ List localDirs = new ArrayList();
+ String[] sDirs = new String[4];
+ for (int i = 0; i < 4; ++i) {
+ localDirs.add(lfs.makeQualified(new Path(basedir, i + "")));
+ sDirs[i] = localDirs.get(i).toString();
+ }
+ conf.setStrings(YarnConfiguration.NM_LOCAL_DIRS, sDirs);
+ String logDir = lfs.makeQualified(new Path(basedir, "logdir " )).toString();
+ conf.set(YarnConfiguration.NM_LOG_DIRS, logDir);
+
+ DrainDispatcher dispatcher = new DrainDispatcher();
+ EventHandler applicationBus = mock(EventHandler.class);
+ dispatcher.register(ApplicationEventType.class, applicationBus);
+ EventHandler containerBus = mock(EventHandler.class);
+ dispatcher.register(ContainerEventType.class, containerBus);
+
+ ContainerExecutor exec = mock(ContainerExecutor.class);
+ DeletionService delService = mock(DeletionService.class);
+ LocalDirsHandlerService dirsHandler = new LocalDirsHandlerService();
+ dirsHandler.init(conf);
+
+ dispatcher.init(conf);
+ dispatcher.start();
+
+ try {
+ ResourceLocalizationService rawService =
+ new ResourceLocalizationService(dispatcher, exec, delService,
+ dirsHandler);
+ ResourceLocalizationService spyService = spy(rawService);
+ doReturn(mockServer).when(spyService).createServer();
+ doReturn(lfs).when(spyService).getLocalFileContext(
+ isA(Configuration.class));
+
+ spyService.init(conf);
+ spyService.start();
+
+ final String user = "user0";
+ // init application
+ final Application app = mock(Application.class);
+ final ApplicationId appId =
+ BuilderUtils.newApplicationId(314159265358979L, 3);
+ when(app.getUser()).thenReturn(user);
+ when(app.getAppId()).thenReturn(appId);
+ spyService.handle(new ApplicationLocalizationEvent(
+ LocalizationEventType.INIT_APPLICATION_RESOURCES, app));
+ dispatcher.await();
+
+ // init container.
+ final Container c = getMockContainer(appId, 42);
+
+ // init resources
+ Random r = new Random();
+ long seed = r.nextLong();
+ System.out.println("SEED: " + seed);
+ r.setSeed(seed);
+
+ // cause chmod to fail after a delay
+ final CyclicBarrier barrier = new CyclicBarrier(2);
+ doAnswer(new Answer() {
+ public Void answer(InvocationOnMock invocation) throws IOException {
+ try {
+ barrier.await();
+ } catch (InterruptedException e) {
+ } catch (BrokenBarrierException e) {
+ }
+ throw new IOException("forced failure");
+ }
+ }).when(spylfs)
+ .setPermission(isA(Path.class), isA(FsPermission.class));
+
+ // Queue up two localization requests for the same public resource
+ final LocalResource pubResource = getPublicMockedResource(r);
+ final LocalResourceRequest pubReq = new LocalResourceRequest(pubResource);
+
+ Map> req =
+ new HashMap>();
+ req.put(LocalResourceVisibility.PUBLIC,
+ Collections.singletonList(pubReq));
+
+ Set pubRsrcs = new HashSet();
+ pubRsrcs.add(pubReq);
+
+ spyService.handle(new ContainerLocalizationRequestEvent(c, req));
+ spyService.handle(new ContainerLocalizationRequestEvent(c, req));
+ dispatcher.await();
+
+ // allow the chmod to fail now that both requests have been queued
+ barrier.await();
+ verify(containerBus, timeout(5000).times(2))
+ .handle(isA(ContainerResourceFailedEvent.class));
+ } finally {
+ dispatcher.stop();
+ }
+ }
+
private static URL getPath(String path) {
URL url = BuilderUtils.newURL("file", null, 0, path);
return url;
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/test/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/logaggregation/TestLogAggregationService.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/test/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/logaggregation/TestLogAggregationService.java
index aad7b845c92..168f619adfb 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/test/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/logaggregation/TestLogAggregationService.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/test/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/logaggregation/TestLogAggregationService.java
@@ -421,8 +421,8 @@ public class TestLogAggregationService extends BaseContainerManagerTest {
dispatcher.await();
ApplicationEvent expectedEvents[] = new ApplicationEvent[]{
- new ApplicationFinishEvent(appId,
- "Application failed to init aggregation: KABOOM!")
+ new ApplicationEvent(appId,
+ ApplicationEventType.APPLICATION_LOG_HANDLING_FAILED)
};
checkEvents(appEventHandler, expectedEvents, false,
"getType", "getApplicationID", "getDiagnostic");
@@ -471,8 +471,8 @@ public class TestLogAggregationService extends BaseContainerManagerTest {
dispatcher.await();
ApplicationEvent expectedEvents[] = new ApplicationEvent[]{
- new ApplicationFinishEvent(appId,
- "Application failed to init aggregation: "+e)
+ new ApplicationEvent(appId,
+ ApplicationEventType.APPLICATION_LOG_HANDLING_FAILED)
};
checkEvents(appEventHandler, expectedEvents, false,
"getType", "getApplicationID", "getDiagnostic");
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/test/java/org/apache/hadoop/yarn/server/nodemanager/util/TestProcessIdFileReader.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/test/java/org/apache/hadoop/yarn/server/nodemanager/util/TestProcessIdFileReader.java
index a8e3e8a989d..0f9e64f3053 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/test/java/org/apache/hadoop/yarn/server/nodemanager/util/TestProcessIdFileReader.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/test/java/org/apache/hadoop/yarn/server/nodemanager/util/TestProcessIdFileReader.java
@@ -26,13 +26,14 @@ import java.io.PrintWriter;
import junit.framework.Assert;
import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.util.Shell;
import org.apache.hadoop.yarn.server.nodemanager.util.ProcessIdFileReader;
import org.junit.Test;
public class TestProcessIdFileReader {
- @Test
+ @Test (timeout = 30000)
public void testNullPath() {
String pid = null;
try {
@@ -44,22 +45,25 @@ public class TestProcessIdFileReader {
assert(pid == null);
}
- @Test
+ @Test (timeout = 30000)
public void testSimpleGet() throws IOException {
String rootDir = new File(System.getProperty(
"test.build.data", "/tmp")).getAbsolutePath();
File testFile = null;
+ String expectedProcessId = Shell.WINDOWS ?
+ "container_1353742680940_0002_01_000001" :
+ "56789";
try {
testFile = new File(rootDir, "temp.txt");
PrintWriter fileWriter = new PrintWriter(testFile);
- fileWriter.println("56789");
+ fileWriter.println(expectedProcessId);
fileWriter.close();
String processId = null;
processId = ProcessIdFileReader.getProcessId(
new Path(rootDir + Path.SEPARATOR + "temp.txt"));
- Assert.assertEquals("56789", processId);
+ Assert.assertEquals(expectedProcessId, processId);
} finally {
if (testFile != null
@@ -70,12 +74,15 @@ public class TestProcessIdFileReader {
}
- @Test
+ @Test (timeout = 30000)
public void testComplexGet() throws IOException {
String rootDir = new File(System.getProperty(
"test.build.data", "/tmp")).getAbsolutePath();
File testFile = null;
-
+ String processIdInFile = Shell.WINDOWS ?
+ " container_1353742680940_0002_01_000001 " :
+ " 23 ";
+ String expectedProcessId = processIdInFile.trim();
try {
testFile = new File(rootDir, "temp.txt");
PrintWriter fileWriter = new PrintWriter(testFile);
@@ -84,14 +91,14 @@ public class TestProcessIdFileReader {
fileWriter.println("abc");
fileWriter.println("-123");
fileWriter.println("-123 ");
- fileWriter.println(" 23 ");
+ fileWriter.println(processIdInFile);
fileWriter.println("6236");
fileWriter.close();
String processId = null;
processId = ProcessIdFileReader.getProcessId(
new Path(rootDir + Path.SEPARATOR + "temp.txt"));
- Assert.assertEquals("23", processId);
+ Assert.assertEquals(expectedProcessId, processId);
} finally {
if (testFile != null
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/ClientRMService.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/ClientRMService.java
index e0522a33fdc..1aaca4e8c29 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/ClientRMService.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/ClientRMService.java
@@ -436,7 +436,6 @@ public class ClientRMService extends AbstractService implements
response.setQueueInfo(queueInfo);
} catch (IOException ioe) {
LOG.info("Failed to getQueueInfo for " + request.getQueueName(), ioe);
- throw RPCUtil.getRemoteException(ioe);
}
return response;
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/ResourceTrackerService.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/ResourceTrackerService.java
index de5fcc68ce7..7aaa0668452 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/ResourceTrackerService.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/ResourceTrackerService.java
@@ -262,8 +262,7 @@ public class ResourceTrackerService extends AbstractService implements
HeartbeatResponse latestResponse = recordFactory
.newRecordInstance(HeartbeatResponse.class);
latestResponse.setResponseId(lastHeartbeatResponse.getResponseId() + 1);
- latestResponse.addAllContainersToCleanup(rmNode.getContainersToCleanUp());
- latestResponse.addAllApplicationsToCleanup(rmNode.getAppsToCleanup());
+ rmNode.updateHeartbeatResponseForCleanup(latestResponse);
latestResponse.setNodeAction(NodeAction.NORMAL);
// Check if node's masterKey needs to be updated and if the currentKey has
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/rmapp/attempt/RMAppAttemptImpl.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/rmapp/attempt/RMAppAttemptImpl.java
index c8bd877efb2..bb3e7338491 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/rmapp/attempt/RMAppAttemptImpl.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/rmapp/attempt/RMAppAttemptImpl.java
@@ -147,6 +147,9 @@ public class RMAppAttemptImpl implements RMAppAttempt, Recoverable {
private Configuration conf;
+ private static final ExpiredTransition EXPIRED_TRANSITION =
+ new ExpiredTransition();
+
private static final StateMachineFactory getAppsToCleanup();
+ /**
+ * Update a {@link HeartbeatResponse} with the list of containers and
+ * applications to clean up for this node.
+ * @param response the {@link HeartbeatResponse} to update
+ */
+ public void updateHeartbeatResponseForCleanup(HeartbeatResponse response);
+
public HeartbeatResponse getLastHeartBeatResponse();
+
+ /**
+ * Get and clear the list of containerUpdates accumulated across NM
+ * heartbeats.
+ *
+ * @return containerUpdates accumulated across NM heartbeats.
+ */
+ public List pullContainerUpdates();
+
}
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/rmnode/RMNodeImpl.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/rmnode/RMNodeImpl.java
index 83833b9bdb3..23a45991390 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/rmnode/RMNodeImpl.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/rmnode/RMNodeImpl.java
@@ -25,6 +25,7 @@ import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.TreeSet;
+import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.locks.ReentrantReadWriteLock;
import java.util.concurrent.locks.ReentrantReadWriteLock.ReadLock;
import java.util.concurrent.locks.ReentrantReadWriteLock.WriteLock;
@@ -60,6 +61,8 @@ import org.apache.hadoop.yarn.state.StateMachine;
import org.apache.hadoop.yarn.state.StateMachineFactory;
import org.apache.hadoop.yarn.util.BuilderUtils.ContainerIdComparator;
+import com.google.common.annotations.VisibleForTesting;
+
/**
* This class is used to keep track of all the applications/containers
* running on a node.
@@ -78,6 +81,9 @@ public class RMNodeImpl implements RMNode, EventHandler {
private final ReadLock readLock;
private final WriteLock writeLock;
+ private final ConcurrentLinkedQueue nodeUpdateQueue;
+ private volatile boolean nextHeartBeat = true;
+
private final NodeId nodeId;
private final RMContext context;
private final String hostName;
@@ -186,6 +192,7 @@ public class RMNodeImpl implements RMNode, EventHandler {
this.stateMachine = stateMachineFactory.make(this);
+ this.nodeUpdateQueue = new ConcurrentLinkedQueue();
}
@Override
@@ -296,6 +303,21 @@ public class RMNodeImpl implements RMNode, EventHandler {
}
};
+ @Override
+ public void updateHeartbeatResponseForCleanup(HeartbeatResponse response) {
+ this.writeLock.lock();
+
+ try {
+ response.addAllContainersToCleanup(
+ new ArrayList(this.containersToClean));
+ response.addAllApplicationsToCleanup(this.finishedApplications);
+ this.containersToClean.clear();
+ this.finishedApplications.clear();
+ } finally {
+ this.writeLock.unlock();
+ }
+ };
+
@Override
public HeartbeatResponse getLastHeartBeatResponse() {
@@ -400,6 +422,7 @@ public class RMNodeImpl implements RMNode, EventHandler {
@Override
public void transition(RMNodeImpl rmNode, RMNodeEvent event) {
// Kill containers since node is rejoining.
+ rmNode.nodeUpdateQueue.clear();
rmNode.context.getDispatcher().getEventHandler().handle(
new NodeRemovedSchedulerEvent(rmNode));
@@ -458,6 +481,7 @@ public class RMNodeImpl implements RMNode, EventHandler {
@Override
public void transition(RMNodeImpl rmNode, RMNodeEvent event) {
// Inform the scheduler
+ rmNode.nodeUpdateQueue.clear();
rmNode.context.getDispatcher().getEventHandler().handle(
new NodeRemovedSchedulerEvent(rmNode));
rmNode.context.getDispatcher().getEventHandler().handle(
@@ -489,6 +513,9 @@ public class RMNodeImpl implements RMNode, EventHandler {
statusEvent.getNodeHealthStatus();
rmNode.setNodeHealthStatus(remoteNodeHealthStatus);
if (!remoteNodeHealthStatus.getIsNodeHealthy()) {
+ LOG.info("Node " + rmNode.nodeId + " reported UNHEALTHY with details: "
+ + remoteNodeHealthStatus.getHealthReport());
+ rmNode.nodeUpdateQueue.clear();
// Inform the scheduler
rmNode.context.getDispatcher().getEventHandler().handle(
new NodeRemovedSchedulerEvent(rmNode));
@@ -538,20 +565,20 @@ public class RMNodeImpl implements RMNode, EventHandler {
completedContainers.add(remoteContainer);
}
}
-
- rmNode.context.getDispatcher().getEventHandler().handle(
- new NodeUpdateSchedulerEvent(rmNode, newlyLaunchedContainers,
- completedContainers));
+ if(newlyLaunchedContainers.size() != 0
+ || completedContainers.size() != 0) {
+ rmNode.nodeUpdateQueue.add(new UpdatedContainerInfo
+ (newlyLaunchedContainers, completedContainers));
+ }
+ if(rmNode.nextHeartBeat) {
+ rmNode.nextHeartBeat = false;
+ rmNode.context.getDispatcher().getEventHandler().handle(
+ new NodeUpdateSchedulerEvent(rmNode));
+ }
rmNode.context.getDelegationTokenRenewer().updateKeepAliveApplications(
statusEvent.getKeepAliveAppIds());
- // HeartBeat processing from our end is done, as node pulls the following
- // lists before sending status-updates. Clear data-structures
- // TODO: These lists could go to the NM multiple times, or never.
- rmNode.containersToClean.clear();
- rmNode.finishedApplications.clear();
-
return NodeState.RUNNING;
}
}
@@ -584,4 +611,25 @@ public class RMNodeImpl implements RMNode, EventHandler {
return NodeState.UNHEALTHY;
}
}
+
+ @Override
+ public List pullContainerUpdates() {
+ List latestContainerInfoList =
+ new ArrayList();
+ while(nodeUpdateQueue.peek() != null){
+ latestContainerInfoList.add(nodeUpdateQueue.poll());
+ }
+ this.nextHeartBeat = true;
+ return latestContainerInfoList;
+ }
+
+ @VisibleForTesting
+ public void setNextHeartBeat(boolean nextHeartBeat) {
+ this.nextHeartBeat = nextHeartBeat;
+ }
+
+ @VisibleForTesting
+ public int getQueueSize() {
+ return nodeUpdateQueue.size();
+ }
}
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/rmnode/UpdatedContainerInfo.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/rmnode/UpdatedContainerInfo.java
new file mode 100644
index 00000000000..284b53665a8
--- /dev/null
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/rmnode/UpdatedContainerInfo.java
@@ -0,0 +1,45 @@
+/**
+ * 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.hadoop.yarn.server.resourcemanager.rmnode;
+
+import java.util.List;
+
+import org.apache.hadoop.yarn.api.records.ContainerStatus;
+
+public class UpdatedContainerInfo {
+ private List newlyLaunchedContainers;
+ private List completedContainers;
+
+ public UpdatedContainerInfo() {
+ }
+
+ public UpdatedContainerInfo(List newlyLaunchedContainers
+ , List completedContainers) {
+ this.newlyLaunchedContainers = newlyLaunchedContainers;
+ this.completedContainers = completedContainers;
+ }
+
+ public List getNewlyLaunchedContainers() {
+ return this.newlyLaunchedContainers;
+ }
+
+ public List getCompletedContainers() {
+ return this.completedContainers;
+ }
+}
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/CapacityScheduler.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/CapacityScheduler.java
index 2ce3a464a86..2fc754069fd 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/CapacityScheduler.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/CapacityScheduler.java
@@ -60,6 +60,7 @@ import org.apache.hadoop.yarn.server.resourcemanager.rmcontainer.RMContainer;
import org.apache.hadoop.yarn.server.resourcemanager.rmcontainer.RMContainerEventType;
import org.apache.hadoop.yarn.server.resourcemanager.rmnode.RMNode;
import org.apache.hadoop.yarn.server.resourcemanager.rmnode.RMNodeCleanContainerEvent;
+import org.apache.hadoop.yarn.server.resourcemanager.rmnode.UpdatedContainerInfo;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.Allocation;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.QueueMetrics;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.ResourceScheduler;
@@ -562,15 +563,20 @@ implements ResourceScheduler, CapacitySchedulerContext, Configurable {
return root.getQueueUserAclInfo(user);
}
- private synchronized void nodeUpdate(RMNode nm,
- List newlyLaunchedContainers,
- List completedContainers) {
+ private synchronized void nodeUpdate(RMNode nm) {
if (LOG.isDebugEnabled()) {
LOG.debug("nodeUpdate: " + nm + " clusterResources: " + clusterResource);
}
-
- FiCaSchedulerNode node = getNode(nm.getNodeID());
+ FiCaSchedulerNode node = getNode(nm.getNodeID());
+ List containerInfoList = nm.pullContainerUpdates();
+ List newlyLaunchedContainers = new ArrayList();
+ List completedContainers = new ArrayList();
+ for(UpdatedContainerInfo containerInfo : containerInfoList) {
+ newlyLaunchedContainers.addAll(containerInfo.getNewlyLaunchedContainers());
+ completedContainers.addAll(containerInfo.getCompletedContainers());
+ }
+
// Processing the newly launched containers
for (ContainerStatus launchedContainer : newlyLaunchedContainers) {
containerLaunchedOnNode(launchedContainer.getContainerId(), node);
@@ -666,9 +672,7 @@ implements ResourceScheduler, CapacitySchedulerContext, Configurable {
case NODE_UPDATE:
{
NodeUpdateSchedulerEvent nodeUpdatedEvent = (NodeUpdateSchedulerEvent)event;
- nodeUpdate(nodeUpdatedEvent.getRMNode(),
- nodeUpdatedEvent.getNewlyLaunchedContainers(),
- nodeUpdatedEvent.getCompletedContainers());
+ nodeUpdate(nodeUpdatedEvent.getRMNode());
}
break;
case APP_ADDED:
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/event/NodeUpdateSchedulerEvent.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/event/NodeUpdateSchedulerEvent.java
index ff51d62d910..7a8686c83fa 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/event/NodeUpdateSchedulerEvent.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/event/NodeUpdateSchedulerEvent.java
@@ -18,35 +18,18 @@
package org.apache.hadoop.yarn.server.resourcemanager.scheduler.event;
-import java.util.List;
-
-import org.apache.hadoop.yarn.api.records.ContainerStatus;
import org.apache.hadoop.yarn.server.resourcemanager.rmnode.RMNode;
public class NodeUpdateSchedulerEvent extends SchedulerEvent {
private final RMNode rmNode;
- private final List newlyLaunchedContainers;
- private final List completedContainersStatuses;
- public NodeUpdateSchedulerEvent(RMNode rmNode,
- List newlyLaunchedContainers,
- List completedContainers) {
+ public NodeUpdateSchedulerEvent(RMNode rmNode) {
super(SchedulerEventType.NODE_UPDATE);
this.rmNode = rmNode;
- this.newlyLaunchedContainers = newlyLaunchedContainers;
- this.completedContainersStatuses = completedContainers;
}
public RMNode getRMNode() {
return rmNode;
}
-
- public List getNewlyLaunchedContainers() {
- return newlyLaunchedContainers;
- }
-
- public List getCompletedContainers() {
- return completedContainersStatuses;
- }
}
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/FairScheduler.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/FairScheduler.java
index ab0f1a4f6f1..45cad65738d 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/FairScheduler.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/FairScheduler.java
@@ -33,7 +33,6 @@ import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.classification.InterfaceAudience.LimitedPrivate;
import org.apache.hadoop.classification.InterfaceStability.Unstable;
import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.security.AccessControlException;
import org.apache.hadoop.security.UserGroupInformation;
import org.apache.hadoop.yarn.Clock;
import org.apache.hadoop.yarn.SystemClock;
@@ -61,6 +60,7 @@ import org.apache.hadoop.yarn.server.resourcemanager.rmcontainer.RMContainer;
import org.apache.hadoop.yarn.server.resourcemanager.rmcontainer.RMContainerEventType;
import org.apache.hadoop.yarn.server.resourcemanager.rmcontainer.RMContainerState;
import org.apache.hadoop.yarn.server.resourcemanager.rmnode.RMNode;
+import org.apache.hadoop.yarn.server.resourcemanager.rmnode.UpdatedContainerInfo;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.ActiveUsersManager;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.Allocation;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.QueueMetrics;
@@ -750,15 +750,20 @@ public class FairScheduler implements ResourceScheduler {
/**
* Process a heartbeat update from a node.
*/
- private synchronized void nodeUpdate(RMNode nm,
- List newlyLaunchedContainers,
- List completedContainers) {
+ private synchronized void nodeUpdate(RMNode nm) {
if (LOG.isDebugEnabled()) {
LOG.debug("nodeUpdate: " + nm + " cluster capacity: " + clusterCapacity);
}
eventLog.log("HEARTBEAT", nm.getHostName());
FSSchedulerNode node = nodes.get(nm.getNodeID());
+ List containerInfoList = nm.pullContainerUpdates();
+ List newlyLaunchedContainers = new ArrayList();
+ List completedContainers = new ArrayList();
+ for(UpdatedContainerInfo containerInfo : containerInfoList) {
+ newlyLaunchedContainers.addAll(containerInfo.getNewlyLaunchedContainers());
+ completedContainers.addAll(containerInfo.getCompletedContainers());
+ }
// Processing the newly launched containers
for (ContainerStatus launchedContainer : newlyLaunchedContainers) {
containerLaunchedOnNode(launchedContainer.getContainerId(), node);
@@ -864,9 +869,7 @@ public class FairScheduler implements ResourceScheduler {
throw new RuntimeException("Unexpected event type: " + event);
}
NodeUpdateSchedulerEvent nodeUpdatedEvent = (NodeUpdateSchedulerEvent)event;
- nodeUpdate(nodeUpdatedEvent.getRMNode(),
- nodeUpdatedEvent.getNewlyLaunchedContainers(),
- nodeUpdatedEvent.getCompletedContainers());
+ nodeUpdate(nodeUpdatedEvent.getRMNode());
break;
case APP_ADDED:
if (!(event instanceof AppAddedSchedulerEvent)) {
@@ -966,7 +969,7 @@ public class FairScheduler implements ResourceScheduler {
public QueueInfo getQueueInfo(String queueName, boolean includeChildQueues,
boolean recursive) throws IOException {
if (!queueMgr.exists(queueName)) {
- return null;
+ throw new IOException("queue " + queueName + " does not exist");
}
return queueMgr.getQueue(queueName).getQueueInfo(includeChildQueues,
recursive);
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fifo/FifoScheduler.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fifo/FifoScheduler.java
index 880b98ae61e..3f255378231 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fifo/FifoScheduler.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fifo/FifoScheduler.java
@@ -67,6 +67,7 @@ import org.apache.hadoop.yarn.server.resourcemanager.rmcontainer.RMContainer;
import org.apache.hadoop.yarn.server.resourcemanager.rmcontainer.RMContainerEventType;
import org.apache.hadoop.yarn.server.resourcemanager.rmnode.RMNode;
import org.apache.hadoop.yarn.server.resourcemanager.rmnode.RMNodeCleanContainerEvent;
+import org.apache.hadoop.yarn.server.resourcemanager.rmnode.UpdatedContainerInfo;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.ActiveUsersManager;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.Allocation;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.NodeType;
@@ -576,11 +577,16 @@ public class FifoScheduler implements ResourceScheduler, Configurable {
return assignedContainers;
}
- private synchronized void nodeUpdate(RMNode rmNode,
- List newlyLaunchedContainers,
- List completedContainers) {
+ private synchronized void nodeUpdate(RMNode rmNode) {
FiCaSchedulerNode node = getNode(rmNode.getNodeID());
+ List containerInfoList = rmNode.pullContainerUpdates();
+ List newlyLaunchedContainers = new ArrayList();
+ List completedContainers = new ArrayList();
+ for(UpdatedContainerInfo containerInfo : containerInfoList) {
+ newlyLaunchedContainers.addAll(containerInfo.getNewlyLaunchedContainers());
+ completedContainers.addAll(containerInfo.getCompletedContainers());
+ }
// Processing the newly launched containers
for (ContainerStatus launchedContainer : newlyLaunchedContainers) {
containerLaunchedOnNode(launchedContainer.getContainerId(), node);
@@ -628,9 +634,7 @@ public class FifoScheduler implements ResourceScheduler, Configurable {
{
NodeUpdateSchedulerEvent nodeUpdatedEvent =
(NodeUpdateSchedulerEvent)event;
- nodeUpdate(nodeUpdatedEvent.getRMNode(),
- nodeUpdatedEvent.getNewlyLaunchedContainers(),
- nodeUpdatedEvent.getCompletedContainers());
+ nodeUpdate(nodeUpdatedEvent.getRMNode());
}
break;
case APP_ADDED:
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/MockNodes.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/MockNodes.java
index 0c56a27ada0..2c0fb6cf1d3 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/MockNodes.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/MockNodes.java
@@ -18,7 +18,9 @@
package org.apache.hadoop.yarn.server.resourcemanager;
+import java.util.ArrayList;
import java.util.List;
+import java.util.concurrent.ConcurrentLinkedQueue;
import org.apache.hadoop.net.Node;
import org.apache.hadoop.yarn.api.records.ApplicationId;
@@ -31,6 +33,7 @@ import org.apache.hadoop.yarn.factories.RecordFactory;
import org.apache.hadoop.yarn.factory.providers.RecordFactoryProvider;
import org.apache.hadoop.yarn.server.api.records.HeartbeatResponse;
import org.apache.hadoop.yarn.server.resourcemanager.rmnode.RMNode;
+import org.apache.hadoop.yarn.server.resourcemanager.rmnode.UpdatedContainerInfo;
import com.google.common.collect.Lists;
@@ -183,10 +186,19 @@ public class MockNodes {
return null;
}
+ @Override
+ public void updateHeartbeatResponseForCleanup(HeartbeatResponse response) {
+ }
+
@Override
public HeartbeatResponse getLastHeartBeatResponse() {
return null;
}
+
+ @Override
+ public List pullContainerUpdates() {
+ return new ArrayList();
+ }
};
private static RMNode buildRMNode(int rack, final Resource perNode, NodeState state, String httpAddr) {
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/TestClientRMService.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/TestClientRMService.java
index 871755c8f5f..4dcf84a34e4 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/TestClientRMService.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/TestClientRMService.java
@@ -19,7 +19,7 @@
package org.apache.hadoop.yarn.server.resourcemanager;
import static org.mockito.Matchers.anyBoolean;
-import static org.mockito.Matchers.anyString;
+import static org.mockito.Matchers.eq;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
@@ -176,6 +176,10 @@ public class TestClientRMService {
List applications = queueInfo.getQueueInfo()
.getApplications();
Assert.assertEquals(2, applications.size());
+ request.setQueueName("nonexistentqueue");
+ request.setIncludeApplications(true);
+ // should not throw exception on nonexistent queue
+ queueInfo = rmService.getQueueInfo(request);
}
private static final UserGroupInformation owner =
@@ -334,8 +338,10 @@ public class TestClientRMService {
when(rmContext.getDispatcher()).thenReturn(dispatcher);
QueueInfo queInfo = recordFactory.newRecordInstance(QueueInfo.class);
queInfo.setQueueName("testqueue");
- when(yarnScheduler.getQueueInfo(anyString(), anyBoolean(), anyBoolean()))
+ when(yarnScheduler.getQueueInfo(eq("testqueue"), anyBoolean(), anyBoolean()))
.thenReturn(queInfo);
+ when(yarnScheduler.getQueueInfo(eq("nonexistentqueue"), anyBoolean(), anyBoolean()))
+ .thenThrow(new IOException("queue does not exist"));
ConcurrentHashMap apps = getRMApps(rmContext,
yarnScheduler);
when(rmContext.getRMApps()).thenReturn(apps);
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/TestFifoScheduler.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/TestFifoScheduler.java
index 2a708ed2d54..9190433ce48 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/TestFifoScheduler.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/TestFifoScheduler.java
@@ -201,7 +201,7 @@ public class TestFifoScheduler {
testMinimumAllocation(conf, allocMB / 2);
}
- @Test
+ @Test (timeout = 5000)
public void testReconnectedNode() throws Exception {
CapacitySchedulerConfiguration conf = new CapacitySchedulerConfiguration();
conf.setQueues("default", new String[] {"default"});
@@ -215,19 +215,19 @@ public class TestFifoScheduler {
fs.handle(new NodeAddedSchedulerEvent(n1));
fs.handle(new NodeAddedSchedulerEvent(n2));
List emptyList = new ArrayList();
- fs.handle(new NodeUpdateSchedulerEvent(n1, emptyList, emptyList));
+ fs.handle(new NodeUpdateSchedulerEvent(n1));
Assert.assertEquals(6 * GB, fs.getRootQueueMetrics().getAvailableMB());
// reconnect n1 with downgraded memory
n1 = MockNodes.newNodeInfo(0, MockNodes.newResource(2 * GB), 1);
fs.handle(new NodeRemovedSchedulerEvent(n1));
fs.handle(new NodeAddedSchedulerEvent(n1));
- fs.handle(new NodeUpdateSchedulerEvent(n1, emptyList, emptyList));
+ fs.handle(new NodeUpdateSchedulerEvent(n1));
Assert.assertEquals(4 * GB, fs.getRootQueueMetrics().getAvailableMB());
}
- @Test
+ @Test (timeout = 5000)
public void testHeadroom() throws Exception {
Configuration conf = new Configuration();
@@ -275,7 +275,7 @@ public class TestFifoScheduler {
fs.allocate(appAttemptId2, ask2, emptyId);
// Trigger container assignment
- fs.handle(new NodeUpdateSchedulerEvent(n1, emptyStatus, emptyStatus));
+ fs.handle(new NodeUpdateSchedulerEvent(n1));
// Get the allocation for the applications and verify headroom
Allocation allocation1 = fs.allocate(appAttemptId1, emptyAsk, emptyId);
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/TestRMNodeTransitions.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/TestRMNodeTransitions.java
index 6c14008626a..e165e417cd9 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/TestRMNodeTransitions.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/TestRMNodeTransitions.java
@@ -22,6 +22,7 @@ import static org.mockito.Mockito.doAnswer;
import static org.mockito.Mockito.doReturn;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.*;
import java.util.ArrayList;
import java.util.Collections;
@@ -29,6 +30,7 @@ import java.util.List;
import junit.framework.Assert;
+import org.apache.hadoop.yarn.api.records.ApplicationId;
import org.apache.hadoop.yarn.api.records.ContainerId;
import org.apache.hadoop.yarn.api.records.ContainerStatus;
import org.apache.hadoop.yarn.api.records.NodeHealthStatus;
@@ -37,11 +39,13 @@ import org.apache.hadoop.yarn.api.records.NodeState;
import org.apache.hadoop.yarn.event.EventHandler;
import org.apache.hadoop.yarn.event.InlineDispatcher;
import org.apache.hadoop.yarn.server.api.records.HeartbeatResponse;
+import org.apache.hadoop.yarn.server.resourcemanager.rmnode.RMNodeCleanAppEvent;
import org.apache.hadoop.yarn.server.resourcemanager.rmnode.RMNodeCleanContainerEvent;
import org.apache.hadoop.yarn.server.resourcemanager.rmnode.RMNodeEvent;
import org.apache.hadoop.yarn.server.resourcemanager.rmnode.RMNodeEventType;
import org.apache.hadoop.yarn.server.resourcemanager.rmnode.RMNodeImpl;
import org.apache.hadoop.yarn.server.resourcemanager.rmnode.RMNodeStatusEvent;
+import org.apache.hadoop.yarn.server.resourcemanager.rmnode.UpdatedContainerInfo;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.YarnScheduler;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.event.NodeAddedSchedulerEvent;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.event.NodeUpdateSchedulerEvent;
@@ -49,6 +53,7 @@ import org.apache.hadoop.yarn.server.resourcemanager.scheduler.event.SchedulerEv
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.event.SchedulerEventType;
import org.apache.hadoop.yarn.server.resourcemanager.security.DelegationTokenRenewer;
import org.apache.hadoop.yarn.util.BuilderUtils;
+import org.apache.hadoop.yarn.util.Records;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;
@@ -63,7 +68,7 @@ public class TestRMNodeTransitions {
private YarnScheduler scheduler;
private SchedulerEventType eventType;
- private List completedContainers;
+ private List completedContainers = new ArrayList();
private final class TestSchedulerEventDispatcher implements
EventHandler {
@@ -89,10 +94,11 @@ public class TestRMNodeTransitions {
final SchedulerEvent event = (SchedulerEvent)(invocation.getArguments()[0]);
eventType = event.getType();
if (eventType == SchedulerEventType.NODE_UPDATE) {
- completedContainers =
- ((NodeUpdateSchedulerEvent)event).getCompletedContainers();
- } else {
- completedContainers = null;
+ List lastestContainersInfoList =
+ ((NodeUpdateSchedulerEvent)event).getRMNode().pullContainerUpdates();
+ for(UpdatedContainerInfo lastestContainersInfo : lastestContainersInfoList) {
+ completedContainers.addAll(lastestContainersInfo.getCompletedContainers());
+ }
}
return null;
}
@@ -125,16 +131,16 @@ public class TestRMNodeTransitions {
return event;
}
- @Test
+ @Test (timeout = 5000)
public void testExpiredContainer() {
// Start the node
node.handle(new RMNodeEvent(null, RMNodeEventType.STARTED));
verify(scheduler).handle(any(NodeAddedSchedulerEvent.class));
// Expire a container
- ContainerId completedContainerId = BuilderUtils.newContainerId(
- BuilderUtils.newApplicationAttemptId(
- BuilderUtils.newApplicationId(0, 0), 0), 0);
+ ContainerId completedContainerId = BuilderUtils.newContainerId(
+ BuilderUtils.newApplicationAttemptId(
+ BuilderUtils.newApplicationId(0, 0), 0), 0);
node.handle(new RMNodeCleanContainerEvent(null, completedContainerId));
Assert.assertEquals(1, node.getContainersToCleanUp().size());
@@ -146,9 +152,110 @@ public class TestRMNodeTransitions {
doReturn(Collections.singletonList(containerStatus)).
when(statusEvent).getContainers();
node.handle(statusEvent);
- Assert.assertEquals(0, completedContainers.size());
+ /* Expect the scheduler call handle function 2 times
+ * 1. RMNode status from new to Running, handle the add_node event
+ * 2. handle the node update event
+ */
+ verify(scheduler,times(2)).handle(any(NodeUpdateSchedulerEvent.class));
}
+
+ @Test (timeout = 5000)
+ public void testContainerUpdate() throws InterruptedException{
+ //Start the node
+ node.handle(new RMNodeEvent(null,RMNodeEventType.STARTED));
+
+ NodeId nodeId = BuilderUtils.newNodeId("localhost:1", 1);
+ RMNodeImpl node2 = new RMNodeImpl(nodeId, rmContext, null, 0, 0, null, null);
+ node2.handle(new RMNodeEvent(null,RMNodeEventType.STARTED));
+
+ ContainerId completedContainerIdFromNode1 = BuilderUtils.newContainerId(
+ BuilderUtils.newApplicationAttemptId(
+ BuilderUtils.newApplicationId(0, 0), 0), 0);
+ ContainerId completedContainerIdFromNode2_1 = BuilderUtils.newContainerId(
+ BuilderUtils.newApplicationAttemptId(
+ BuilderUtils.newApplicationId(1, 1), 1), 1);
+ ContainerId completedContainerIdFromNode2_2 = BuilderUtils.newContainerId(
+ BuilderUtils.newApplicationAttemptId(
+ BuilderUtils.newApplicationId(1, 1), 1), 2);
+
+ RMNodeStatusEvent statusEventFromNode1 = getMockRMNodeStatusEvent();
+ RMNodeStatusEvent statusEventFromNode2_1 = getMockRMNodeStatusEvent();
+ RMNodeStatusEvent statusEventFromNode2_2 = getMockRMNodeStatusEvent();
+
+ ContainerStatus containerStatusFromNode1 = mock(ContainerStatus.class);
+ ContainerStatus containerStatusFromNode2_1 = mock(ContainerStatus.class);
+ ContainerStatus containerStatusFromNode2_2 = mock(ContainerStatus.class);
+ doReturn(completedContainerIdFromNode1).when(containerStatusFromNode1)
+ .getContainerId();
+ doReturn(Collections.singletonList(containerStatusFromNode1))
+ .when(statusEventFromNode1).getContainers();
+ node.handle(statusEventFromNode1);
+ Assert.assertEquals(1, completedContainers.size());
+ Assert.assertEquals(completedContainerIdFromNode1,
+ completedContainers.get(0).getContainerId());
+
+ completedContainers.clear();
+
+ doReturn(completedContainerIdFromNode2_1).when(containerStatusFromNode2_1)
+ .getContainerId();
+ doReturn(Collections.singletonList(containerStatusFromNode2_1))
+ .when(statusEventFromNode2_1).getContainers();
+
+ doReturn(completedContainerIdFromNode2_2).when(containerStatusFromNode2_2)
+ .getContainerId();
+ doReturn(Collections.singletonList(containerStatusFromNode2_2))
+ .when(statusEventFromNode2_2).getContainers();
+
+ node2.setNextHeartBeat(false);
+ node2.handle(statusEventFromNode2_1);
+ node2.setNextHeartBeat(true);
+ node2.handle(statusEventFromNode2_2);
+
+ Assert.assertEquals(2, completedContainers.size());
+ Assert.assertEquals(completedContainerIdFromNode2_1,completedContainers.get(0)
+ .getContainerId());
+ Assert.assertEquals(completedContainerIdFromNode2_2,completedContainers.get(1)
+ .getContainerId());
+ }
+
+ @Test (timeout = 5000)
+ public void testStatusChange(){
+ //Start the node
+ node.handle(new RMNodeEvent(null,RMNodeEventType.STARTED));
+ //Add info to the queue first
+ node.setNextHeartBeat(false);
+
+ ContainerId completedContainerId1 = BuilderUtils.newContainerId(
+ BuilderUtils.newApplicationAttemptId(
+ BuilderUtils.newApplicationId(0, 0), 0), 0);
+ ContainerId completedContainerId2 = BuilderUtils.newContainerId(
+ BuilderUtils.newApplicationAttemptId(
+ BuilderUtils.newApplicationId(1, 1), 1), 1);
+
+ RMNodeStatusEvent statusEvent1 = getMockRMNodeStatusEvent();
+ RMNodeStatusEvent statusEvent2 = getMockRMNodeStatusEvent();
+
+ ContainerStatus containerStatus1 = mock(ContainerStatus.class);
+ ContainerStatus containerStatus2 = mock(ContainerStatus.class);
+
+ doReturn(completedContainerId1).when(containerStatus1).getContainerId();
+ doReturn(Collections.singletonList(containerStatus1))
+ .when(statusEvent1).getContainers();
+
+ doReturn(completedContainerId2).when(containerStatus2).getContainerId();
+ doReturn(Collections.singletonList(containerStatus2))
+ .when(statusEvent2).getContainers();
+
+ verify(scheduler,times(1)).handle(any(NodeUpdateSchedulerEvent.class));
+ node.handle(statusEvent1);
+ node.handle(statusEvent2);
+ verify(scheduler,times(1)).handle(any(NodeUpdateSchedulerEvent.class));
+ Assert.assertEquals(2, node.getQueueSize());
+ node.handle(new RMNodeEvent(node.getNodeID(), RMNodeEventType.EXPIRE));
+ Assert.assertEquals(0, node.getQueueSize());
+ }
+
@Test
public void testRunningExpire() {
RMNodeImpl node = getRunningNode();
@@ -195,6 +302,39 @@ public class TestRMNodeTransitions {
Assert.assertEquals(NodeState.REBOOTED, node.getState());
}
+ @Test(timeout=20000)
+ public void testUpdateHeartbeatResponseForCleanup() {
+ RMNodeImpl node = getRunningNode();
+ NodeId nodeId = node.getNodeID();
+
+ // Expire a container
+ ContainerId completedContainerId = BuilderUtils.newContainerId(
+ BuilderUtils.newApplicationAttemptId(
+ BuilderUtils.newApplicationId(0, 0), 0), 0);
+ node.handle(new RMNodeCleanContainerEvent(nodeId, completedContainerId));
+ Assert.assertEquals(1, node.getContainersToCleanUp().size());
+
+ // Finish an application
+ ApplicationId finishedAppId = BuilderUtils.newApplicationId(0, 1);
+ node.handle(new RMNodeCleanAppEvent(nodeId, finishedAppId));
+ Assert.assertEquals(1, node.getAppsToCleanup().size());
+
+ // Verify status update does not clear containers/apps to cleanup
+ // but updating heartbeat response for cleanup does
+ RMNodeStatusEvent statusEvent = getMockRMNodeStatusEvent();
+ node.handle(statusEvent);
+ Assert.assertEquals(1, node.getContainersToCleanUp().size());
+ Assert.assertEquals(1, node.getAppsToCleanup().size());
+ HeartbeatResponse hbrsp = Records.newRecord(HeartbeatResponse.class);
+ node.updateHeartbeatResponseForCleanup(hbrsp);
+ Assert.assertEquals(0, node.getContainersToCleanUp().size());
+ Assert.assertEquals(0, node.getAppsToCleanup().size());
+ Assert.assertEquals(1, hbrsp.getContainersToCleanupCount());
+ Assert.assertEquals(completedContainerId, hbrsp.getContainerToCleanup(0));
+ Assert.assertEquals(1, hbrsp.getApplicationsToCleanupCount());
+ Assert.assertEquals(finishedAppId, hbrsp.getApplicationsToCleanup(0));
+ }
+
private RMNodeImpl getRunningNode() {
NodeId nodeId = BuilderUtils.newNodeId("localhost", 0);
RMNodeImpl node = new RMNodeImpl(nodeId, rmContext,null, 0, 0,
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/rmapp/attempt/TestRMAppAttemptTransitions.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/rmapp/attempt/TestRMAppAttemptTransitions.java
index c907df389b5..f10e646b85d 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/rmapp/attempt/TestRMAppAttemptTransitions.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/rmapp/attempt/TestRMAppAttemptTransitions.java
@@ -22,6 +22,7 @@ import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertTrue;
import static org.mockito.Matchers.any;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.times;
@@ -665,6 +666,39 @@ public class TestRMAppAttemptTransitions {
assertEquals(rmAppPageUrl, applicationAttempt.getTrackingUrl());
}
+ @Test(timeout=10000)
+ public void testLaunchedExpire() {
+ Container amContainer = allocateApplicationAttempt();
+ launchApplicationAttempt(amContainer);
+ applicationAttempt.handle(new RMAppAttemptEvent(
+ applicationAttempt.getAppAttemptId(), RMAppAttemptEventType.EXPIRE));
+ assertEquals(RMAppAttemptState.FAILED,
+ applicationAttempt.getAppAttemptState());
+ assertTrue("expire diagnostics missing",
+ applicationAttempt.getDiagnostics().contains("timed out"));
+ String rmAppPageUrl = pjoin(RM_WEBAPP_ADDR, "cluster", "app",
+ applicationAttempt.getAppAttemptId().getApplicationId());
+ assertEquals(rmAppPageUrl, applicationAttempt.getOriginalTrackingUrl());
+ assertEquals(rmAppPageUrl, applicationAttempt.getTrackingUrl());
+ }
+
+ @Test(timeout=20000)
+ public void testRunningExpire() {
+ Container amContainer = allocateApplicationAttempt();
+ launchApplicationAttempt(amContainer);
+ runApplicationAttempt(amContainer, "host", 8042, "oldtrackingurl");
+ applicationAttempt.handle(new RMAppAttemptEvent(
+ applicationAttempt.getAppAttemptId(), RMAppAttemptEventType.EXPIRE));
+ assertEquals(RMAppAttemptState.FAILED,
+ applicationAttempt.getAppAttemptState());
+ assertTrue("expire diagnostics missing",
+ applicationAttempt.getDiagnostics().contains("timed out"));
+ String rmAppPageUrl = pjoin(RM_WEBAPP_ADDR, "cluster", "app",
+ applicationAttempt.getAppAttemptId().getApplicationId());
+ assertEquals(rmAppPageUrl, applicationAttempt.getOriginalTrackingUrl());
+ assertEquals(rmAppPageUrl, applicationAttempt.getTrackingUrl());
+ }
+
@Test
public void testUnregisterToKilledFinishing() {
Container amContainer = allocateApplicationAttempt();
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/TestFairScheduler.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/TestFairScheduler.java
index a0e17588e03..a5cfadd2558 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/TestFairScheduler.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/TestFairScheduler.java
@@ -30,7 +30,6 @@ import java.io.PrintWriter;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
-import java.util.LinkedList;
import java.util.List;
import java.util.Map;
@@ -276,7 +275,7 @@ public class TestFairScheduler {
Assert.assertEquals(3, queueManager.getLeafQueues().size());
}
- @Test
+ @Test (timeout = 5000)
public void testSimpleContainerAllocation() {
// Add a node
RMNode node1 = MockNodes.newNodeInfo(1, Resources.createResource(1024));
@@ -292,8 +291,7 @@ public class TestFairScheduler {
scheduler.update();
- NodeUpdateSchedulerEvent updateEvent = new NodeUpdateSchedulerEvent(node1,
- new ArrayList(), new ArrayList());
+ NodeUpdateSchedulerEvent updateEvent = new NodeUpdateSchedulerEvent(node1);
scheduler.handle(updateEvent);
// Asked for less than min_allocation.
@@ -301,15 +299,14 @@ public class TestFairScheduler {
scheduler.getQueueManager().getQueue("queue1").
getResourceUsage().getMemory());
- NodeUpdateSchedulerEvent updateEvent2 = new NodeUpdateSchedulerEvent(node2,
- new ArrayList(), new ArrayList());
+ NodeUpdateSchedulerEvent updateEvent2 = new NodeUpdateSchedulerEvent(node2);
scheduler.handle(updateEvent2);
assertEquals(1024, scheduler.getQueueManager().getQueue("queue1").
getResourceUsage().getMemory());
}
- @Test
+ @Test (timeout = 5000)
public void testSimpleContainerReservation() throws InterruptedException {
// Add a node
RMNode node1 = MockNodes.newNodeInfo(1, Resources.createResource(1024));
@@ -319,8 +316,7 @@ public class TestFairScheduler {
// Queue 1 requests full capacity of node
createSchedulingRequest(1024, "queue1", "user1", 1);
scheduler.update();
- NodeUpdateSchedulerEvent updateEvent = new NodeUpdateSchedulerEvent(node1,
- new ArrayList(), new ArrayList());
+ NodeUpdateSchedulerEvent updateEvent = new NodeUpdateSchedulerEvent(node1);
scheduler.handle(updateEvent);
// Make sure queue 1 is allocated app capacity
@@ -340,8 +336,7 @@ public class TestFairScheduler {
// Now another node checks in with capacity
RMNode node2 = MockNodes.newNodeInfo(1, Resources.createResource(1024));
NodeAddedSchedulerEvent nodeEvent2 = new NodeAddedSchedulerEvent(node2);
- NodeUpdateSchedulerEvent updateEvent2 = new NodeUpdateSchedulerEvent(node2,
- new ArrayList(), new ArrayList());
+ NodeUpdateSchedulerEvent updateEvent2 = new NodeUpdateSchedulerEvent(node2);
scheduler.handle(nodeEvent2);
scheduler.handle(updateEvent2);
@@ -738,7 +733,7 @@ public class TestFairScheduler {
assertEquals(300000, queueManager.getFairSharePreemptionTimeout());
}
- @Test
+ @Test (timeout = 5000)
public void testIsStarvedForMinShare() throws Exception {
Configuration conf = createConfiguration();
conf.set(FairSchedulerConfiguration.ALLOCATION_FILE, ALLOC_FILE);
@@ -767,8 +762,7 @@ public class TestFairScheduler {
// Queue A wants 3 * 1024. Node update gives this all to A
createSchedulingRequest(3 * 1024, "queueA", "user1");
scheduler.update();
- NodeUpdateSchedulerEvent nodeEvent2 = new NodeUpdateSchedulerEvent(node1,
- new LinkedList(), new LinkedList());
+ NodeUpdateSchedulerEvent nodeEvent2 = new NodeUpdateSchedulerEvent(node1);
scheduler.handle(nodeEvent2);
// Queue B arrives and wants 1 * 1024
@@ -797,7 +791,7 @@ public class TestFairScheduler {
}
}
- @Test
+ @Test (timeout = 5000)
public void testIsStarvedForFairShare() throws Exception {
Configuration conf = createConfiguration();
conf.set(FairSchedulerConfiguration.ALLOCATION_FILE, ALLOC_FILE);
@@ -826,8 +820,7 @@ public class TestFairScheduler {
// Queue A wants 3 * 1024. Node update gives this all to A
createSchedulingRequest(3 * 1024, "queueA", "user1");
scheduler.update();
- NodeUpdateSchedulerEvent nodeEvent2 = new NodeUpdateSchedulerEvent(node1,
- new LinkedList(), new LinkedList());
+ NodeUpdateSchedulerEvent nodeEvent2 = new NodeUpdateSchedulerEvent(node1);
scheduler.handle(nodeEvent2);
// Queue B arrives and wants 1 * 1024
@@ -857,7 +850,7 @@ public class TestFairScheduler {
}
}
- @Test
+ @Test (timeout = 5000)
/**
* Make sure containers are chosen to be preempted in the correct order. Right
* now this means decreasing order of priority.
@@ -921,16 +914,13 @@ public class TestFairScheduler {
// Sufficient node check-ins to fully schedule containers
for (int i = 0; i < 2; i++) {
- NodeUpdateSchedulerEvent nodeUpdate1 = new NodeUpdateSchedulerEvent(node1,
- new LinkedList(), new LinkedList());
+ NodeUpdateSchedulerEvent nodeUpdate1 = new NodeUpdateSchedulerEvent(node1);
scheduler.handle(nodeUpdate1);
- NodeUpdateSchedulerEvent nodeUpdate2 = new NodeUpdateSchedulerEvent(node2,
- new LinkedList(), new LinkedList());
+ NodeUpdateSchedulerEvent nodeUpdate2 = new NodeUpdateSchedulerEvent(node2);
scheduler.handle(nodeUpdate2);
- NodeUpdateSchedulerEvent nodeUpdate3 = new NodeUpdateSchedulerEvent(node3,
- new LinkedList(), new LinkedList());
+ NodeUpdateSchedulerEvent nodeUpdate3 = new NodeUpdateSchedulerEvent(node3);
scheduler.handle(nodeUpdate3);
}
@@ -991,7 +981,7 @@ public class TestFairScheduler {
assertEquals(0, scheduler.applications.get(app6).getLiveContainers().size());
}
- @Test
+ @Test (timeout = 5000)
/**
* Tests the timing of decision to preempt tasks.
*/
@@ -1062,16 +1052,13 @@ public class TestFairScheduler {
// Sufficient node check-ins to fully schedule containers
for (int i = 0; i < 2; i++) {
- NodeUpdateSchedulerEvent nodeUpdate1 = new NodeUpdateSchedulerEvent(node1,
- new LinkedList(), new LinkedList());
+ NodeUpdateSchedulerEvent nodeUpdate1 = new NodeUpdateSchedulerEvent(node1);
scheduler.handle(nodeUpdate1);
- NodeUpdateSchedulerEvent nodeUpdate2 = new NodeUpdateSchedulerEvent(node2,
- new LinkedList(), new LinkedList());
+ NodeUpdateSchedulerEvent nodeUpdate2 = new NodeUpdateSchedulerEvent(node2);
scheduler.handle(nodeUpdate2);
- NodeUpdateSchedulerEvent nodeUpdate3 = new NodeUpdateSchedulerEvent(node3,
- new LinkedList(), new LinkedList());
+ NodeUpdateSchedulerEvent nodeUpdate3 = new NodeUpdateSchedulerEvent(node3);
scheduler.handle(nodeUpdate3);
}
@@ -1119,7 +1106,7 @@ public class TestFairScheduler {
Resources.createResource(1536), scheduler.resToPreempt(schedD, clock.getTime())));
}
- @Test
+ @Test (timeout = 5000)
public void testMultipleContainersWaitingForReservation() {
// Add a node
RMNode node1 = MockNodes.newNodeInfo(1, Resources.createResource(1024));
@@ -1129,8 +1116,7 @@ public class TestFairScheduler {
// Request full capacity of node
createSchedulingRequest(1024, "queue1", "user1", 1);
scheduler.update();
- NodeUpdateSchedulerEvent updateEvent = new NodeUpdateSchedulerEvent(node1,
- new ArrayList(), new ArrayList());
+ NodeUpdateSchedulerEvent updateEvent = new NodeUpdateSchedulerEvent(node1);
scheduler.handle(updateEvent);
ApplicationAttemptId attId1 = createSchedulingRequest(1024, "queue2", "user2", 1);
@@ -1146,7 +1132,7 @@ public class TestFairScheduler {
scheduler.applications.get(attId2).getCurrentReservation().getMemory());
}
- @Test
+ @Test (timeout = 5000)
public void testUserMaxRunningApps() throws Exception {
// Set max running apps
Configuration conf = createConfiguration();
@@ -1175,8 +1161,7 @@ public class TestFairScheduler {
"user1", 1);
scheduler.update();
- NodeUpdateSchedulerEvent updateEvent = new NodeUpdateSchedulerEvent(node1,
- new ArrayList(), new ArrayList());
+ NodeUpdateSchedulerEvent updateEvent = new NodeUpdateSchedulerEvent(node1);
scheduler.handle(updateEvent);
// App 1 should be running
@@ -1201,7 +1186,7 @@ public class TestFairScheduler {
assertEquals(2, scheduler.applications.get(attId1).getLiveContainers().size());
}
- @Test
+ @Test (timeout = 5000)
public void testReservationWhileMultiplePriorities() {
// Add a node
RMNode node1 = MockNodes.newNodeInfo(1, Resources.createResource(1024));
@@ -1211,8 +1196,7 @@ public class TestFairScheduler {
ApplicationAttemptId attId = createSchedulingRequest(1024, "queue1",
"user1", 1, 2);
scheduler.update();
- NodeUpdateSchedulerEvent updateEvent = new NodeUpdateSchedulerEvent(node1,
- new ArrayList(), new ArrayList());
+ NodeUpdateSchedulerEvent updateEvent = new NodeUpdateSchedulerEvent(node1);
scheduler.handle(updateEvent);
FSSchedulerApp app = scheduler.applications.get(attId);
@@ -1285,7 +1269,7 @@ public class TestFairScheduler {
assertNull("The application was allowed", app2);
}
- @Test
+ @Test (timeout = 5000)
public void testMultipleNodesSingleRackRequest() throws Exception {
RMNode node1 = MockNodes.newNodeInfo(1, Resources.createResource(1024));
RMNode node2 = MockNodes.newNodeInfo(1, Resources.createResource(1024));
@@ -1312,22 +1296,20 @@ public class TestFairScheduler {
// node 1 checks in
scheduler.update();
- NodeUpdateSchedulerEvent updateEvent1 = new NodeUpdateSchedulerEvent(node1,
- new ArrayList(), new ArrayList());
+ NodeUpdateSchedulerEvent updateEvent1 = new NodeUpdateSchedulerEvent(node1);
scheduler.handle(updateEvent1);
// should assign node local
assertEquals(1, scheduler.applications.get(appId).getLiveContainers().size());
// node 2 checks in
scheduler.update();
- NodeUpdateSchedulerEvent updateEvent2 = new NodeUpdateSchedulerEvent(node2,
- new ArrayList(), new ArrayList());
+ NodeUpdateSchedulerEvent updateEvent2 = new NodeUpdateSchedulerEvent(node2);
scheduler.handle(updateEvent2);
// should assign rack local
assertEquals(2, scheduler.applications.get(appId).getLiveContainers().size());
}
- @Test
+ @Test (timeout = 5000)
public void testFifoWithinQueue() throws Exception {
RMNode node1 = MockNodes.newNodeInfo(1, Resources.createResource(3072));
NodeAddedSchedulerEvent nodeEvent1 = new NodeAddedSchedulerEvent(node1);
@@ -1351,8 +1333,7 @@ public class TestFairScheduler {
// Because tests set assignmultiple to false, each heartbeat assigns a single
// container.
- NodeUpdateSchedulerEvent updateEvent = new NodeUpdateSchedulerEvent(node1,
- new ArrayList(), new ArrayList());
+ NodeUpdateSchedulerEvent updateEvent = new NodeUpdateSchedulerEvent(node1);
scheduler.handle(updateEvent);
assertEquals(1, app1.getLiveContainers().size());
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-tests/src/test/java/org/apache/hadoop/yarn/server/MiniYARNCluster.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-tests/src/test/java/org/apache/hadoop/yarn/server/MiniYARNCluster.java
index 1bb4dea0dde..452508ff4af 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-tests/src/test/java/org/apache/hadoop/yarn/server/MiniYARNCluster.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-tests/src/test/java/org/apache/hadoop/yarn/server/MiniYARNCluster.java
@@ -29,6 +29,8 @@ import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileContext;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.metrics2.lib.DefaultMetricsSystem;
+import org.apache.hadoop.util.Shell;
+import org.apache.hadoop.util.Shell.ShellCommandExecutor;
import org.apache.hadoop.yarn.YarnException;
import org.apache.hadoop.yarn.conf.YarnConfiguration;
import org.apache.hadoop.yarn.event.Dispatcher;
@@ -83,15 +85,51 @@ public class MiniYARNCluster extends CompositeService {
super(testName.replace("$", ""));
this.numLocalDirs = numLocalDirs;
this.numLogDirs = numLogDirs;
- this.testWorkDir = new File("target",
- testName.replace("$", ""));
+ String testSubDir = testName.replace("$", "");
+ File targetWorkDir = new File("target", testSubDir);
try {
FileContext.getLocalFSFileContext().delete(
- new Path(testWorkDir.getAbsolutePath()), true);
+ new Path(targetWorkDir.getAbsolutePath()), true);
} catch (Exception e) {
LOG.warn("COULD NOT CLEANUP", e);
throw new YarnException("could not cleanup test dir", e);
}
+
+ if (Shell.WINDOWS) {
+ // The test working directory can exceed the maximum path length supported
+ // by some Windows APIs and cmd.exe (260 characters). To work around this,
+ // create a symlink in temporary storage with a much shorter path,
+ // targeting the full path to the test working directory. Then, use the
+ // symlink as the test working directory.
+ String targetPath = targetWorkDir.getAbsolutePath();
+ File link = new File(System.getProperty("java.io.tmpdir"),
+ String.valueOf(System.currentTimeMillis()));
+ String linkPath = link.getAbsolutePath();
+
+ try {
+ FileContext.getLocalFSFileContext().delete(new Path(linkPath), true);
+ } catch (IOException e) {
+ throw new YarnException("could not cleanup symlink: " + linkPath, e);
+ }
+
+ // Guarantee target exists before creating symlink.
+ targetWorkDir.mkdirs();
+
+ ShellCommandExecutor shexec = new ShellCommandExecutor(
+ Shell.getSymlinkCommand(targetPath, linkPath));
+ try {
+ shexec.execute();
+ } catch (IOException e) {
+ throw new YarnException(String.format(
+ "failed to create symlink from %s to %s, shell output: %s", linkPath,
+ targetPath, shexec.getOutput()), e);
+ }
+
+ this.testWorkDir = link;
+ } else {
+ this.testWorkDir = targetWorkDir;
+ }
+
resourceManagerWrapper = new ResourceManagerWrapper();
addService(resourceManagerWrapper);
nodeManagers = new CustomNodeManager[noOfNodeManagers];
@@ -192,6 +230,19 @@ public class MiniYARNCluster extends CompositeService {
resourceManager.stop();
}
super.stop();
+
+ if (Shell.WINDOWS) {
+ // On Windows, clean up the short temporary symlink that was created to
+ // work around path length limitation.
+ String testWorkDirPath = testWorkDir.getAbsolutePath();
+ try {
+ FileContext.getLocalFSFileContext().delete(new Path(testWorkDirPath),
+ true);
+ } catch (IOException e) {
+ LOG.warn("could not cleanup symlink: " +
+ testWorkDir.getAbsolutePath());
+ }
+ }
}
}
@@ -220,7 +271,7 @@ public class MiniYARNCluster extends CompositeService {
for (int i = 0; i < numDirs; i++) {
dirs[i]= new File(testWorkDir, MiniYARNCluster.this.getName()
+ "-" + dirType + "Dir-nm-" + index + "_" + i);
- dirs[i].mkdir();
+ dirs[i].mkdirs();
LOG.info("Created " + dirType + "Dir in " + dirs[i].getAbsolutePath());
String delimiter = (i > 0) ? "," : "";
dirsString = dirsString.concat(delimiter + dirs[i].getAbsolutePath());
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-tests/src/test/resources/capacity-scheduler.xml b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-tests/src/test/resources/capacity-scheduler.xml
new file mode 100644
index 00000000000..0e4fe6b02e9
--- /dev/null
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-tests/src/test/resources/capacity-scheduler.xml
@@ -0,0 +1,111 @@
+
+
+
+
+ yarn.scheduler.capacity.maximum-applications
+ 10000
+
+ Maximum number of applications that can be pending and running.
+
+
+
+
+ yarn.scheduler.capacity.maximum-am-resource-percent
+ 0.1
+
+ Maximum percent of resources in the cluster which can be used to run
+ application masters i.e. controls number of concurrent running
+ applications.
+
+
+
+
+ yarn.scheduler.capacity.resource-calculator
+ org.apache.hadoop.yarn.server.resourcemanager.resource.DefaultResourceCalculator
+
+ The ResourceCalculator implementation to be used to compare
+ Resources in the scheduler.
+ The default i.e. DefaultResourceCalculator only uses Memory while
+ DominantResourceCalculator uses dominant-resource to compare
+ multi-dimensional resources such as Memory, CPU etc.
+
+
+
+
+ yarn.scheduler.capacity.root.queues
+ default
+
+ The queues at the this level (root is the root queue).
+
+
+
+
+ yarn.scheduler.capacity.root.default.capacity
+ 100
+ Default queue target capacity.
+
+
+
+ yarn.scheduler.capacity.root.default.user-limit-factor
+ 1
+
+ Default queue user limit a percentage from 0.0 to 1.0.
+
+
+
+
+ yarn.scheduler.capacity.root.default.maximum-capacity
+ 100
+
+ The maximum capacity of the default queue.
+
+
+
+
+ yarn.scheduler.capacity.root.default.state
+ RUNNING
+
+ The state of the default queue. State can be one of RUNNING or STOPPED.
+
+
+
+
+ yarn.scheduler.capacity.root.default.acl_submit_applications
+ *
+
+ The ACL of who can submit jobs to the default queue.
+
+
+
+
+ yarn.scheduler.capacity.root.default.acl_administer_queue
+ *
+
+ The ACL of who can administer jobs on the default queue.
+
+
+
+
+ yarn.scheduler.capacity.node-locality-delay
+ -1
+
+ Number of missed scheduling opportunities after which the CapacityScheduler
+ attempts to schedule rack-local containers.
+ Typically this should be set to number of racks in the cluster, this
+ feature is disabled by default, set to -1.
+
+
+
+
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/pom.xml b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/pom.xml
index 9ba8a9a25f3..47f7082ac9d 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/pom.xml
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/pom.xml
@@ -28,6 +28,27 @@
hadoop-yarn-server
pom
+
+
+
+ org.apache.maven.plugins
+ maven-surefire-plugin
+
+
+
+ ${basedir}/../../../../hadoop-common-project/hadoop-common/target
+
+
+
+ listener
+ org.apache.hadoop.test.TimedOutTestsListener
+
+
+
+
+
+
+
org.apache.hadoop
diff --git a/hadoop-yarn-project/hadoop-yarn/pom.xml b/hadoop-yarn-project/hadoop-yarn/pom.xml
index 5d261a2804d..738d6c5fcad 100644
--- a/hadoop-yarn-project/hadoop-yarn/pom.xml
+++ b/hadoop-yarn-project/hadoop-yarn/pom.xml
@@ -85,7 +85,7 @@
guice-servlet
- org.jboss.netty
+ io.netty
netty
@@ -159,6 +159,10 @@
org.apache.maven.plugins
maven-surefire-plugin
+
+
+ ${basedir}/../../../hadoop-common-project/hadoop-common/target
+
listener
diff --git a/hadoop-yarn-project/pom.xml b/hadoop-yarn-project/pom.xml
index 4366b387e13..610091182ba 100644
--- a/hadoop-yarn-project/pom.xml
+++ b/hadoop-yarn-project/pom.xml
@@ -59,7 +59,7 @@
ant
- org.jboss.netty
+ io.netty
netty
@@ -149,7 +149,7 @@
junit
- org.jboss.netty
+ io.netty
netty
@@ -180,15 +180,8 @@
-
- which cygpath 2> /dev/null
- if [ $? = 1 ]; then
- BUILD_DIR="${project.build.directory}"
- else
- BUILD_DIR=`cygpath --unix '${project.build.directory}'`
- fi
- cd $BUILD_DIR
- tar czf ${project.artifactId}-${project.version}.tar.gz ${project.artifactId}-${project.version}
+ cd "${project.build.directory}"
+ tar cf - ${project.artifactId}-${project.version} | gzip > ${project.artifactId}-${project.version}.tar.gz
|