Merge branch 'master' of github.com:mesos/spark into scala-2.10

core/src/main/scala/spark/scheduler/mesos/MesosSchedulerBackend.scala

@@ -43,8 +43,12 @@ private[spark] class MesosSchedulerBackend(
   // An ExecutorInfo for our tasks
   var execArgs: Array[Byte] = null
 
+  var classLoader: ClassLoader = null
+
   override def start() {
     synchronized {
+      classLoader = Thread.currentThread.getContextClassLoader
+
       new Thread("MesosSchedulerBackend driver") {
         setDaemon(true)
         override def run() {
@@ -114,14 +118,29 @@ private[spark] class MesosSchedulerBackend(
     return execArgs
   }
 
+  private def setClassLoader(): ClassLoader = {
+    val oldClassLoader = Thread.currentThread.getContextClassLoader
+    Thread.currentThread.setContextClassLoader(classLoader)
+    return oldClassLoader
+  }
+
+  private def restoreClassLoader(oldClassLoader: ClassLoader) {
+    Thread.currentThread.setContextClassLoader(oldClassLoader)
+  }
+
   override def offerRescinded(d: SchedulerDriver, o: OfferID) {}
 
   override def registered(d: SchedulerDriver, frameworkId: FrameworkID, masterInfo: MasterInfo) {
+    val oldClassLoader = setClassLoader()
+    try {
       logInfo("Registered as framework ID " + frameworkId.getValue)
       registeredLock.synchronized {
         isRegistered = true
         registeredLock.notifyAll()
       }
+    } finally {
+      restoreClassLoader(oldClassLoader)
+    }
   }
 
   def waitForRegister() {
@@ -142,6 +161,8 @@ private[spark] class MesosSchedulerBackend(
    * tasks are balanced across the cluster.
    */
   override def resourceOffers(d: SchedulerDriver, offers: JList[Offer]) {
+    val oldClassLoader = setClassLoader()
+    try {
       synchronized {
         // Build a big list of the offerable workers, and remember their indices so that we can
         // figure out which Offer to reply to for each worker
@@ -186,6 +207,9 @@ private[spark] class MesosSchedulerBackend(
           d.launchTasks(offers(i).getId, mesosTasks(i), filters)
         }
       }
+    } finally {
+      restoreClassLoader(oldClassLoader)
+    }
   }
 
   /** Helper function to pull out a resource from a Mesos Resources protobuf */
@@ -224,6 +248,8 @@ private[spark] class MesosSchedulerBackend(
   }
 
   override def statusUpdate(d: SchedulerDriver, status: TaskStatus) {
+    val oldClassLoader = setClassLoader()
+    try {
       val tid = status.getTaskId.getValue.toLong
       val state = TaskState.fromMesos(status.getState)
       synchronized {
@@ -236,11 +262,19 @@ private[spark] class MesosSchedulerBackend(
         }
       }
       scheduler.statusUpdate(tid, state, status.getData.asReadOnlyByteBuffer)
+    } finally {
+      restoreClassLoader(oldClassLoader)
+    }
   }
 
   override def error(d: SchedulerDriver, message: String) {
+    val oldClassLoader = setClassLoader()
+    try {
       logError("Mesos error: " + message)
       scheduler.error(message)
+    } finally {
+      restoreClassLoader(oldClassLoader)
+    }
   }
 
   override def stop() {
@@ -256,11 +290,16 @@ private[spark] class MesosSchedulerBackend(
   override def frameworkMessage(d: SchedulerDriver, e: ExecutorID, s: SlaveID, b: Array[Byte]) {}
 
   private def recordSlaveLost(d: SchedulerDriver, slaveId: SlaveID, reason: ExecutorLossReason) {
+    val oldClassLoader = setClassLoader()
+    try {
       logInfo("Mesos slave lost: " + slaveId.getValue)
       synchronized {
         slaveIdsWithExecutors -= slaveId.getValue
       }
       scheduler.executorLost(slaveId.getValue, reason)
+    } finally {
+      restoreClassLoader(oldClassLoader)
+    }
   }
 
   override def slaveLost(d: SchedulerDriver, slaveId: SlaveID) {