[SPARK-17759] [CORE] Avoid adding duplicate schedulables

core/src/main/scala/org/apache/spark/scheduler/Pool.scala

@@ -61,8 +61,12 @@ private[spark] class Pool(
 
   override def addSchedulable(schedulable: Schedulable) {
     require(schedulable != null)
-    schedulableQueue.add(schedulable)
-    schedulableNameToSchedulable.put(schedulable.name, schedulable)
+    if (schedulableNameToSchedulable.containsKey(schedulable.name)) {
+      logWarning(s"Schedulable: ${schedulable.name} already exists so duplicate is not created.")
+    } else {
+      schedulableQueue.add(schedulable)
+      schedulableNameToSchedulable.put(schedulable.name, schedulable)
+    }
     schedulable.parent = this
   }
 

core/src/test/resources/fairscheduler-duplicate-pools.xml

@@ -0,0 +1,42 @@
+<?xml version="1.0"?>
+<!--
+  ~ 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.
+  -->
+
+<allocations>
+    <allocations>
+        <pool name="default">
+            <minShare>0</minShare>
+            <weight>1</weight>
+            <schedulingMode>FIFO</schedulingMode>
+        </pool>
+        <pool name="default">
+            <minShare>0</minShare>
+            <weight>1</weight>
+            <schedulingMode>FIFO</schedulingMode>
+        </pool>
+        <pool name="duplicate_pool1">
+            <minShare>1</minShare>
+            <weight>1</weight>
+            <schedulingMode>FAIR</schedulingMode>
+        </pool>
+        <pool name="duplicate_pool1">
+            <minShare>2</minShare>
+            <weight>2</weight>
+            <schedulingMode>FAIR</schedulingMode>
+        </pool>
+    </allocations>
+</allocations>

core/src/test/scala/org/apache/spark/scheduler/PoolSuite.scala

@@ -79,9 +79,7 @@ class PoolSuite extends SparkFunSuite with LocalSparkContext {
    * algorithm properly orders the two scheduling pools.
    */
   test("Fair Scheduler Test") {
-    val xmlPath = getClass.getClassLoader.getResource("fairscheduler.xml").getFile()
-    val conf = new SparkConf().set(SCHEDULER_ALLOCATION_FILE, xmlPath)
-    sc = new SparkContext(LOCAL, APP_NAME, conf)
+    sc = createSparkContext("fairscheduler.xml")
     val taskScheduler = new TaskSchedulerImpl(sc)
 
     val rootPool = new Pool("", FAIR, 0, 0)
@@ -295,9 +293,7 @@ class PoolSuite extends SparkFunSuite with LocalSparkContext {
 
   test("Fair Scheduler should build fair scheduler when " +
     "valid spark.scheduler.allocation.file property is set") {
-    val xmlPath = getClass.getClassLoader.getResource("fairscheduler-with-valid-data.xml").getFile()
-    val conf = new SparkConf().set(SCHEDULER_ALLOCATION_FILE, xmlPath)
-    sc = new SparkContext(LOCAL, APP_NAME, conf)
+    sc = createSparkContext("fairscheduler-with-valid-data.xml")
 
     val rootPool = new Pool("", SchedulingMode.FAIR, 0, 0)
     val schedulableBuilder = new FairSchedulableBuilder(rootPool, sc.conf)
@@ -336,6 +332,29 @@ class PoolSuite extends SparkFunSuite with LocalSparkContext {
     }
   }
 
+  test("Fair Scheduler should not create duplicate pool") {
+    // Load the scheduler pools from fairscheduler-duplicate-pools, which has 4 entries,
+    // but two are duplicates, and make sure that the duplicates are ignored.
+    sc = createSparkContext("fairscheduler-duplicate-pools.xml")
+
+    val rootPool = new Pool("", SchedulingMode.FAIR, 0, 0)
+    val schedulableBuilder = new FairSchedulableBuilder(rootPool, sc.conf)
+    schedulableBuilder.buildPools()
+
+    assert(rootPool.schedulableQueue.size === 2)
+    assert(rootPool.schedulableNameToSchedulable.size === 2)
+
+    verifyPool(rootPool, schedulableBuilder.DEFAULT_POOL_NAME, 0, 1, SchedulingMode.FIFO)
+    // The first pool specified is used if duplicate pools exist
+    verifyPool(rootPool, "duplicate_pool1", 1, 1, SchedulingMode.FAIR)
+  }
+
+  private def createSparkContext(fileName: String): SparkContext = {
+    val xmlPath = getClass.getClassLoader.getResource(fileName).getFile()
+    val conf = new SparkConf().set(SCHEDULER_ALLOCATION_FILE, xmlPath)
+    new SparkContext(LOCAL, APP_NAME, conf)
+  }
+
   private def verifyPool(rootPool: Pool, poolName: String, expectedInitMinShare: Int,
                          expectedInitWeight: Int, expectedSchedulingMode: SchedulingMode): Unit = {
     val selectedPool = rootPool.getSchedulableByName(poolName)