Address review comments; fix findBugs and build issues

- Defined a new 'myriadSchedulerConf' to exclude hadoop dependencies from being copied into myriad-scheduler/build/libs. Tests are now unaffected.
- Fixed a few findBugs issues.
- In ResourceOffersEventHandler, fixed code to ensure an offer used to launch a zero profile NM is not used again for FGS.
- Fixed a test in TestMyriadScheduler. The test for NodeStore needs to be coded differently due to the recent FGS changes. Will be done in a separate changeset.

Author: Santosh Marella

build.gradle

@@ -45,9 +45,6 @@ subprojects {
         capsule
         myriadExecutorConf
         myriadExecutorConf.transitive = false
-        // exclude hadoop/yarn deps for 'runtime'
-        // TODO (Kannan Rajah) Exclude only while packaging
-        //runtime.exclude group: 'org.apache.hadoop', module: '*'
     }
 
     repositories {

myriad-scheduler/build.gradle

@@ -14,12 +14,17 @@ dependencies {
 	compile "com.fasterxml.jackson.core:jackson-databind:2.5.1"
 	compile "com.fasterxml.jackson.dataformat:jackson-dataformat-yaml:2.5.1"
 	compile "org.apache.curator:curator-framework:2.7.1"
-  testCompile "org.apache.hadoop:hadoop-yarn-server-resourcemanager:${hadoopVer}:tests"
+    testCompile "org.apache.hadoop:hadoop-yarn-server-resourcemanager:${hadoopVer}:tests"
+}
+
+configurations {
+    myriadSchedulerConf.extendsFrom(runtime)
+    myriadSchedulerConf.exclude group: 'org.apache.hadoop', module: '*'
 }
 
 // copies dependencies to build/libs dir
 task copyRunTimeDeps(type: Sync) {
-    from configurations.runtime
+    from configurations.myriadSchedulerConf
     into "$buildDir/libs"
 }
 

myriad-scheduler/src/main/java/com/ebay/myriad/Main.java

@@ -163,12 +163,13 @@ private void validateNMInstances(Injector injector) {
         NMProfileManager profileManager = injector.getInstance(NMProfileManager.class);
         long maxCpu = Long.MIN_VALUE;
         long maxMem = Long.MIN_VALUE;
-        for (String profile : nmInstances.keySet()) {
+        for (Map.Entry<String, Integer> entry : nmInstances.entrySet()) {
+          String profile = entry.getKey();
           NMProfile nmProfile = profileManager.get(profile);
           if (nmProfile == null) {
             throw new RuntimeException("Invalid profile name '" + profile + "' specified in 'nmInstances'");
           }
-          if (nmInstances.get(profile) > 0) {
+          if (entry.getValue() > 0) {
             if (nmProfile.getCpus() > maxCpu) { // find the profile with largest number of cpus
               maxCpu = nmProfile.getCpus();
               maxMem = nmProfile.getMemory(); // use the memory from the same profile
@@ -184,8 +185,8 @@ private void validateNMInstances(Injector injector) {
     private void startNMInstances(Injector injector) {
       Map<String, Integer> nmInstances = injector.getInstance(MyriadConfiguration.class).getNmInstances();
       MyriadOperations myriadOperations = injector.getInstance(MyriadOperations.class);
-      for (String profile : nmInstances.keySet()) {
-        myriadOperations.flexUpCluster(nmInstances.get(profile), profile);
+      for (Map.Entry<String, Integer> entry : nmInstances.entrySet()) {
+        myriadOperations.flexUpCluster(entry.getValue(), entry.getKey());
       }
     }
 

myriad-scheduler/src/main/java/com/ebay/myriad/policy/LeastAMNodesFirstPolicy.java

@@ -122,7 +122,7 @@ private void onNodeUpdated(NodeUpdateSchedulerEvent event) {
 
     private void onNodeRemoved(NodeRemovedSchedulerEvent event) {
         SchedulerNode schedulerNode = schedulerNodes.get(event.getRemovedRMNode().getNodeID().getHost());
-        if (schedulerNode.getNodeID().equals(event.getRemovedRMNode().getNodeID())) {
+        if (schedulerNode != null && schedulerNode.getNodeID().equals(event.getRemovedRMNode().getNodeID())) {
             schedulerNodes.remove(schedulerNode.getNodeID().getHost());
         }
     }

myriad-scheduler/src/main/java/com/ebay/myriad/scheduler/event/handlers/ResourceOffersEventHandler.java

@@ -22,6 +22,7 @@
 import com.ebay.myriad.state.SchedulerState;
 import com.lmax.disruptor.EventHandler;
 
+import java.util.Iterator;
 import org.apache.commons.collections.CollectionUtils;
 import org.apache.mesos.Protos;
 import org.apache.mesos.Protos.Offer;
@@ -75,7 +76,8 @@ public void onEvent(ResourceOffersEvent event, long sequence,
         try {
             Set<Protos.TaskID> pendingTasks = schedulerState.getPendingTaskIds();
             if (CollectionUtils.isNotEmpty(pendingTasks)) {
-                for (Offer offer : offers) {
+                for (Iterator<Offer> iterator = offers.iterator(); iterator.hasNext();) {
+                    Offer offer = iterator.next();
                     boolean offerMatch = false;
                     Protos.TaskID launchedTaskId = null;
                     for (Protos.TaskID pendingTaskId : pendingTasks) {
@@ -100,6 +102,7 @@ public void onEvent(ResourceOffersEvent event, long sequence,
                             taskToLaunch.setHostname(offer.getHostname());
                             taskToLaunch.setSlaveId(offer.getSlaveId());
                             offerMatch = true;
+                            iterator.remove(); // remove the used offer from offers list
                             break;
                         }
                     }

myriad-scheduler/src/main/java/com/ebay/myriad/scheduler/fgs/OfferLifecycleManager.java

@@ -2,7 +2,6 @@
 
 import com.ebay.myriad.scheduler.MyriadDriver;
 import java.util.HashMap;
-import java.util.List;
 import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
 

myriad-scheduler/src/main/java/com/ebay/myriad/scheduler/yarn/interceptor/CompositeInterceptor.java

@@ -108,13 +108,13 @@ public void afterSchedulerEventHandled(SchedulerEvent event) {
   private NodeId getNodeIdForSchedulerEvent(SchedulerEvent event) {
       switch (event.getType()) {
         case NODE_ADDED:
-          return ((NodeAddedSchedulerEvent)event).getAddedRMNode().getNodeID();
+          return ((NodeAddedSchedulerEvent) event).getAddedRMNode().getNodeID();
         case NODE_REMOVED:
-          return ((NodeRemovedSchedulerEvent)event).getRemovedRMNode().getNodeID();
+          return ((NodeRemovedSchedulerEvent) event).getRemovedRMNode().getNodeID();
         case NODE_UPDATE:
-          return ((NodeUpdateSchedulerEvent)event).getRMNode().getNodeID();
+          return ((NodeUpdateSchedulerEvent) event).getRMNode().getNodeID();
         case NODE_RESOURCE_UPDATE:
-          return ((NodeResourceUpdateSchedulerEvent)event).getRMNode().getNodeID();
+          return ((NodeResourceUpdateSchedulerEvent) event).getRMNode().getNodeID();
       }
       return null;
     }

myriad-scheduler/src/main/java/com/ebay/myriad/scheduler/yarn/interceptor/YarnSchedulerInterceptor.java

@@ -16,6 +16,9 @@
  */
 public interface YarnSchedulerInterceptor {
 
+    /**
+     * Filters the method callbacks.
+     */
     interface CallBackFilter {
       /**
        * Method to determine if any other methods in {@link YarnSchedulerInterceptor}

myriad-scheduler/src/test/java/com/ebay/myriad/scheduler/TestMyriadScheduler.java

@@ -1,11 +1,7 @@
 package com.ebay.myriad.scheduler;
 
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertNotNull;
-
-import com.ebay.myriad.scheduler.fgs.NodeStore;
+import com.ebay.myriad.scheduler.yarn.MyriadFairScheduler;
 import java.io.IOException;
-
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.yarn.conf.YarnConfiguration;
 import org.apache.hadoop.yarn.event.AsyncDispatcher;
@@ -22,9 +18,7 @@
 import org.junit.Before;
 import org.junit.Test;
 
-import com.ebay.myriad.Main;
-import com.ebay.myriad.scheduler.yarn.MyriadFairScheduler;
-import com.google.inject.Injector;
+import static org.junit.Assert.assertEquals;
 
 /**
  * Tests myriad scheduler.
@@ -79,11 +73,7 @@ public Configuration createConfiguration() {
   }
 
   @Test
-  public void testNodeStore() throws Exception {
-    Injector injector = Main.getInjector();
-    // TODO (Kannan Rajah) Find a better way to inject instances
-    NodeStore nodeStore = injector.getInstance(NodeStore.class);
-
+  public void testClusterMemory() throws Exception {
     // Add a node
     RMNode node1 =
         MockNodes
@@ -92,8 +82,6 @@ public void testNodeStore() throws Exception {
     scheduler.handle(nodeEvent1);
     assertEquals(1024, scheduler.getClusterResource().getMemory());
 
-    assertNotNull(nodeStore.getNode("127.0.0.1"));
-
     // Add another node
     RMNode node2 =
         MockNodes.newNodeInfo(1, Resources.createResource(512), 2, "127.0.0.2");