diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/scheduler/DistributedOpportunisticContainerAllocator.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/scheduler/DistributedOpportunisticContainerAllocator.java
new file mode 100644
index 00000000000..da90167fa4c
--- /dev/null
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/scheduler/DistributedOpportunisticContainerAllocator.java
@@ -0,0 +1,357 @@
+/**
+ * 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.scheduler;
+
+import org.apache.commons.lang3.StringUtils;
+import org.apache.hadoop.yarn.api.records.ApplicationAttemptId;
+import org.apache.hadoop.yarn.api.records.Container;
+import org.apache.hadoop.yarn.api.records.Resource;
+import org.apache.hadoop.yarn.api.records.ResourceBlacklistRequest;
+import org.apache.hadoop.yarn.api.records.ResourceRequest;
+import org.apache.hadoop.yarn.exceptions.YarnException;
+
+import org.apache.hadoop.yarn.server.api.protocolrecords.RemoteNode;
+import org.apache.hadoop.yarn.server.metrics.OpportunisticSchedulerMetrics;
+import org.apache.hadoop.yarn.server.security.BaseContainerTokenSecretManager;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+/**
+ *
+ * The DistributedOpportunisticContainerAllocator allocates containers on a
+ * given list of nodes, after modifying the container sizes to respect the
+ * limits set by the ResourceManager. It tries to distribute the containers
+ * as evenly as possible.
+ *
+ */
+public class DistributedOpportunisticContainerAllocator
+ extends OpportunisticContainerAllocator {
+
+ private static final int NODE_LOCAL_LOOP = 0;
+ private static final int RACK_LOCAL_LOOP = 1;
+ private static final int OFF_SWITCH_LOOP = 2;
+
+ private static final Logger LOG =
+ LoggerFactory.getLogger(DistributedOpportunisticContainerAllocator.class);
+
+ /**
+ * Create a new Opportunistic Container Allocator.
+ * @param tokenSecretManager TokenSecretManager
+ */
+ public DistributedOpportunisticContainerAllocator(
+ BaseContainerTokenSecretManager tokenSecretManager) {
+ super(tokenSecretManager);
+ }
+
+ /**
+ * Create a new Opportunistic Container Allocator.
+ * @param tokenSecretManager TokenSecretManager
+ * @param maxAllocationsPerAMHeartbeat max number of containers to be
+ * allocated in one AM heartbeat
+ */
+ public DistributedOpportunisticContainerAllocator(
+ BaseContainerTokenSecretManager tokenSecretManager,
+ int maxAllocationsPerAMHeartbeat) {
+ super(tokenSecretManager, maxAllocationsPerAMHeartbeat);
+ }
+
+ @Override
+ public List allocateContainers(ResourceBlacklistRequest blackList,
+ List oppResourceReqs,
+ ApplicationAttemptId applicationAttemptId,
+ OpportunisticContainerContext opportContext, long rmIdentifier,
+ String appSubmitter) throws YarnException {
+
+ // Update black list.
+ updateBlacklist(blackList, opportContext);
+
+ // Add OPPORTUNISTIC requests to the outstanding ones.
+ opportContext.addToOutstandingReqs(oppResourceReqs);
+ Set nodeBlackList = new HashSet<>(opportContext.getBlacklist());
+ Set allocatedNodes = new HashSet<>();
+ List allocatedContainers = new ArrayList<>();
+
+ // Satisfy the outstanding OPPORTUNISTIC requests.
+ boolean continueLoop = true;
+ while (continueLoop) {
+ continueLoop = false;
+ List