From c11d2d7f110e2bfb40ce8a526fbfcc3d783b561a Mon Sep 17 00:00:00 2001
From: Tom Arnfeld <tarnfeld@me.com>
Date: Sat, 11 Oct 2014 02:01:06 +0100
Subject: [PATCH 1/2] Added support for strict role validation

Sometimes the user might want to restrict Hadoop to only launching tasks on top
of resources for the given role, and not accept offers for "*"
---
 configuration.md                                     |  8 ++++++++
 .../org/apache/hadoop/mapred/ResourcePolicy.java     | 12 ++++++++++++
 2 files changed, 20 insertions(+)

diff --git a/configuration.md b/configuration.md
index ab66d78..24f9bfa 100644
--- a/configuration.md
+++ b/configuration.md
@@ -133,6 +133,14 @@ default values.
       Mesos reservations. Consult the Mesos documentation for details.
     </description>
   </property>
+  <property>
+    <name>mapred.mesos.role.strict</name>
+    <value>false</value>
+    <description>
+      Force the framework to only ever accept resource offers that are of the
+      role configured in "mapred.mesos.role".
+    </description>
+  </property>
 
   <!-- If you're using a custom Mesos Containerizer -->
   <property>
diff --git a/src/main/java/org/apache/hadoop/mapred/ResourcePolicy.java b/src/main/java/org/apache/hadoop/mapred/ResourcePolicy.java
index e6b10ea..4387a44 100644
--- a/src/main/java/org/apache/hadoop/mapred/ResourcePolicy.java
+++ b/src/main/java/org/apache/hadoop/mapred/ResourcePolicy.java
@@ -260,6 +260,18 @@ public void resourceOffers(SchedulerDriver schedulerDriver,
           }
         }
 
+        // Verify the resource roles are what we need
+        if (scheduler.conf.getBoolean("mapred.mesos.role.strict", false)) {
+          String expectedRole = scheduler.conf.get("mapred.mesos.role", "*");
+          if (!cpuRole.equals(expectedRole) ||
+              !memRole.equals(expectedRole) ||
+              !diskRole.equals(expectedRole) ||
+              !portsRole.equals(expectedRole)) {
+            schedulerDriver.declineOffer(offer.getId());
+            continue;
+          }
+        }
+
         final boolean sufficient = computeSlots();
 
         double taskCpus = (mapSlots + reduceSlots) * slotCpus + containerCpus;

From c5c9ea232bcbab189c6d34d6f7b919514c346271 Mon Sep 17 00:00:00 2001
From: Tom Arnfeld <tarnfeld@me.com>
Date: Sat, 11 Oct 2014 02:07:39 +0100
Subject: [PATCH 2/2] Added extra logging

---
 src/main/java/org/apache/hadoop/mapred/ResourcePolicy.java | 2 ++
 1 file changed, 2 insertions(+)

diff --git a/src/main/java/org/apache/hadoop/mapred/ResourcePolicy.java b/src/main/java/org/apache/hadoop/mapred/ResourcePolicy.java
index 4387a44..c2d4aff 100644
--- a/src/main/java/org/apache/hadoop/mapred/ResourcePolicy.java
+++ b/src/main/java/org/apache/hadoop/mapred/ResourcePolicy.java
@@ -267,6 +267,8 @@ public void resourceOffers(SchedulerDriver schedulerDriver,
               !memRole.equals(expectedRole) ||
               !diskRole.equals(expectedRole) ||
               !portsRole.equals(expectedRole)) {
+            LOG.info("Declining offer with invalid role " + expectedRole);
+
             schedulerDriver.declineOffer(offer.getId());
             continue;
           }