You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@myriad.apache.org by mo...@apache.org on 2015/09/26 00:45:56 UTC
[16/50] [abbrv] incubator-myriad git commit: Update Myriad's hadoop
dependency from version 2.5 to 2.7.
Update Myriad's hadoop dependency from version 2.5 to 2.7.
The following problems surfaced during this change:
1. The API that Myriad was using to update NM's capacity is removed in 2.7. It's replaced with a NODE_RESOURCE_UPDATE event (YARN-1506).
2. RM rejects app submissions if the size of the AM container requested is greater than the capacity of the largest NM (YARN-3079, YARN-2604).
Since FGS sets the NM's capacity to zero during NM's registration with RM, the app submission fails.
Project: http://git-wip-us.apache.org/repos/asf/incubator-myriad/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-myriad/commit/2b76199d
Tree: http://git-wip-us.apache.org/repos/asf/incubator-myriad/tree/2b76199d
Diff: http://git-wip-us.apache.org/repos/asf/incubator-myriad/diff/2b76199d
Branch: refs/heads/master
Commit: 2b76199d341a46a8c3bb234eecd67572f496bc02
Parents: b543e17
Author: Santosh Marella <sm...@maprtech.com>
Authored: Fri Jul 10 17:07:53 2015 -0700
Committer: Santosh Marella <sm...@maprtech.com>
Committed: Fri Jul 10 17:07:53 2015 -0700
----------------------------------------------------------------------
build.gradle | 2 +-
.../src/main/java/com/ebay/myriad/Main.java | 11 +-
.../main/java/com/ebay/myriad/MyriadModule.java | 5 +
.../myriad/policy/LeastAMNodesFirstPolicy.java | 2 +-
.../myriad/scheduler/NMHeartBeatHandler.java | 112 +++++++++-----
.../java/com/ebay/myriad/scheduler/Node.java | 2 +-
.../com/ebay/myriad/scheduler/OfferUtils.java | 35 +++++
.../scheduler/YarnNodeCapacityManager.java | 151 +++++--------------
.../scheduler/yarn/MyriadCapacityScheduler.java | 73 +++++----
.../scheduler/yarn/MyriadFairScheduler.java | 23 +--
.../scheduler/yarn/MyriadFifoScheduler.java | 73 +++++----
.../yarn/interceptor/BaseInterceptor.java | 2 +-
.../yarn/interceptor/CompositeInterceptor.java | 5 +-
.../MyriadInitializationInterceptor.java | 5 +-
.../interceptor/YarnSchedulerInterceptor.java | 3 +-
15 files changed, 265 insertions(+), 239 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-myriad/blob/2b76199d/build.gradle
----------------------------------------------------------------------
diff --git a/build.gradle b/build.gradle
index 66e1245..25fa361 100644
--- a/build.gradle
+++ b/build.gradle
@@ -34,7 +34,7 @@ subprojects {
ext {
mesosVer = "0.21.1"
- hadoopVer = "2.5.0"
+ hadoopVer = "2.7.0"
metricsVer = "3.1.0"
}
http://git-wip-us.apache.org/repos/asf/incubator-myriad/blob/2b76199d/myriad-scheduler/src/main/java/com/ebay/myriad/Main.java
----------------------------------------------------------------------
diff --git a/myriad-scheduler/src/main/java/com/ebay/myriad/Main.java b/myriad-scheduler/src/main/java/com/ebay/myriad/Main.java
index 7358a3a..a756bb0 100644
--- a/myriad-scheduler/src/main/java/com/ebay/myriad/Main.java
+++ b/myriad-scheduler/src/main/java/com/ebay/myriad/Main.java
@@ -36,6 +36,7 @@ import com.google.inject.Guice;
import com.google.inject.Injector;
import org.apache.commons.collections.MapUtils;
import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.yarn.server.resourcemanager.RMContext;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.AbstractYarnScheduler;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -62,19 +63,20 @@ public class Main {
public static void initialize(Configuration hadoopConf,
AbstractYarnScheduler yarnScheduler,
+ RMContext rmContext,
InterceptorRegistry registry) throws Exception {
ObjectMapper mapper = new ObjectMapper(new YAMLFactory());
MyriadConfiguration cfg = mapper.readValue(
Thread.currentThread().getContextClassLoader().getResource("myriad-config-default.yml"),
MyriadConfiguration.class);
- MyriadModule myriadModule = new MyriadModule(cfg, hadoopConf, yarnScheduler, registry);
+ MyriadModule myriadModule = new MyriadModule(cfg, hadoopConf, yarnScheduler, rmContext, registry);
MesosModule mesosModule = new MesosModule();
injector = Guice.createInjector(
myriadModule, mesosModule,
new WebAppGuiceModule());
- new Main().run(cfg, hadoopConf, yarnScheduler, registry);
+ new Main().run(cfg);
}
// TODO (Kannan Rajah) Hack to get injector in unit test.
@@ -82,10 +84,7 @@ public class Main {
return injector;
}
- public void run(MyriadConfiguration cfg,
- Configuration hadoopConf,
- AbstractYarnScheduler yarnScheduler,
- InterceptorRegistry registry) throws Exception {
+ public void run(MyriadConfiguration cfg) throws Exception {
if (LOGGER.isDebugEnabled()) {
LOGGER.debug("Bindings: " + injector.getAllBindings());
}
http://git-wip-us.apache.org/repos/asf/incubator-myriad/blob/2b76199d/myriad-scheduler/src/main/java/com/ebay/myriad/MyriadModule.java
----------------------------------------------------------------------
diff --git a/myriad-scheduler/src/main/java/com/ebay/myriad/MyriadModule.java b/myriad-scheduler/src/main/java/com/ebay/myriad/MyriadModule.java
index 7d04992..1ced6bb 100644
--- a/myriad-scheduler/src/main/java/com/ebay/myriad/MyriadModule.java
+++ b/myriad-scheduler/src/main/java/com/ebay/myriad/MyriadModule.java
@@ -38,6 +38,7 @@ import com.google.inject.Scopes;
import com.google.inject.Singleton;
import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.yarn.server.resourcemanager.RMContext;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.AbstractYarnScheduler;
import org.apache.mesos.state.State;
import org.slf4j.Logger;
@@ -53,15 +54,18 @@ public class MyriadModule extends AbstractModule {
private MyriadConfiguration cfg;
private Configuration hadoopConf;
private AbstractYarnScheduler yarnScheduler;
+ private final RMContext rmContext;
private InterceptorRegistry interceptorRegistry;
public MyriadModule(MyriadConfiguration cfg,
Configuration hadoopConf,
AbstractYarnScheduler yarnScheduler,
+ RMContext rmContext,
InterceptorRegistry interceptorRegistry) {
this.cfg = cfg;
this.hadoopConf = hadoopConf;
this.yarnScheduler = yarnScheduler;
+ this.rmContext = rmContext;
this.interceptorRegistry = interceptorRegistry;
}
@@ -70,6 +74,7 @@ public class MyriadModule extends AbstractModule {
LOGGER.debug("Configuring guice");
bind(MyriadConfiguration.class).toInstance(cfg);
bind(Configuration.class).toInstance(hadoopConf);
+ bind(RMContext.class).toInstance(rmContext);
bind(AbstractYarnScheduler.class).toInstance(yarnScheduler);
bind(InterceptorRegistry.class).toInstance(interceptorRegistry);
bind(MyriadDriverManager.class).in(Scopes.SINGLETON);
http://git-wip-us.apache.org/repos/asf/incubator-myriad/blob/2b76199d/myriad-scheduler/src/main/java/com/ebay/myriad/policy/LeastAMNodesFirstPolicy.java
----------------------------------------------------------------------
diff --git a/myriad-scheduler/src/main/java/com/ebay/myriad/policy/LeastAMNodesFirstPolicy.java b/myriad-scheduler/src/main/java/com/ebay/myriad/policy/LeastAMNodesFirstPolicy.java
index 3624ddd..13a4f82 100644
--- a/myriad-scheduler/src/main/java/com/ebay/myriad/policy/LeastAMNodesFirstPolicy.java
+++ b/myriad-scheduler/src/main/java/com/ebay/myriad/policy/LeastAMNodesFirstPolicy.java
@@ -99,8 +99,8 @@ public class LeastAMNodesFirstPolicy extends BaseInterceptor implements NodeScal
case NODE_REMOVED:
onNodeRemoved((NodeRemovedSchedulerEvent) event);
break;
+
default:
- LOGGER.warn("event type not supported");
break;
}
} catch (ClassCastException e) {
http://git-wip-us.apache.org/repos/asf/incubator-myriad/blob/2b76199d/myriad-scheduler/src/main/java/com/ebay/myriad/scheduler/NMHeartBeatHandler.java
----------------------------------------------------------------------
diff --git a/myriad-scheduler/src/main/java/com/ebay/myriad/scheduler/NMHeartBeatHandler.java b/myriad-scheduler/src/main/java/com/ebay/myriad/scheduler/NMHeartBeatHandler.java
index 9cec3b5..e99baa3 100644
--- a/myriad-scheduler/src/main/java/com/ebay/myriad/scheduler/NMHeartBeatHandler.java
+++ b/myriad-scheduler/src/main/java/com/ebay/myriad/scheduler/NMHeartBeatHandler.java
@@ -1,28 +1,28 @@
package com.ebay.myriad.scheduler;
+import com.ebay.myriad.executor.ContainerTaskStatusRequest;
+import com.ebay.myriad.scheduler.yarn.interceptor.BaseInterceptor;
+import com.ebay.myriad.scheduler.yarn.interceptor.InterceptorRegistry;
+import com.google.gson.Gson;
import java.nio.charset.Charset;
+import java.util.ArrayList;
import java.util.List;
-
import javax.inject.Inject;
-
import org.apache.hadoop.yarn.api.records.ContainerId;
import org.apache.hadoop.yarn.api.records.ContainerState;
import org.apache.hadoop.yarn.api.records.ContainerStatus;
+import org.apache.hadoop.yarn.api.records.Resource;
import org.apache.hadoop.yarn.server.resourcemanager.RMContext;
import org.apache.hadoop.yarn.server.resourcemanager.rmnode.RMNode;
import org.apache.hadoop.yarn.server.resourcemanager.rmnode.RMNodeEvent;
import org.apache.hadoop.yarn.server.resourcemanager.rmnode.RMNodeStatusEvent;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.AbstractYarnScheduler;
-import org.apache.hadoop.yarn.server.resourcemanager.scheduler.SchedulerNode;
+import org.apache.hadoop.yarn.util.resource.Resources;
import org.apache.mesos.Protos;
+import org.apache.mesos.Protos.Offer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import com.ebay.myriad.executor.ContainerTaskStatusRequest;
-import com.ebay.myriad.scheduler.yarn.interceptor.BaseInterceptor;
-import com.ebay.myriad.scheduler.yarn.interceptor.InterceptorRegistry;
-import com.google.gson.Gson;
-
/**
* Handles node manager heartbeat.
*/
@@ -60,13 +60,10 @@ public class NMHeartBeatHandler extends BaseInterceptor {
public void beforeRMNodeEventHandled(RMNodeEvent event, RMContext context) {
switch (event.getType()) {
case STARTED: {
- // TODO (Santosh) We can't zero out resources here in all cases. For e.g.
- // this event might be fired when an existing node is rejoining the
- // cluster (NM restart) as well. Sometimes this event can have a list of
- // container statuses, which, capacity/fifo schedulers seem to handle
- // (perhaps due to work preserving NM restart).
+ // TODO (Santosh) Don't zero out NM's capacity as it's causing trouble with app submissions.
RMNode rmNode = context.getRMNodes().get(event.getNodeId());
- yarnNodeCapacityMgr.removeCapacity(rmNode);
+ rmNode.getTotalCapability().setMemory(0);
+ rmNode.getTotalCapability().setVirtualCores(0);
}
break;
@@ -80,10 +77,8 @@ public class NMHeartBeatHandler extends BaseInterceptor {
}
break;
- default: {
- LOGGER.debug("Unhandled event: {}", event.getClass().getName());
- }
- break;
+ default:
+ break;
}
}
@@ -96,40 +91,83 @@ public class NMHeartBeatHandler extends BaseInterceptor {
RMNodeStatusEvent statusEvent = (RMNodeStatusEvent) event;
RMNode rmNode = context.getRMNodes().get(event.getNodeId());
- SchedulerNode schedulerNode = yarnScheduler.getSchedulerNode(rmNode.getNodeID());
String hostName = rmNode.getNodeID().getHost();
- int rmSchNodeNumContainers = schedulerNode.getNumContainers();
+ validateContainerCount(statusEvent, rmNode);
+ nodeStore.getNode(hostName).snapshotRunningContainers();
+ sendStatusUpdatesToMesosForCompletedContainers(statusEvent);
+
+ // New capacity of the node =
+ // original capacity the NM registered with (profile) +
+ // resources under use on the node (due to previous offers) +
+ // new resources offered by mesos for the node
+ yarnNodeCapacityMgr.setNodeCapacity(rmNode,
+ Resources.add(rmNode.getTotalCapability(),
+ Resources.add(getResourcesUnderUse(statusEvent),
+ getNewResourcesOfferedByMesos(hostName))));
+ }
+
+ private void validateContainerCount(RMNodeStatusEvent statusEvent, RMNode rmNode) {
+ int rmSchNodeNumContainers = yarnScheduler.getSchedulerNode(rmNode.getNodeID()).getNumContainers();
List<ContainerStatus> nmContainers = statusEvent.getContainers();
if (nmContainers.size() != rmSchNodeNumContainers) {
LOGGER.warn("Node: {}, Num Containers known by RM scheduler: {}, "
+ "Num containers NM is reporting: {}",
- hostName, rmSchNodeNumContainers, nmContainers.size());
+ rmNode.getNodeID().getHost(), rmSchNodeNumContainers, nmContainers.size());
}
+ }
+ private Resource getNewResourcesOfferedByMesos(String hostname) {
+ OfferFeed feed = offerLifecycleMgr.getOfferFeed(hostname);
+ if (feed == null) {
+ LOGGER.debug("No offer feed for: {}", hostname);
+ return Resource.newInstance(0, 0);
+ }
+ Resource offeredResources = Resource.newInstance(0, 0);
+ List<Offer> offers = new ArrayList<>();
+ Protos.Offer offer;
+ while ((offer = feed.poll()) != null) {
+ offers.add(offer);
+ offerLifecycleMgr.markAsConsumed(offer);
+ }
+ Resources.addTo(offeredResources, OfferUtils.getYarnResourcesFromMesosOffers(offers));
+ return offeredResources;
+ }
+
+ private Resource getResourcesUnderUse(RMNodeStatusEvent statusEvent) {
+ Resource usedResources = Resource.newInstance(0, 0);
+ for (ContainerStatus status : statusEvent.getContainers()) {
+ if (status.getState() == ContainerState.NEW || status.getState() == ContainerState.RUNNING) {
+ Resources.addTo(usedResources, yarnScheduler.getRMContainer(status.getContainerId()).getAllocatedResource());
+ }
+ }
+ return usedResources;
+ }
+
+ private void sendStatusUpdatesToMesosForCompletedContainers(RMNodeStatusEvent statusEvent) {
// Send task update to Mesos
- for (ContainerStatus status : nmContainers) {
+ Protos.SlaveID slaveId = nodeStore.getNode(statusEvent.getNodeId().getHost()).getSlaveId();
+ for (ContainerStatus status : statusEvent.getContainers()) {
ContainerId containerId = status.getContainerId();
- Protos.SlaveID slaveId = nodeStore.getNode(hostName).getSlaveId();
if (status.getState() == ContainerState.COMPLETE) {
- LOGGER.debug("Task complete: {}", containerId);
- yarnNodeCapacityMgr.removeCapacity(rmNode, containerId);
- requestExecutorToSendTaskStatusUpdate(
- slaveId,
- containerId, Protos.TaskState.TASK_FINISHED);
+ requestExecutorToSendTaskStatusUpdate(slaveId, containerId, Protos.TaskState.TASK_FINISHED);
} else { // state == NEW | RUNNING
- requestExecutorToSendTaskStatusUpdate(
- slaveId,
- containerId, Protos.TaskState.TASK_RUNNING);
+ requestExecutorToSendTaskStatusUpdate(slaveId, containerId, Protos.TaskState.TASK_RUNNING);
}
}
}
+
/**
* sends a request to executor on the given slave to send back a status update
* for the mesos task launched for this container.
*
+ * TODO(Santosh):
+ * Framework messages are unreliable. Try a NM auxiliary service that can help
+ * send out the status messages from NM itself. NM and MyriadExecutor would need
+ * to be merged into a single process.
+ *
* @param slaveId
* @param containerId
* @param taskState
@@ -137,13 +175,15 @@ public class NMHeartBeatHandler extends BaseInterceptor {
private void requestExecutorToSendTaskStatusUpdate(Protos.SlaveID slaveId,
ContainerId containerId,
Protos.TaskState taskState) {
+ final String mesosTaskId = ContainerTaskStatusRequest.YARN_CONTAINER_TASK_ID_PREFIX + containerId.toString();
+ if (LOGGER.isDebugEnabled()) {
+ LOGGER.debug("Sending out framework message requesting the executor to send {} status for task: {}",
+ taskState.name(), mesosTaskId);
+ }
ContainerTaskStatusRequest containerTaskStatusRequest = new ContainerTaskStatusRequest();
- containerTaskStatusRequest.setMesosTaskId(
- ContainerTaskStatusRequest.YARN_CONTAINER_TASK_ID_PREFIX + containerId.toString());
+ containerTaskStatusRequest.setMesosTaskId(mesosTaskId);
containerTaskStatusRequest.setState(taskState.name());
- myriadDriver.getDriver().sendFrameworkMessage(
- getExecutorId(slaveId),
- slaveId,
+ myriadDriver.getDriver().sendFrameworkMessage(getExecutorId(slaveId), slaveId,
new Gson().toJson(containerTaskStatusRequest).getBytes(Charset.defaultCharset()));
}
http://git-wip-us.apache.org/repos/asf/incubator-myriad/blob/2b76199d/myriad-scheduler/src/main/java/com/ebay/myriad/scheduler/Node.java
----------------------------------------------------------------------
diff --git a/myriad-scheduler/src/main/java/com/ebay/myriad/scheduler/Node.java b/myriad-scheduler/src/main/java/com/ebay/myriad/scheduler/Node.java
index 03ab623..ae478d4 100644
--- a/myriad-scheduler/src/main/java/com/ebay/myriad/scheduler/Node.java
+++ b/myriad-scheduler/src/main/java/com/ebay/myriad/scheduler/Node.java
@@ -57,7 +57,7 @@ public class Node {
this.execInfo = execInfo;
}
- public void snapshotContainers() {
+ public void snapshotRunningContainers() {
this.containerSnapshot = new HashSet<>(node.getRunningContainers());
}
http://git-wip-us.apache.org/repos/asf/incubator-myriad/blob/2b76199d/myriad-scheduler/src/main/java/com/ebay/myriad/scheduler/OfferUtils.java
----------------------------------------------------------------------
diff --git a/myriad-scheduler/src/main/java/com/ebay/myriad/scheduler/OfferUtils.java b/myriad-scheduler/src/main/java/com/ebay/myriad/scheduler/OfferUtils.java
new file mode 100644
index 0000000..e215427
--- /dev/null
+++ b/myriad-scheduler/src/main/java/com/ebay/myriad/scheduler/OfferUtils.java
@@ -0,0 +1,35 @@
+package com.ebay.myriad.scheduler;
+
+import java.util.Collection;
+import org.apache.hadoop.yarn.api.records.Resource;
+import org.apache.mesos.Protos;
+import org.apache.mesos.Protos.Offer;
+
+/**
+ * Utility class that provides useful methods that deal with Mesos offers.
+ */
+public class OfferUtils {
+
+ /**
+ * Transforms a collection of mesos offers into {@link Resource}.
+ *
+ * @param offers collection of mesos offers
+ * @return a single resource object equivalent to the cumulative sum of mesos offers
+ */
+ public static Resource getYarnResourcesFromMesosOffers(Collection<Offer> offers) {
+ double cpus = 0.0;
+ double mem = 0.0;
+
+ for (Protos.Offer offer : offers) {
+ for (Protos.Resource resource : offer.getResourcesList()) {
+ if (resource.getName().equalsIgnoreCase("cpus")) {
+ cpus += resource.getScalar().getValue();
+ } else if (resource.getName().equalsIgnoreCase("mem")) {
+ mem += resource.getScalar().getValue();
+ }
+ }
+ }
+ return Resource.newInstance((int) mem, (int) cpus);
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-myriad/blob/2b76199d/myriad-scheduler/src/main/java/com/ebay/myriad/scheduler/YarnNodeCapacityManager.java
----------------------------------------------------------------------
diff --git a/myriad-scheduler/src/main/java/com/ebay/myriad/scheduler/YarnNodeCapacityManager.java b/myriad-scheduler/src/main/java/com/ebay/myriad/scheduler/YarnNodeCapacityManager.java
index e72317b..9787e28 100644
--- a/myriad-scheduler/src/main/java/com/ebay/myriad/scheduler/YarnNodeCapacityManager.java
+++ b/myriad-scheduler/src/main/java/com/ebay/myriad/scheduler/YarnNodeCapacityManager.java
@@ -1,23 +1,25 @@
package com.ebay.myriad.scheduler;
-import java.util.Arrays;
-import java.util.Collection;
+import com.ebay.myriad.executor.ContainerTaskStatusRequest;
+import com.ebay.myriad.scheduler.yarn.interceptor.BaseInterceptor;
+import com.ebay.myriad.scheduler.yarn.interceptor.InterceptorRegistry;
+import com.google.common.collect.Lists;
+import com.google.common.collect.Sets;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
-
import javax.inject.Inject;
-
import org.apache.hadoop.yarn.api.records.Container;
-import org.apache.hadoop.yarn.api.records.ContainerId;
import org.apache.hadoop.yarn.api.records.NodeId;
import org.apache.hadoop.yarn.api.records.Resource;
import org.apache.hadoop.yarn.api.records.ResourceOption;
+import org.apache.hadoop.yarn.server.resourcemanager.RMContext;
import org.apache.hadoop.yarn.server.resourcemanager.rmcontainer.RMContainer;
import org.apache.hadoop.yarn.server.resourcemanager.rmnode.RMNode;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.AbstractYarnScheduler;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.SchedulerNode;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.event.NodeAddedSchedulerEvent;
+import org.apache.hadoop.yarn.server.resourcemanager.scheduler.event.NodeResourceUpdateSchedulerEvent;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.event.NodeUpdateSchedulerEvent;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.event.SchedulerEvent;
import org.apache.hadoop.yarn.util.resource.Resources;
@@ -25,16 +27,10 @@ import org.apache.mesos.Protos;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import com.ebay.myriad.executor.ContainerTaskStatusRequest;
-import com.ebay.myriad.scheduler.yarn.interceptor.BaseInterceptor;
-import com.ebay.myriad.scheduler.yarn.interceptor.InterceptorRegistry;
-import com.google.common.collect.Lists;
-import com.google.common.collect.Sets;
-
/**
* Manages the capacity exposed by NodeManager. It uses the offers available
* from Mesos to inflate the node capacity and lets ResourceManager make the
- * scheduling decision. After the sheduling decision is done, there are 2 cases:
+ * scheduling decision. After the scheduling decision is done, there are 2 cases:
*
* 1. If ResourceManager did not use the expanded capacity, then the node's
* capacity is reverted back to original value and the offer is declined.
@@ -47,6 +43,7 @@ public class YarnNodeCapacityManager extends BaseInterceptor {
YarnNodeCapacityManager.class);
private final AbstractYarnScheduler yarnScheduler;
+ private final RMContext rmContext;
private final MyriadDriver myriadDriver;
private final OfferLifecycleManager offerLifecycleMgr;
private final NodeStore nodeStore;
@@ -55,6 +52,7 @@ public class YarnNodeCapacityManager extends BaseInterceptor {
@Inject
public YarnNodeCapacityManager(InterceptorRegistry registry,
AbstractYarnScheduler yarnScheduler,
+ RMContext rmContext,
MyriadDriver myriadDriver,
TaskFactory taskFactory,
OfferLifecycleManager offerLifecycleMgr,
@@ -63,6 +61,7 @@ public class YarnNodeCapacityManager extends BaseInterceptor {
registry.register(this);
}
this.yarnScheduler = yarnScheduler;
+ this.rmContext = rmContext;
this.myriadDriver = myriadDriver;
this.taskFactory = taskFactory;
this.offerLifecycleMgr = offerLifecycleMgr;
@@ -70,41 +69,6 @@ public class YarnNodeCapacityManager extends BaseInterceptor {
}
@Override
- public void beforeSchedulerEventHandled(SchedulerEvent event) {
- // TODO (Kannan) This will not work with continuous scheduling
- // Need to find out a hook to execute before the actual scheduling attempt
- // is done.
- if (!(event instanceof NodeUpdateSchedulerEvent)) {
- return;
- }
-
- RMNode rmNode = ((NodeUpdateSchedulerEvent) event).getRMNode();
- applyOffers(rmNode);
- }
-
- /**
- * Applies the offers stashed till now for this node.
- */
- private void applyOffers(RMNode rmNode) {
- String hostname = rmNode.getNodeID().getHost();
- OfferFeed feed = offerLifecycleMgr.getOfferFeed(hostname);
- if (feed == null) {
- LOGGER.debug("No offer feed for: {}", hostname);
- return;
- }
-
- Protos.Offer offer;
- Node node = nodeStore.getNode(hostname);
- node.snapshotContainers();
- while ((offer = feed.poll()) != null) {
- Resource resOffered = getResourceFromOffer(Arrays.asList(offer));
- addCapacity(rmNode, resOffered);
-
- offerLifecycleMgr.markAsConsumed(offer);
- }
- }
-
- @Override
public void afterSchedulerEventHandled(SchedulerEvent event) {
switch (event.getType()) {
case NODE_ADDED: {
@@ -141,10 +105,8 @@ public class YarnNodeCapacityManager extends BaseInterceptor {
}
break;
- default: {
- LOGGER.debug("Unhandled event: {}", event.getClass().getName());
- }
- break;
+ default:
+ break;
}
}
@@ -177,10 +139,9 @@ public class YarnNodeCapacityManager extends BaseInterceptor {
LOGGER.debug("No containers allocated using Mesos offers for host: {}", host);
for (Protos.Offer offer : consumedOffer.getOffers()) {
offerLifecycleMgr.declineOffer(offer);
-
- Resource resUnused = getResourceFromOffer(Arrays.asList(offer));
- removeCapacity(rmNode, resUnused);
}
+ setNodeCapacity(rmNode, Resources.subtract(rmNode.getTotalCapability(),
+ OfferUtils.getYarnResourcesFromMesosOffers(consumedOffer.getOffers())));
} else {
LOGGER.debug("Containers allocated using Mesos offers for host: {} count: {}",
host, containersAllocatedByMesosOffer.size());
@@ -195,9 +156,9 @@ public class YarnNodeCapacityManager extends BaseInterceptor {
}
// Reduce node capacity to account for unused offers
- Resource resOffered = getResourceFromOffer(consumedOffer.getOffers());
+ Resource resOffered = OfferUtils.getYarnResourcesFromMesosOffers(consumedOffer.getOffers());
Resource resUnused = Resources.subtract(resOffered, resUsed);
- removeCapacity(rmNode, resUnused);
+ setNodeCapacity(rmNode, Resources.subtract(rmNode.getTotalCapability(), resUnused));
myriadDriver.getDriver().launchTasks(consumedOffer.getOfferIds(), tasks);
}
@@ -206,62 +167,28 @@ public class YarnNodeCapacityManager extends BaseInterceptor {
node.removeContainerSnapshot();
}
- private void addCapacity(RMNode rmNode, Resource resource) {
- Resource currentResource = rmNode.getResourceOption().getResource();
- Resource newResource = Resources.add(currentResource, resource);
-
- rmNode.setResourceOption(
- ResourceOption.newInstance(newResource,
- RMNode.OVER_COMMIT_TIMEOUT_MILLIS_DEFAULT));
-
- LOGGER.debug("Added_Resource: {}, Delta: {}, New NM Capacity: {}",
- rmNode.getNodeID(), resource, newResource);
- }
-
- private void removeCapacity(RMNode rmNode, Resource resource) {
- Resource currentResource = rmNode.getResourceOption().getResource();
- // Explicitly subtracting each resource since Mesos does not
- // support Disk IO.
- currentResource.setVirtualCores(currentResource.getVirtualCores() - resource.getVirtualCores());
- currentResource.setMemory(currentResource.getMemory() - resource.getMemory());
-
- LOGGER.debug("Removed_Resource: {}, Delta: {}, New NM Capacity: {}",
- rmNode.getNodeID(), resource, currentResource);
- }
-
- public void removeCapacity(RMNode rmNode, ContainerId containerId) {
- removeCapacity(rmNode,
- yarnScheduler.getRMContainer(containerId).getAllocatedResource());
- }
-
- /**
- * Removes the entire capacity of the node.
- */
- public void removeCapacity(RMNode rmNode) {
- Resource currentResource = rmNode.getResourceOption().getResource();
- // Mesos does not support disk IO. So not zeroing out disks.
- currentResource.setVirtualCores(0);
- currentResource.setMemory(0);
- }
-
- private Resource getResourceFromOffer(Collection<Protos.Offer> offers) {
- double cpus = 0.0;
- double mem = 0.0;
-
- for (Protos.Offer offer : offers) {
- for (Protos.Resource resource : offer.getResourcesList()) {
- if (resource.getName().equalsIgnoreCase("cpus")) {
- cpus += resource.getScalar().getValue();
- } else if (resource.getName().equalsIgnoreCase("mem")) {
- mem += resource.getScalar().getValue();
- }
- }
- }
-
- return Resource.newInstance((int) mem, (int) cpus);
- }
-
- private Protos.TaskInfo getTaskInfoForContainer(RMContainer rmContainer,
+ /**
+ * 1. Updates {@link RMNode#getTotalCapability()} with newCapacity.
+ * 2. Sends out a {@link NodeResourceUpdateSchedulerEvent} that's handled by YARN's scheduler.
+ * The scheduler updates the corresponding {@link SchedulerNode} with the newCapacity.
+ *
+ * @param rmNode
+ * @param newCapacity
+ */
+ @SuppressWarnings("unchecked")
+ public void setNodeCapacity(RMNode rmNode, Resource newCapacity) {
+ rmNode.getTotalCapability().setMemory(newCapacity.getMemory());
+ rmNode.getTotalCapability().setVirtualCores(newCapacity.getVirtualCores());
+
+ // updates the scheduler with the new capacity for the NM.
+ // the event is handled by the scheduler asynchronously
+ rmContext.getDispatcher().getEventHandler().handle(
+ new NodeResourceUpdateSchedulerEvent(rmNode,
+ ResourceOption.newInstance(rmNode.getTotalCapability(),
+ RMNode.OVER_COMMIT_TIMEOUT_MILLIS_DEFAULT)));
+ }
+
+ private Protos.TaskInfo getTaskInfoForContainer(RMContainer rmContainer,
ConsumedOffer consumedOffer, Node node) {
Protos.Offer offer = consumedOffer.getOffers().get(0);
http://git-wip-us.apache.org/repos/asf/incubator-myriad/blob/2b76199d/myriad-scheduler/src/main/java/com/ebay/myriad/scheduler/yarn/MyriadCapacityScheduler.java
----------------------------------------------------------------------
diff --git a/myriad-scheduler/src/main/java/com/ebay/myriad/scheduler/yarn/MyriadCapacityScheduler.java b/myriad-scheduler/src/main/java/com/ebay/myriad/scheduler/yarn/MyriadCapacityScheduler.java
index 0351f06..8355ab3 100644
--- a/myriad-scheduler/src/main/java/com/ebay/myriad/scheduler/yarn/MyriadCapacityScheduler.java
+++ b/myriad-scheduler/src/main/java/com/ebay/myriad/scheduler/yarn/MyriadCapacityScheduler.java
@@ -4,45 +4,58 @@ import com.ebay.myriad.scheduler.yarn.interceptor.CompositeInterceptor;
import com.ebay.myriad.scheduler.yarn.interceptor.YarnSchedulerInterceptor;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.yarn.server.resourcemanager.RMContext;
+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.scheduler.capacity.CapacityScheduler;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.event.SchedulerEvent;
-import java.io.IOException;
-
/**
* {@link MyriadCapacityScheduler} just extends YARN's {@link CapacityScheduler} and
* allows some of the {@link CapacityScheduler} methods to be intercepted
* via the {@link YarnSchedulerInterceptor} interface.
*/
public class MyriadCapacityScheduler extends CapacityScheduler {
- private final YarnSchedulerInterceptor yarnSchedulerInterceptor;
-
- public MyriadCapacityScheduler() {
- super();
- this.yarnSchedulerInterceptor = new CompositeInterceptor();
- }
-
- /**
- * ******** Methods overridden from YARN {@link CapacityScheduler} *********************
- */
-
- @Override
- public void reinitialize(Configuration conf, RMContext rmContext) throws IOException {
- this.yarnSchedulerInterceptor.init(conf, this);
- super.reinitialize(conf, rmContext);
- }
-
- @Override
- public void serviceInit(Configuration conf) throws Exception {
- this.yarnSchedulerInterceptor.init(conf, this);
- super.serviceInit(conf);
- }
-
- @Override
- public void handle(SchedulerEvent event) {
- super.handle(event);
- this.yarnSchedulerInterceptor.afterSchedulerEventHandled(event);
- }
+ private RMContext rmContext;
+ private YarnSchedulerInterceptor yarnSchedulerInterceptor;
+ private RMNodeEventHandler rmNodeEventHandler;
+
+ public MyriadCapacityScheduler() {
+ super();
+ }
+
+ /**
+ * Register an event handler that receives {@link RMNodeEvent} events.
+ * This event handler is registered ahead of RM's own event handler for RMNodeEvents.
+ * For e.g. myriad can inspect a node's HB (RMNodeStatusEvent) before the HB is handled by
+ * RM and the scheduler.
+ *
+ * @param rmContext
+ */
+ @Override
+ public synchronized void setRMContext(RMContext rmContext) {
+ this.rmContext = rmContext;
+ this.yarnSchedulerInterceptor = new CompositeInterceptor();
+ rmNodeEventHandler = new RMNodeEventHandler(yarnSchedulerInterceptor, rmContext);
+ rmContext.getDispatcher().register(RMNodeEventType.class, rmNodeEventHandler);
+ super.setRMContext(rmContext);
+ }
+
+ /**
+ * ******** Methods overridden from YARN {@link CapacityScheduler} *********************
+ */
+
+ @Override
+ public synchronized void serviceInit(Configuration conf) throws Exception {
+ this.yarnSchedulerInterceptor.init(conf, this, rmContext);
+ super.serviceInit(conf);
+ }
+
+ @Override
+ public synchronized void handle(SchedulerEvent event) {
+ this.yarnSchedulerInterceptor.beforeSchedulerEventHandled(event);
+ super.handle(event);
+ this.yarnSchedulerInterceptor.afterSchedulerEventHandled(event);
+ }
}
http://git-wip-us.apache.org/repos/asf/incubator-myriad/blob/2b76199d/myriad-scheduler/src/main/java/com/ebay/myriad/scheduler/yarn/MyriadFairScheduler.java
----------------------------------------------------------------------
diff --git a/myriad-scheduler/src/main/java/com/ebay/myriad/scheduler/yarn/MyriadFairScheduler.java b/myriad-scheduler/src/main/java/com/ebay/myriad/scheduler/yarn/MyriadFairScheduler.java
index 237f70e..9d5363e 100644
--- a/myriad-scheduler/src/main/java/com/ebay/myriad/scheduler/yarn/MyriadFairScheduler.java
+++ b/myriad-scheduler/src/main/java/com/ebay/myriad/scheduler/yarn/MyriadFairScheduler.java
@@ -8,10 +8,6 @@ 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.scheduler.event.SchedulerEvent;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.fair.FairScheduler;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.io.IOException;
/**
* {@link MyriadFairScheduler} just extends YARN's {@link FairScheduler} and
@@ -19,14 +15,13 @@ import java.io.IOException;
* via the {@link YarnSchedulerInterceptor} interface.
*/
public class MyriadFairScheduler extends FairScheduler {
- private static final Logger LOGGER = LoggerFactory.getLogger(MyriadFairScheduler.class);
- private final YarnSchedulerInterceptor yarnSchedulerInterceptor;
+ private RMContext rmContext;
+ private YarnSchedulerInterceptor yarnSchedulerInterceptor;
private RMNodeEventHandler rmNodeEventHandler;
public MyriadFairScheduler() {
super();
- this.yarnSchedulerInterceptor = new CompositeInterceptor();
}
/**
@@ -39,6 +34,8 @@ public class MyriadFairScheduler extends FairScheduler {
*/
@Override
public synchronized void setRMContext(RMContext rmContext) {
+ this.rmContext = rmContext;
+ this.yarnSchedulerInterceptor = new CompositeInterceptor();
rmNodeEventHandler = new RMNodeEventHandler(yarnSchedulerInterceptor, rmContext);
rmContext.getDispatcher().register(RMNodeEventType.class, rmNodeEventHandler);
super.setRMContext(rmContext);
@@ -49,19 +46,13 @@ public class MyriadFairScheduler extends FairScheduler {
*/
@Override
- public void reinitialize(Configuration conf, RMContext rmContext) throws IOException {
- this.yarnSchedulerInterceptor.init(conf, this);
- super.reinitialize(conf, rmContext);
- }
-
- @Override
- public void serviceInit(Configuration conf) throws Exception {
- this.yarnSchedulerInterceptor.init(conf, this);
+ public synchronized void serviceInit(Configuration conf) throws Exception {
+ this.yarnSchedulerInterceptor.init(conf, this, rmContext);
super.serviceInit(conf);
}
@Override
- public void handle(SchedulerEvent event) {
+ public synchronized void handle(SchedulerEvent event) {
this.yarnSchedulerInterceptor.beforeSchedulerEventHandled(event);
super.handle(event);
this.yarnSchedulerInterceptor.afterSchedulerEventHandled(event);
http://git-wip-us.apache.org/repos/asf/incubator-myriad/blob/2b76199d/myriad-scheduler/src/main/java/com/ebay/myriad/scheduler/yarn/MyriadFifoScheduler.java
----------------------------------------------------------------------
diff --git a/myriad-scheduler/src/main/java/com/ebay/myriad/scheduler/yarn/MyriadFifoScheduler.java b/myriad-scheduler/src/main/java/com/ebay/myriad/scheduler/yarn/MyriadFifoScheduler.java
index 6975645..f03b7ca 100644
--- a/myriad-scheduler/src/main/java/com/ebay/myriad/scheduler/yarn/MyriadFifoScheduler.java
+++ b/myriad-scheduler/src/main/java/com/ebay/myriad/scheduler/yarn/MyriadFifoScheduler.java
@@ -4,45 +4,58 @@ import com.ebay.myriad.scheduler.yarn.interceptor.CompositeInterceptor;
import com.ebay.myriad.scheduler.yarn.interceptor.YarnSchedulerInterceptor;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.yarn.server.resourcemanager.RMContext;
+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.scheduler.event.SchedulerEvent;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.fifo.FifoScheduler;
-import java.io.IOException;
-
/**
* {@link MyriadFifoScheduler} just extends YARN's {@link FifoScheduler} and
* allows some of the {@link FifoScheduler} methods to be intercepted
* via the {@link YarnSchedulerInterceptor} interface.
*/
public class MyriadFifoScheduler extends FifoScheduler {
- private final YarnSchedulerInterceptor yarnSchedulerInterceptor;
-
- public MyriadFifoScheduler() {
- super();
- this.yarnSchedulerInterceptor = new CompositeInterceptor();
- }
-
- /**
- * ******** Methods overridden from YARN {@link FifoScheduler} *********************
- */
-
- @Override
- public void reinitialize(Configuration conf, RMContext rmContext) throws IOException {
- this.yarnSchedulerInterceptor.init(conf, this);
- super.reinitialize(conf, rmContext);
- }
-
- @Override
- public void serviceInit(Configuration conf) throws Exception {
- this.yarnSchedulerInterceptor.init(conf, this);
- super.serviceInit(conf);
- }
-
- @Override
- public void handle(SchedulerEvent event) {
- super.handle(event);
- this.yarnSchedulerInterceptor.afterSchedulerEventHandled(event);
- }
+ private RMContext rmContext;
+ private YarnSchedulerInterceptor yarnSchedulerInterceptor;
+ private RMNodeEventHandler rmNodeEventHandler;
+
+ public MyriadFifoScheduler() {
+ super();
+ }
+
+ /**
+ * Register an event handler that receives {@link RMNodeEvent} events.
+ * This event handler is registered ahead of RM's own event handler for RMNodeEvents.
+ * For e.g. myriad can inspect a node's HB (RMNodeStatusEvent) before the HB is handled by
+ * RM and the scheduler.
+ *
+ * @param rmContext
+ */
+ @Override
+ public synchronized void setRMContext(RMContext rmContext) {
+ this.rmContext = rmContext;
+ this.yarnSchedulerInterceptor = new CompositeInterceptor();
+ rmNodeEventHandler = new RMNodeEventHandler(yarnSchedulerInterceptor, rmContext);
+ rmContext.getDispatcher().register(RMNodeEventType.class, rmNodeEventHandler);
+ super.setRMContext(rmContext);
+ }
+
+ /**
+ * ******** Methods overridden from YARN {@link FifoScheduler} *********************
+ */
+
+ @Override
+ public synchronized void serviceInit(Configuration conf) throws Exception {
+ this.yarnSchedulerInterceptor.init(conf, this, rmContext);
+ super.serviceInit(conf);
+ }
+
+ @Override
+ public synchronized void handle(SchedulerEvent event) {
+ this.yarnSchedulerInterceptor.beforeSchedulerEventHandled(event);
+ super.handle(event);
+ this.yarnSchedulerInterceptor.afterSchedulerEventHandled(event);
+ }
}
http://git-wip-us.apache.org/repos/asf/incubator-myriad/blob/2b76199d/myriad-scheduler/src/main/java/com/ebay/myriad/scheduler/yarn/interceptor/BaseInterceptor.java
----------------------------------------------------------------------
diff --git a/myriad-scheduler/src/main/java/com/ebay/myriad/scheduler/yarn/interceptor/BaseInterceptor.java b/myriad-scheduler/src/main/java/com/ebay/myriad/scheduler/yarn/interceptor/BaseInterceptor.java
index fe3cb2a..e2d63da 100644
--- a/myriad-scheduler/src/main/java/com/ebay/myriad/scheduler/yarn/interceptor/BaseInterceptor.java
+++ b/myriad-scheduler/src/main/java/com/ebay/myriad/scheduler/yarn/interceptor/BaseInterceptor.java
@@ -19,7 +19,7 @@ public class BaseInterceptor implements YarnSchedulerInterceptor {
}
@Override
- public void init(Configuration conf, AbstractYarnScheduler yarnScheduler) throws IOException {
+ public void init(Configuration conf, AbstractYarnScheduler yarnScheduler, RMContext rmContext) throws IOException {
}
@Override
http://git-wip-us.apache.org/repos/asf/incubator-myriad/blob/2b76199d/myriad-scheduler/src/main/java/com/ebay/myriad/scheduler/yarn/interceptor/CompositeInterceptor.java
----------------------------------------------------------------------
diff --git a/myriad-scheduler/src/main/java/com/ebay/myriad/scheduler/yarn/interceptor/CompositeInterceptor.java b/myriad-scheduler/src/main/java/com/ebay/myriad/scheduler/yarn/interceptor/CompositeInterceptor.java
index 2be366a..e93a6a5 100644
--- a/myriad-scheduler/src/main/java/com/ebay/myriad/scheduler/yarn/interceptor/CompositeInterceptor.java
+++ b/myriad-scheduler/src/main/java/com/ebay/myriad/scheduler/yarn/interceptor/CompositeInterceptor.java
@@ -53,11 +53,12 @@ public class CompositeInterceptor implements YarnSchedulerInterceptor, Intercept
*
* @param conf
* @param yarnScheduler
+ * @param rmContext
* @throws IOException
*/
@Override
- public void init(Configuration conf, AbstractYarnScheduler yarnScheduler) throws IOException {
- myriadInitInterceptor.init(conf, yarnScheduler);
+ public void init(Configuration conf, AbstractYarnScheduler yarnScheduler, RMContext rmContext) throws IOException {
+ myriadInitInterceptor.init(conf, yarnScheduler, rmContext);
}
@Override
http://git-wip-us.apache.org/repos/asf/incubator-myriad/blob/2b76199d/myriad-scheduler/src/main/java/com/ebay/myriad/scheduler/yarn/interceptor/MyriadInitializationInterceptor.java
----------------------------------------------------------------------
diff --git a/myriad-scheduler/src/main/java/com/ebay/myriad/scheduler/yarn/interceptor/MyriadInitializationInterceptor.java b/myriad-scheduler/src/main/java/com/ebay/myriad/scheduler/yarn/interceptor/MyriadInitializationInterceptor.java
index 91d65b3..4e071c5 100644
--- a/myriad-scheduler/src/main/java/com/ebay/myriad/scheduler/yarn/interceptor/MyriadInitializationInterceptor.java
+++ b/myriad-scheduler/src/main/java/com/ebay/myriad/scheduler/yarn/interceptor/MyriadInitializationInterceptor.java
@@ -2,6 +2,7 @@ package com.ebay.myriad.scheduler.yarn.interceptor;
import com.ebay.myriad.Main;
import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.yarn.server.resourcemanager.RMContext;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.AbstractYarnScheduler;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -26,9 +27,9 @@ public class MyriadInitializationInterceptor extends BaseInterceptor {
* the myriad web application, initializing guice modules etc.
*/
@Override
- public void init(Configuration conf, AbstractYarnScheduler yarnScheduler) throws IOException {
+ public void init(Configuration conf, AbstractYarnScheduler yarnScheduler, RMContext rmContext) throws IOException {
try {
- Main.initialize(conf, yarnScheduler, registry);
+ Main.initialize(conf, yarnScheduler, rmContext, registry);
} catch (Exception e) {
// Abort bringing up RM
throw new RuntimeException("Failed to initialize myriad", e);
http://git-wip-us.apache.org/repos/asf/incubator-myriad/blob/2b76199d/myriad-scheduler/src/main/java/com/ebay/myriad/scheduler/yarn/interceptor/YarnSchedulerInterceptor.java
----------------------------------------------------------------------
diff --git a/myriad-scheduler/src/main/java/com/ebay/myriad/scheduler/yarn/interceptor/YarnSchedulerInterceptor.java b/myriad-scheduler/src/main/java/com/ebay/myriad/scheduler/yarn/interceptor/YarnSchedulerInterceptor.java
index 7a81c74..4d58857 100644
--- a/myriad-scheduler/src/main/java/com/ebay/myriad/scheduler/yarn/interceptor/YarnSchedulerInterceptor.java
+++ b/myriad-scheduler/src/main/java/com/ebay/myriad/scheduler/yarn/interceptor/YarnSchedulerInterceptor.java
@@ -20,9 +20,10 @@ public interface YarnSchedulerInterceptor {
*
* @param conf
* @param yarnScheduler
+ * @param rmContext
* @throws IOException
*/
- public void init(Configuration conf, AbstractYarnScheduler yarnScheduler) throws IOException;
+ public void init(Configuration conf, AbstractYarnScheduler yarnScheduler, RMContext rmContext) throws IOException;
/**
* Invoked *before* {@link RMNodeImpl#handle(RMNodeEvent)}.