From af3e3a84ac7038e7d73d20d32382bb66cbcca0fa Mon Sep 17 00:00:00 2001
From: root <root@localhost.localdomain>
Date: Tue, 31 Dec 2019 17:29:51 +0800
Subject: [PATCH] Hadoop2.7.3 NUMA Aware

---
 .../apache/hadoop/yarn/conf/YarnConfiguration.java |  33 +++
 .../src/main/resources/yarn-default.xml            |  43 ++++
 .../yarn/server/nodemanager/ContainerExecutor.java |  17 ++
 .../nodemanager/DefaultContainerExecutor.java      |  21 +-
 .../server/nodemanager/LinuxContainerExecutor.java |  20 ++
 .../yarn/server/nodemanager/NodeManager.java       |   9 +
 .../containermanager/ContainerManagerImpl.java     |  13 +-
 .../launcher/ContainersLauncher.java               |  12 +-
 .../nodemanager/numa/AssignedNumaNodeInfo.java     |  39 +++
 .../nodemanager/numa/NUMAResourcesManager.java     | 280 +++++++++++++++++++++
 .../yarn/server/nodemanager/numa/NumaNode.java     | 154 ++++++++++++
 .../yarn/server/nodemanager/numa/package-info.java |  28 +++
 .../nodemanager/numa/TestNUMAResourcesManager.java | 211 ++++++++++++++++
 13 files changed, 873 insertions(+), 7 deletions(-)
 create mode 100644 hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/numa/AssignedNumaNodeInfo.java
 create mode 100644 hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/numa/NUMAResourcesManager.java
 create mode 100644 hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/numa/NumaNode.java
 create mode 100644 hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/numa/package-info.java
 create mode 100644 hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/test/java/org/apache/hadoop/yarn/server/nodemanager/numa/TestNUMAResourcesManager.java

diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/conf/YarnConfiguration.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/conf/YarnConfiguration.java
index 8922e06..66e80f9 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/conf/YarnConfiguration.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/conf/YarnConfiguration.java
@@ -1820,6 +1820,26 @@ private static void addDeprecatedKeys() {
       + "enabled";
   public static final boolean DEFAULT_NODE_LABELS_ENABLED = false;
 
+  // NUMA awareness props
+  public static final String NM_NUMA_AWARENESS_ENABLED = NM_PREFIX
+      + "numa-awareness.enabled";
+  public static final boolean DEFAULT_NM_NUMA_AWARENESS_ENABLED = false;
+  public static final String NM_NUMA_AWARENESS_READ_TOPOLOGY = NM_PREFIX
+      + "numa-awareness.read-topology";
+  public static final boolean DEFAULT_NM_NUMA_AWARENESS_READ_TOPOLOGY = false;
+  public static final String NM_NUMA_AWARENESS_NODE_IDS = NM_PREFIX
+      + "numa-awareness.node-ids";
+  public static final String NM_NUMA_AWARENESS_NODE_MEMORY = NM_PREFIX
+      + "numa-awareness.<NODE_ID>.memory";
+  public static final String NM_NUMA_AWARENESS_NODE_CPUS = NM_PREFIX
+      + "numa-awareness.<NODE_ID>.cpus";
+  public static final String NM_NUMA_AWARENESS_NUMACTL_CMD = NM_PREFIX
+      + "numa-awareness.numactl.cmd";
+  public static final String DEFAULT_NM_NUMA_AWARENESS_NUMACTL_CMD =
+      "/usr/bin/numactl";
+
+
+
   public YarnConfiguration() {
     super();
   }
@@ -1914,4 +1934,17 @@ public static String getClusterId(Configuration conf) {
     }
     return clusterId;
   }
+
+  /**
+   * Returns whether the NUMA awareness is enabled.
+   *
+   * @param conf the configuration
+   * @return whether the NUMA awareness is enabled.
+   */
+  public static boolean numaAwarenessEnabled(Configuration conf) {
+    return conf.getBoolean(NM_NUMA_AWARENESS_ENABLED,
+        DEFAULT_NM_NUMA_AWARENESS_ENABLED);
+  }
+
+
 }
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/resources/yarn-default.xml b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/resources/yarn-default.xml
index 12e42f8..098410d 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/resources/yarn-default.xml
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/resources/yarn-default.xml
@@ -1710,4 +1710,47 @@
     <name>yarn.nodemanager.webapp.cross-origin.enabled</name>
     <value>false</value>
   </property>
+  <property>
+    <description>
+    Whether to enable the NUMA awareness for containers in Node Manager.
+    </description>
+    <name>yarn.nodemanager.numa-awareness.enabled</name>
+    <value>false</value>
+  </property>
+
+  <property>
+    <description>
+    Whether to read the NUMA topology from the system or from the
+    configurations. If the value is true then NM reads the NUMA topology from
+    system using the command 'numactl --hardware'. If the value is false then NM
+    reads the topology from the configurations
+    'yarn.nodemanager.numa-awareness.node-ids'(for node id's),
+    'yarn.nodemanager.numa-awareness.&lt;NODE_ID&gt;.memory'(for each node memory),
+    'yarn.nodemanager.numa-awareness.&lt;NODE_ID&gt;.cpus'(for each node cpus).
+    </description>
+    <name>yarn.nodemanager.numa-awareness.read-topology</name>
+    <value>false</value>
+  </property>
+
+  <property>
+    <description>
+    NUMA node id's in the form of comma separated list. Memory and No of CPUs
+    will be read using the properties
+    'yarn.nodemanager.numa-awareness.&lt;NODE_ID&gt;.memory' and
+    'yarn.nodemanager.numa-awareness.&lt;NODE_ID&gt;.cpus' for each id specified
+    in this value. This property value will be read only when
+    'yarn.nodemanager.numa-awareness.read-topology=false'.
+    </description>
+    <name>yarn.nodemanager.numa-awareness.node-ids</name>
+    <value></value>
+  </property>
+
+  <property>
+    <description>
+    The numactl command path which controls NUMA policy for processes or
+    shared memory.
+    </description>
+    <name>yarn.nodemanager.numa-awareness.numactl.cmd</name>
+    <value>/usr/bin/numactl</value>
+  </property>
 </configuration>
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 377fd1d2..777a4cb 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
@@ -46,6 +46,7 @@
 import org.apache.hadoop.yarn.server.nodemanager.containermanager.container.Container;
 import org.apache.hadoop.yarn.server.nodemanager.containermanager.container.ContainerDiagnosticsUpdateEvent;
 import org.apache.hadoop.yarn.server.nodemanager.containermanager.launcher.ContainerLaunch;
+import org.apache.hadoop.yarn.server.nodemanager.numa.NUMAResourcesManager;
 import org.apache.hadoop.yarn.server.nodemanager.util.ProcessIdFileReader;
 import org.apache.hadoop.util.Shell;
 import org.apache.hadoop.util.StringUtils;
@@ -64,6 +65,7 @@
   private ReentrantReadWriteLock lock = new ReentrantReadWriteLock();
   private final ReadLock readLock = lock.readLock();
   private final WriteLock writeLock = lock.writeLock();
+  private NUMAResourcesManager numaResourcesManager;
 
   @Override
   public void setConf(Configuration conf) {
@@ -449,6 +451,21 @@ public String getProcessId(ContainerId containerID) {
     return pid;
   }
 
+/**
+ * NUMA Resources Manager for assigning numa nodes for a container.
+ *
+ * @param numaResourcesManager instance of the NUMAResourcesManager
+ */
+  public void setNumaResourcesManager(
+      NUMAResourcesManager numaResourcesManager) {
+    this.numaResourcesManager = numaResourcesManager;
+  }
+
+  public NUMAResourcesManager getNumaResourcesManager() {
+    return numaResourcesManager;
+  }
+
+
   public static class DelayedProcessKiller extends Thread {
     private Container container;
     private final String user;
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 7cb15cf..702dd0d 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
@@ -43,6 +43,7 @@
 import org.apache.hadoop.fs.UnsupportedFileSystemException;
 import org.apache.hadoop.fs.permission.FsPermission;
 import org.apache.hadoop.io.IOUtils;
+import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.util.Shell;
 import org.apache.hadoop.util.Shell.ExitCodeException;
 import org.apache.hadoop.util.Shell.CommandExecutor;
@@ -56,6 +57,7 @@
 import org.apache.hadoop.yarn.server.nodemanager.containermanager.container.ContainerDiagnosticsUpdateEvent;
 import org.apache.hadoop.yarn.server.nodemanager.containermanager.launcher.ContainerLaunch;
 import org.apache.hadoop.yarn.server.nodemanager.containermanager.localizer.ContainerLocalizer;
+import org.apache.hadoop.yarn.server.nodemanager.numa.AssignedNumaNodeInfo;
 import org.apache.hadoop.yarn.util.ConverterUtils;
 
 import com.google.common.annotations.VisibleForTesting;
@@ -261,9 +263,22 @@ protected CommandExecutor buildCommandExecutor(String wrapperScriptPath,
       String containerIdStr, String user, Path pidFile, Resource resource,
       File wordDir, Map<String, String> environment)
           throws IOException {
-    
+    Configuration conf = this.getConf(); 
     String[] command = getRunCommand(wrapperScriptPath,
-        containerIdStr, user, pidFile, this.getConf(), resource);
+        containerIdStr, user, pidFile, conf, resource);
+    if (YarnConfiguration.numaAwarenessEnabled(conf)) {
+      String[] cmd = new String[command.length + 3];
+      AssignedNumaNodeInfo numaNodeInfo = getNumaResourcesManager()
+          .assignNumaNode(ContainerId.fromString(containerIdStr), resource);
+      if (numaNodeInfo != null) {
+        cmd[0] = conf.get(YarnConfiguration.NM_NUMA_AWARENESS_NUMACTL_CMD,
+            YarnConfiguration.DEFAULT_NM_NUMA_AWARENESS_NUMACTL_CMD);
+        cmd[1] = "--interleave=" + numaNodeInfo.getMemNode();
+        cmd[2] = "--cpunodebind=" + numaNodeInfo.getCpuNode();
+        System.arraycopy(command, 0, cmd, 3, command.length);
+        command = cmd;
+      }
+    }
 
       LOG.info("launchContainer: " + Arrays.toString(command));
       return new ShellCommandExecutor(
@@ -356,7 +371,7 @@ private void writeSessionScript(Path launchDst, Path pidFile)
         pout.println();
         pout.println("echo $$ > " + pidFile.toString() + ".tmp");
         pout.println("/bin/mv -f " + pidFile.toString() + ".tmp " + pidFile);
-        String exec = Shell.isSetsidAvailable? "exec setsid" : "exec";
+        String exec = Shell.isSetsidAvailable? "exec" : "exec";
         pout.println(exec + " /bin/bash \"" +
             launchDst.toUri().getPath().toString() + "\"");
       } finally {
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/LinuxContainerExecutor.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/LinuxContainerExecutor.java
index e194c95..91138f5 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/LinuxContainerExecutor.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/LinuxContainerExecutor.java
@@ -47,6 +47,7 @@
 import org.apache.hadoop.yarn.server.nodemanager.util.DefaultLCEResourcesHandler;
 import org.apache.hadoop.yarn.server.nodemanager.util.LCEResourcesHandler;
 import org.apache.hadoop.yarn.util.ConverterUtils;
+import org.apache.hadoop.yarn.server.nodemanager.numa.AssignedNumaNodeInfo;
 
 public class LinuxContainerExecutor extends ContainerExecutor {
 
@@ -169,6 +170,24 @@ protected void addSchedPriorityCommand(List<String> command) {
     } 
   }
 
+  private void addNumaAwarenessParams(List<String> command,
+      Container container) {
+    Configuration conf = getConf();
+    if (YarnConfiguration.numaAwarenessEnabled(conf)) {
+      AssignedNumaNodeInfo numaNodeInfo = getNumaResourcesManager()
+          .assignNumaNode(container.getContainerId(), container.getResource());
+      if (numaNodeInfo != null) {
+        String numactlCmd = conf.get(
+            YarnConfiguration.NM_NUMA_AWARENESS_NUMACTL_CMD,
+            YarnConfiguration.DEFAULT_NM_NUMA_AWARENESS_NUMACTL_CMD);
+        command.addAll(Arrays.asList(numactlCmd,
+            "--interleave=" + numaNodeInfo.getMemNode(),
+            "--cpunodebind=" + numaNodeInfo.getCpuNode()));
+      }
+    }
+  }
+
+
   @Override 
   public void init() throws IOException {        
     // Send command to executor which will just start up, 
@@ -278,6 +297,7 @@ public int launchContainer(Container container,
       Path pidFilePath = getPidFilePath(containerId);
       if (pidFilePath != null) {
         List<String> command = new ArrayList<String>();
+	addNumaAwarenessParams(command, container);
         addSchedPriorityCommand(command);
         command.addAll(Arrays.asList(
             containerExecutorExe, runAsUser, user, Integer
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/NodeManager.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/NodeManager.java
index 3df0242..17c0eca 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/NodeManager.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/NodeManager.java
@@ -64,6 +64,7 @@
 import org.apache.hadoop.yarn.server.nodemanager.security.NMContainerTokenSecretManager;
 import org.apache.hadoop.yarn.server.nodemanager.security.NMTokenSecretManagerInNM;
 import org.apache.hadoop.yarn.server.nodemanager.webapp.WebServer;
+import org.apache.hadoop.yarn.server.nodemanager.numa.NUMAResourcesManager;
 import org.apache.hadoop.yarn.server.security.ApplicationACLsManager;
 
 import com.google.common.annotations.VisibleForTesting;
@@ -247,6 +248,14 @@ protected void serviceInit(Configuration conf) throws Exception {
     addService(webServer);
     ((NMContext) context).setWebServer(webServer);
 
+   if (YarnConfiguration.numaAwarenessEnabled(conf)) {
+      NUMAResourcesManager numaNodesScheduler = new NUMAResourcesManager();
+      addService(numaNodesScheduler);
+      containerManager.setNumaNodesScheduler(numaNodesScheduler);
+      exec.setNumaResourcesManager(numaNodesScheduler);
+    }
+
+
     dispatcher.register(ContainerManagerEventType.class, containerManager);
     dispatcher.register(NodeManagerEventType.class, this);
     addService(dispatcher);
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/ContainerManagerImpl.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/ContainerManagerImpl.java
index 0c8ed05..46cbd9d 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/ContainerManagerImpl.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/ContainerManagerImpl.java
@@ -139,7 +139,7 @@
 
 import com.google.common.annotations.VisibleForTesting;
 import com.google.protobuf.ByteString;
-
+import org.apache.hadoop.yarn.server.nodemanager.numa.NUMAResourcesManager;
 public class ContainerManagerImpl extends CompositeService implements
     ServiceStateChangeListener, ContainerManagementProtocol,
     EventHandler<ContainerManagerEvent> {
@@ -170,7 +170,7 @@
   private boolean serviceStopped = false;
   private final ReadLock readLock;
   private final WriteLock writeLock;
-
+  private NUMAResourcesManager numaNodesScheduler;
   private long waitForContainersOnShutdownMillis;
 
   public ContainerManagerImpl(Context context, ContainerExecutor exec,
@@ -246,6 +246,15 @@ public void serviceInit(Configuration conf) throws Exception {
     recover();
   }
 
+
+  public NUMAResourcesManager getNumaNodesScheduler() {
+    return numaNodesScheduler;
+  }
+
+  public void setNumaNodesScheduler(NUMAResourcesManager numaNodesScheduler) {
+    this.numaNodesScheduler = numaNodesScheduler;
+  }
+
   @SuppressWarnings("unchecked")
   private void recover() throws IOException, URISyntaxException {
     NMStateStoreService stateStore = context.getNMStateStore();
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/ContainersLauncher.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/ContainersLauncher.java
index 6950aa9..cd01a2b 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/ContainersLauncher.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/ContainersLauncher.java
@@ -42,7 +42,7 @@
 import org.apache.hadoop.yarn.server.nodemanager.containermanager.application.Application;
 import org.apache.hadoop.yarn.server.nodemanager.containermanager.container.Container;
 import org.apache.hadoop.yarn.server.nodemanager.containermanager.localizer.ResourceLocalizationService;
-
+import org.apache.hadoop.yarn.conf.YarnConfiguration;
 import com.google.common.annotations.VisibleForTesting;
 import com.google.common.util.concurrent.ThreadFactoryBuilder;
 
@@ -61,7 +61,7 @@
   private final ContainerExecutor exec;
   private final Dispatcher dispatcher;
   private final ContainerManagerImpl containerManager;
-
+  private boolean numaAwarenessEnabled;
   private LocalDirsHandlerService dirsHandler;
   @VisibleForTesting
   public ExecutorService containerLauncher =
@@ -92,6 +92,7 @@ protected void serviceInit(Configuration conf) throws Exception {
     } catch (UnsupportedFileSystemException e) {
       throw new YarnRuntimeException("Failed to start ContainersLauncher", e);
     }
+    numaAwarenessEnabled = YarnConfiguration.numaAwarenessEnabled(conf);
     super.serviceInit(conf);
   }
 
@@ -128,6 +129,13 @@ public void handle(ContainersLauncherEvent event) {
         break;
       case CLEANUP_CONTAINER:
         ContainerLaunch launcher = running.remove(containerId);
+        // Remove from NUMA Resources Manager if enabled
+        if (numaAwarenessEnabled
+            && containerManager.getNumaNodesScheduler() != null) {
+          containerManager.getNumaNodesScheduler()
+              .releaseNumaResources(containerId);
+        }
+
         if (launcher == null) {
           // Container not launched. So nothing needs to be done.
           return;
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/numa/AssignedNumaNodeInfo.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/numa/AssignedNumaNodeInfo.java
new file mode 100644
index 0000000..6a35432
--- /dev/null
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/numa/AssignedNumaNodeInfo.java
@@ -0,0 +1,39 @@
+/**
+ * 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.nodemanager.numa;
+
+/**
+ * AssignedNumaNodeInfo contains Memory nodes and CPU nodes.
+ */
+public class AssignedNumaNodeInfo {
+  private String memNodes;
+  private String cpuNodes;
+
+  public AssignedNumaNodeInfo(String memNodes, String cpuNodes) {
+    this.memNodes = memNodes;
+    this.cpuNodes = cpuNodes;
+  }
+
+  public String getMemNode() {
+    return memNodes;
+  }
+
+  public String getCpuNode() {
+    return cpuNodes;
+  }
+}
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/numa/NUMAResourcesManager.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/numa/NUMAResourcesManager.java
new file mode 100644
index 0000000..c7187fd
--- /dev/null
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/numa/NUMAResourcesManager.java
@@ -0,0 +1,280 @@
+/**
+ * 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.nodemanager.numa;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.List;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.service.CompositeService;
+import org.apache.hadoop.util.Shell.ShellCommandExecutor;
+import org.apache.hadoop.yarn.api.records.ContainerId;
+import org.apache.hadoop.yarn.api.records.Resource;
+import org.apache.hadoop.yarn.conf.YarnConfiguration;
+import org.apache.hadoop.yarn.exceptions.YarnException;
+
+import com.google.common.annotations.VisibleForTesting;
+
+/**
+ * NUMA Resources Manager reads the NUMA topology and assigns NUMA nodes to the
+ * containers.
+ */
+public class NUMAResourcesManager extends CompositeService {
+
+  private static final Log LOG = LogFactory.getLog(NUMAResourcesManager.class);
+
+  // Regex to find node ids, Ex: 'available: 2 nodes (0-1)'
+  private static final String NUMA_NODEIDS_REGEX =
+      "available:\\s*[0-9]+\\s*nodes\\s*\\(([0-9\\-,]*)\\)";
+
+  // Regex to find node memory, Ex: 'node 0 size: 73717 MB'
+  private static final String NUMA_NODE_MEMORY_REGEX =
+      "node\\s*<NUMA-NODE>\\s*size:\\s*([0-9]+)\\s*([KMG]B)";
+
+  // Regex to find node cpus, Ex: 'node 0 cpus: 0 2 4 6'
+  private static final String NUMA_NODE_CPUS_REGEX =
+      "node\\s*<NUMA-NODE>\\s*cpus:\\s*([0-9\\s]+)";
+
+  private static final String GB = "GB";
+  private static final String KB = "KB";
+  private static final String NUMA_NODE = "<NUMA-NODE>";
+  private static final String SPACE = "\\s";
+  private static final long DEFAULT_NUMA_NODE_MEMORY = 1024;
+  private static final int DEFAULT_NUMA_NODE_CPUS = 1;
+
+  private List<NumaNode> numaNodesList = new ArrayList<>();
+  private int noOfNumaNodes;
+  private int currentAssignNode;
+
+  public NUMAResourcesManager() {
+    super(NUMAResourcesManager.class.getName());
+  }
+
+  @Override
+  public synchronized void serviceInit(Configuration conf) throws Exception {
+    super.serviceInit(conf);
+    if (conf.getBoolean(YarnConfiguration.NM_NUMA_AWARENESS_READ_TOPOLOGY,
+        YarnConfiguration.DEFAULT_NM_NUMA_AWARENESS_READ_TOPOLOGY)) {
+      LOG.info("Reading NUMA topology using 'numactl --hardware' command.");
+      String cmdOutput = executeNGetCmdOutput(conf);
+      String[] outputLines = cmdOutput.split("\\n");
+      Pattern pattern = Pattern.compile(NUMA_NODEIDS_REGEX);
+      String nodeIdsStr = null;
+      for (String line : outputLines) {
+        Matcher matcher = pattern.matcher(line);
+        if (matcher.find()) {
+          nodeIdsStr = matcher.group(1);
+          break;
+        }
+      }
+      if (nodeIdsStr == null) {
+        throw new YarnException("Failed to get numa nodes from"
+            + " 'numactl --hardware' output and output is:\n" + cmdOutput);
+      }
+      String[] nodeIdCommaSplits = nodeIdsStr.split("[,\\s]");
+      for (String nodeIdOrRange : nodeIdCommaSplits) {
+        if (nodeIdOrRange.contains("-")) {
+          String[] beginNEnd = nodeIdOrRange.split("-");
+          for (int nodeId = Integer.parseInt(beginNEnd[0]); nodeId <= Integer
+              .parseInt(beginNEnd[1]); nodeId++) {
+            long memory = parseMemory(outputLines, String.valueOf(nodeId));
+            int cpus = parseCpus(outputLines, String.valueOf(nodeId));
+            numaNodesList
+                .add(new NumaNode(String.valueOf(nodeId), memory, cpus));
+          }
+        } else {
+          long memory = parseMemory(outputLines, nodeIdOrRange);
+          int cpus = parseCpus(outputLines, nodeIdOrRange);
+          numaNodesList.add(new NumaNode(nodeIdOrRange, memory, cpus));
+        }
+      }
+    } else {
+      LOG.info("Reading NUMA topology using configurations.");
+      Collection<String> nodeIds = conf
+          .getStringCollection(YarnConfiguration.NM_NUMA_AWARENESS_NODE_IDS);
+      for (String nodeId : nodeIds) {
+        long mem = conf.getLong(
+            "yarn.nodemanager.numa-awareness." + nodeId + ".memory",
+            DEFAULT_NUMA_NODE_MEMORY);
+        int cpus = conf.getInt(
+            "yarn.nodemanager.numa-awareness." + nodeId + ".cpus",
+            DEFAULT_NUMA_NODE_CPUS);
+        numaNodesList.add(new NumaNode(nodeId, mem, cpus));
+      }
+    }
+    if (numaNodesList.isEmpty()) {
+      throw new YarnException("There are no available NUMA nodes"
+          + " for making containers NUMA aware.");
+    }
+    noOfNumaNodes = numaNodesList.size();
+    LOG.info("Available numa nodes with capacities : " + numaNodesList);
+  }
+
+  @VisibleForTesting
+  String executeNGetCmdOutput(Configuration conf) throws YarnException {
+    String numaCtlCmd = conf.get(
+        YarnConfiguration.NM_NUMA_AWARENESS_NUMACTL_CMD,
+        YarnConfiguration.DEFAULT_NM_NUMA_AWARENESS_NUMACTL_CMD);
+    String[] args = new String[] {numaCtlCmd, "--hardware"};
+    ShellCommandExecutor shExec = new ShellCommandExecutor(args);
+    try {
+      shExec.execute();
+    } catch (IOException e) {
+      throw new YarnException("Failed to read the numa configurations.", e);
+    }
+    String cmdOutput = shExec.getOutput();
+    return cmdOutput;
+  }
+
+  private int parseCpus(String[] outputLines, String nodeId) {
+    int cpus = 0;
+    Pattern patternNodeCPUs = Pattern
+        .compile(NUMA_NODE_CPUS_REGEX.replace(NUMA_NODE, nodeId));
+    for (String line : outputLines) {
+      Matcher matcherNodeCPUs = patternNodeCPUs.matcher(line);
+      if (matcherNodeCPUs.find()) {
+        String cpusStr = matcherNodeCPUs.group(1);
+        cpus = cpusStr.split(SPACE).length;
+        break;
+      }
+    }
+    return cpus;
+  }
+
+  private long parseMemory(String[] outputLines, String nodeId)
+      throws YarnException {
+    long memory = 0;
+    String units;
+    Pattern patternNodeMem = Pattern
+        .compile(NUMA_NODE_MEMORY_REGEX.replace(NUMA_NODE, nodeId));
+    for (String line : outputLines) {
+      Matcher matcherNodeMem = patternNodeMem.matcher(line);
+      if (matcherNodeMem.find()) {
+        try {
+          memory = Long.parseLong(matcherNodeMem.group(1));
+          units = matcherNodeMem.group(2);
+          if (GB.equals(units)) {
+            memory = memory * 1024;
+          } else if (KB.equals(units)) {
+            memory = memory / 1024;
+          }
+        } catch (Exception ex) {
+          throw new YarnException("Failed to get memory for node:" + nodeId,
+              ex);
+        }
+        break;
+      }
+    }
+    return memory;
+  }
+
+  /**
+   * Assigns the available NUMA nodes for the requested containerId with
+   * resource in a round robin fashion.
+   *
+   * @param containerId the container ID
+   * @param resource resource for the Numa Node
+   * @return the assigned NUMA Node info or null if resources not available.
+   */
+  public synchronized AssignedNumaNodeInfo assignNumaNode(
+      ContainerId containerId, Resource resource) {
+    for (int index = 0; index < noOfNumaNodes; index++) {
+      NumaNode numaNode = numaNodesList
+          .get((currentAssignNode + index) % noOfNumaNodes);
+      if (numaNode.isResourcesAvailable(resource)) {
+        numaNode.assignResources(resource, containerId);
+        LOG.info("Assigning NUMA node " + numaNode.getNodeId() + " for memory, "
+            + numaNode.getNodeId() + " for cpus for the " + containerId);
+        currentAssignNode = (currentAssignNode + index + 1) % noOfNumaNodes;
+        return new AssignedNumaNodeInfo(numaNode.getNodeId(),
+            numaNode.getNodeId());
+      }
+    }
+
+    // If there is no single node matched for the container resource
+    // Check the NUMA nodes for Memory resources
+    StringBuilder memNodes = new StringBuilder();
+   // long memreq = resource.getMemorySize();
+    long memreq = resource.getMemory();
+    for (NumaNode numaNode : numaNodesList) {
+      memreq = numaNode.assignAvailableMemory(memreq, containerId);
+      memNodes.append(numaNode.getNodeId());
+      if (memreq == 0) {
+        break;
+      } else {
+        memNodes.append(',');
+      }
+    }
+    if (memreq != 0) {
+      LOG.info("There is no available memory:" + resource.getMemory()
+          + " in numa nodes for " + containerId);
+      releaseNumaResources(containerId);
+      return null;
+    }
+
+    // Check the NUMA nodes for CPU resources
+    StringBuilder cpuNodes = new StringBuilder();
+    int cpusreq = resource.getVirtualCores();
+    for (int index = 0; index < noOfNumaNodes; index++) {
+      NumaNode numaNode = numaNodesList
+          .get((currentAssignNode + index) % noOfNumaNodes);
+      cpusreq = numaNode.assignAvailableCpus(cpusreq, containerId);
+      cpuNodes.append(numaNode.getNodeId());
+      if (cpusreq == 0) {
+        currentAssignNode = (currentAssignNode + index + 1) % noOfNumaNodes;
+        break;
+      } else {
+        cpuNodes.append(',');
+      }
+    }
+
+    if (cpusreq != 0) {
+      LOG.info("There are no available cpus:" + resource.getVirtualCores()
+          + " in numa nodes for " + containerId);
+      releaseNumaResources(containerId);
+      return null;
+    }
+    LOG.info("Assigning multiple NUMA nodes (" + memNodes.toString()
+        + ") for memory, (" + cpuNodes.toString() + ") for cpus for "
+        + containerId);
+    return new AssignedNumaNodeInfo(memNodes.toString(), cpuNodes.toString());
+  }
+
+  /**
+   * Release assigned NUMA resources for the container.
+   *
+   * @param containerId the container ID
+   */
+  public synchronized void releaseNumaResources(ContainerId containerId) {
+    LOG.info("Releasing the assigned NUMA resources for " + containerId);
+    for (NumaNode numaNode : numaNodesList) {
+      numaNode.releaseResources(containerId);
+    }
+  }
+
+  @VisibleForTesting
+  Collection<NumaNode> getNumaNodesList() {
+    return numaNodesList;
+  }
+}
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/numa/NumaNode.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/numa/NumaNode.java
new file mode 100644
index 0000000..e610b6f
--- /dev/null
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/numa/NumaNode.java
@@ -0,0 +1,154 @@
+/**
+ * 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.nodemanager.numa;
+
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.yarn.api.records.ContainerId;
+import org.apache.hadoop.yarn.api.records.Resource;
+
+/**
+ * NumaNode class holds the NUMA node topology with the total and used
+ * resources.
+ */
+public class NumaNode {
+  private String nodeId;
+  private long totalMemory;
+  private int totalCpus;
+  private long usedMemory;
+  private int usedCpus;
+
+  private static final Log LOG = LogFactory.getLog(NumaNode.class);
+
+  private Map<ContainerId, Long> containerVsMemUsage =
+      new ConcurrentHashMap<>();
+  private Map<ContainerId, Integer> containerVsCpusUsage =
+      new ConcurrentHashMap<>();
+
+  public NumaNode(String nodeId, long totalMemory, int totalCpus) {
+    this.nodeId = nodeId;
+    this.totalMemory = totalMemory;
+    this.totalCpus = totalCpus;
+  }
+
+  public synchronized boolean isResourcesAvailable(Resource resource) {
+    LOG.debug(
+        "Memory available:" + (totalMemory - usedMemory) + ", CPUs available:"
+            + (totalCpus - usedCpus) + ", requested:" + resource);
+    if ((totalMemory - usedMemory) >= resource.getMemory()
+        && (totalCpus - usedCpus) >= resource.getVirtualCores()) {
+      return true;
+    }
+    return false;
+  }
+
+  public synchronized long assignAvailableMemory(long memreq,
+      ContainerId containerId) {
+    long memAvailable = totalMemory - usedMemory;
+    if (memAvailable >= memreq) {
+      containerVsMemUsage.put(containerId, memreq);
+      usedMemory += memreq;
+      return 0;
+    } else {
+      usedMemory += memAvailable;
+      containerVsMemUsage.put(containerId, memAvailable);
+      return memreq - memAvailable;
+    }
+  }
+
+  public synchronized int assignAvailableCpus(int cpusreq,
+      ContainerId containerId) {
+    int cpusAvailable = totalCpus - usedCpus;
+    if (cpusAvailable >= cpusreq) {
+      containerVsCpusUsage.put(containerId, cpusreq);
+      usedCpus += cpusreq;
+      return 0;
+    } else {
+      usedCpus += cpusAvailable;
+      containerVsCpusUsage.put(containerId, cpusAvailable);
+      return cpusreq - cpusAvailable;
+    }
+  }
+
+  public synchronized void assignResources(Resource resource,
+      ContainerId containerId) {
+    containerVsMemUsage.put(containerId, (long)resource.getMemory());
+    containerVsCpusUsage.put(containerId, resource.getVirtualCores());
+    usedMemory += resource.getMemory();
+    usedCpus += resource.getVirtualCores();
+  }
+
+  public synchronized void releaseResources(ContainerId containerId) {
+    if (containerVsMemUsage.containsKey(containerId)) {
+      usedMemory -= containerVsMemUsage.get(containerId);
+      containerVsMemUsage.remove(containerId);
+    }
+    if (containerVsCpusUsage.containsKey(containerId)) {
+      usedCpus -= containerVsCpusUsage.get(containerId);
+      containerVsCpusUsage.remove(containerId);
+    }
+  }
+
+  @Override
+  public String toString() {
+    return "Node Id:" + nodeId + "\tMemory:" + totalMemory + "\tCPus:"
+        + totalCpus;
+  }
+
+  @Override
+  public int hashCode() {
+    final int prime = 31;
+    int result = 1;
+    result = prime * result + ((nodeId == null) ? 0 : nodeId.hashCode());
+    result = prime * result + (int) (totalMemory ^ (totalMemory >>> 32));
+    result = prime * result + totalCpus;
+    return result;
+  }
+
+  @Override
+  public boolean equals(Object obj) {
+    if (this == obj) {
+      return true;
+    }
+    if (obj == null || getClass() != obj.getClass()) {
+      return false;
+    }
+    NumaNode other = (NumaNode) obj;
+    if (nodeId == null) {
+      if (other.nodeId != null) {
+        return false;
+      }
+    } else if (!nodeId.equals(other.nodeId)) {
+      return false;
+    }
+    if (totalMemory != other.totalMemory) {
+      return false;
+    }
+    if (totalCpus != other.totalCpus) {
+      return false;
+    }
+    return true;
+  }
+
+  public String getNodeId() {
+    return nodeId;
+  }
+}
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/numa/package-info.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/numa/package-info.java
new file mode 100644
index 0000000..bbbafa3
--- /dev/null
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/numa/package-info.java
@@ -0,0 +1,28 @@
+/*
+ * 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.nodemanager.numa contains classes
+ * related to NUMA aware containers functionality.
+ */
+@InterfaceAudience.Private
+@InterfaceStability.Unstable
+package org.apache.hadoop.yarn.server.nodemanager.numa;
+
+import org.apache.hadoop.classification.InterfaceAudience;
+import org.apache.hadoop.classification.InterfaceStability;
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/test/java/org/apache/hadoop/yarn/server/nodemanager/numa/TestNUMAResourcesManager.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/test/java/org/apache/hadoop/yarn/server/nodemanager/numa/TestNUMAResourcesManager.java
new file mode 100644
index 0000000..4a26a8b
--- /dev/null
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/test/java/org/apache/hadoop/yarn/server/nodemanager/numa/TestNUMAResourcesManager.java
@@ -0,0 +1,211 @@
+/**
+ * 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.nodemanager.numa;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collection;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.yarn.api.records.ContainerId;
+import org.apache.hadoop.yarn.api.records.Resource;
+import org.apache.hadoop.yarn.conf.YarnConfiguration;
+import org.apache.hadoop.yarn.exceptions.YarnRuntimeException;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+
+/**
+ * Test class for NUMAResourcesManager.
+ */
+public class TestNUMAResourcesManager {
+
+  private Configuration conf;
+  private NUMAResourcesManager numaResourcesManager;
+
+  @Before
+  public void setUp() throws IOException {
+    conf = new YarnConfiguration();
+    numaResourcesManager = new NUMAResourcesManager();
+  }
+
+  @Test
+  public void testReadNumaTopologyFromConfigurations() throws Exception {
+    setNumaTopologyConfigs();
+    numaResourcesManager.serviceInit(conf);
+    Collection<NumaNode> nodesList = numaResourcesManager.getNumaNodesList();
+    Collection<NumaNode> expectedNodesList = getExpectedNumaNodesList();
+    Assert.assertEquals(expectedNodesList, nodesList);
+  }
+
+  @Test
+  public void testReadNumaTopologyFromCmdOutput() throws Exception {
+    conf.setBoolean(YarnConfiguration.NM_NUMA_AWARENESS_READ_TOPOLOGY, true);
+    final String cmdOutput = "available: 2 nodes (0-1)\n\t"
+        + "node 0 cpus: 0 2 4 6\n\t"
+        + "node 0 size: 73717 MB\n\t"
+        + "node 0 free: 17272 MB\n\t"
+        + "node 1 cpus: 1 3 5 7\n\t"
+        + "node 1 size: 73727 MB\n\t"
+        + "node 1 free: 10699 MB\n\t"
+        + "node distances:\n\t"
+        + "node 0 1\n\t"
+        + "0: 10 20\n\t"
+        + "1: 20 10";
+    numaResourcesManager = new NUMAResourcesManager() {
+      @Override
+      String executeNGetCmdOutput(Configuration conf) throws YarnRuntimeException {
+        return cmdOutput;
+      }
+    };
+    numaResourcesManager.serviceInit(conf);
+    Collection<NumaNode> nodesList = numaResourcesManager.getNumaNodesList();
+    Collection<NumaNode> expectedNodesList = getExpectedNumaNodesList();
+    Assert.assertEquals(expectedNodesList, nodesList);
+  }
+
+  @Test
+  public void testAssignNumaNode() throws Exception {
+    setNumaTopologyConfigs();
+    numaResourcesManager.serviceInit(conf);
+    AssignedNumaNodeInfo nodeInfo = numaResourcesManager.assignNumaNode(
+        ContainerId.fromString("container_1481156246874_0001_01_000001"),
+        Resource.newInstance(2048, 2));
+    Assert.assertEquals("0", nodeInfo.getMemNode());
+    Assert.assertEquals("0", nodeInfo.getCpuNode());
+  }
+
+  @Test
+  public void testAssignNumaNodeWithRoundRobinFashionAssignment()
+      throws Exception {
+    setNumaTopologyConfigs();
+    numaResourcesManager.serviceInit(conf);
+    AssignedNumaNodeInfo nodeInfo1 = numaResourcesManager.assignNumaNode(
+        ContainerId.fromString("container_1481156246874_0001_01_000001"),
+        Resource.newInstance(2048, 2));
+    Assert.assertEquals("0", nodeInfo1.getMemNode());
+    Assert.assertEquals("0", nodeInfo1.getCpuNode());
+
+    AssignedNumaNodeInfo nodeInfo2 = numaResourcesManager.assignNumaNode(
+        ContainerId.fromString("container_1481156246874_0001_01_000002"),
+        Resource.newInstance(2048, 2));
+    Assert.assertEquals("1", nodeInfo2.getMemNode());
+    Assert.assertEquals("1", nodeInfo2.getCpuNode());
+
+    AssignedNumaNodeInfo nodeInfo3 = numaResourcesManager.assignNumaNode(
+        ContainerId.fromString("container_1481156246874_0001_01_000003"),
+        Resource.newInstance(2048, 2));
+    Assert.assertEquals("0", nodeInfo3.getMemNode());
+    Assert.assertEquals("0", nodeInfo3.getCpuNode());
+
+    AssignedNumaNodeInfo nodeInfo4 = numaResourcesManager.assignNumaNode(
+        ContainerId.fromString("container_1481156246874_0001_01_000003"),
+        Resource.newInstance(2048, 2));
+    Assert.assertEquals("1", nodeInfo4.getMemNode());
+    Assert.assertEquals("1", nodeInfo4.getCpuNode());
+  }
+
+  @Test
+  public void testAssignNumaNodeWithMultipleNodesForMemory() throws Exception {
+    setNumaTopologyConfigs();
+    numaResourcesManager.serviceInit(conf);
+    AssignedNumaNodeInfo nodeInfo = numaResourcesManager.assignNumaNode(
+        ContainerId.fromString("container_1481156246874_0001_01_000001"),
+        Resource.newInstance(102400, 2));
+    Assert.assertEquals("0,1", nodeInfo.getMemNode());
+    Assert.assertEquals("0", nodeInfo.getCpuNode());
+  }
+
+  @Test
+  public void testAssignNumaNodeWithMultipleNodesForCpus() throws Exception {
+    setNumaTopologyConfigs();
+    numaResourcesManager.serviceInit(conf);
+    AssignedNumaNodeInfo nodeInfo = numaResourcesManager.assignNumaNode(
+        ContainerId.fromString("container_1481156246874_0001_01_000001"),
+        Resource.newInstance(2048, 6));
+    Assert.assertEquals("0", nodeInfo.getMemNode());
+    Assert.assertEquals("0,1", nodeInfo.getCpuNode());
+  }
+
+  @Test
+  public void testAssignNumaNodeWhenNoNumaMemResourcesAvailable()
+      throws Exception {
+    setNumaTopologyConfigs();
+    numaResourcesManager.serviceInit(conf);
+    AssignedNumaNodeInfo nodeInfo = numaResourcesManager.assignNumaNode(
+        ContainerId.fromString("container_1481156246874_0001_01_000001"),
+        Resource.newInstance(2048000, 6));
+    Assert.assertNull("Should not assign numa nodes when there"
+        + " are no sufficient memory resources available.", nodeInfo);
+  }
+
+  @Test
+  public void testAssignNumaNodeWhenNoNumaCpuResourcesAvailable()
+      throws Exception {
+    setNumaTopologyConfigs();
+    numaResourcesManager.serviceInit(conf);
+    AssignedNumaNodeInfo nodeInfo = numaResourcesManager.assignNumaNode(
+        ContainerId.fromString("container_1481156246874_0001_01_000001"),
+        Resource.newInstance(2048, 600));
+    Assert.assertNull("Should not assign numa nodes when there"
+        + " are no sufficient cpu resources available.", nodeInfo);
+  }
+
+  @Test
+  public void testReleaseNumaResourcess() throws Exception {
+    setNumaTopologyConfigs();
+    numaResourcesManager.serviceInit(conf);
+    AssignedNumaNodeInfo nodeInfo = numaResourcesManager.assignNumaNode(
+        ContainerId.fromString("container_1481156246874_0001_01_000001"),
+        Resource.newInstance(2048, 8));
+    Assert.assertEquals("0", nodeInfo.getMemNode());
+    Assert.assertEquals("0,1", nodeInfo.getCpuNode());
+
+    // Request the resource when all cpu nodes occupied
+    nodeInfo = numaResourcesManager.assignNumaNode(
+        ContainerId.fromString("container_1481156246874_0001_01_000002"),
+        Resource.newInstance(2048, 4));
+    Assert.assertNull("Should not assign numa nodes when there"
+        + " are no sufficient cpu resources available.", nodeInfo);
+
+    // Release the resources
+    numaResourcesManager.releaseNumaResources(
+        ContainerId.fromString("container_1481156246874_0001_01_000001"));
+    // Request the resources
+    nodeInfo = numaResourcesManager.assignNumaNode(
+        ContainerId.fromString("container_1481156246874_0001_01_000003"),
+        Resource.newInstance(1024, 2));
+    Assert.assertEquals("0", nodeInfo.getMemNode());
+    Assert.assertEquals("0", nodeInfo.getCpuNode());
+  }
+
+  private void setNumaTopologyConfigs() {
+    conf.set(YarnConfiguration.NM_NUMA_AWARENESS_NODE_IDS, "0,1");
+    conf.set("yarn.nodemanager.numa-awareness.0.memory", "73717");
+    conf.set("yarn.nodemanager.numa-awareness.0.cpus", "4");
+    conf.set("yarn.nodemanager.numa-awareness.1.memory", "73727");
+    conf.set("yarn.nodemanager.numa-awareness.1.cpus", "4");
+  }
+
+  private Collection<NumaNode> getExpectedNumaNodesList() {
+    Collection<NumaNode> expectedNodesList = new ArrayList<>(2);
+    expectedNodesList.add(new NumaNode("0", 73717, 4));
+    expectedNodesList.add(new NumaNode("1", 73727, 4));
+    return expectedNodesList;
+  }
+}
-- 
1.8.3.1

