Incorporating suggestions from @sameeragarwal. Removing objectives based replication strategy and constraint solver associate with it.

Author: shubhamchopra

core/src/main/scala/org/apache/spark/storage/BlockReplicationObjectives.scala

@@ -222,56 +222,3 @@ class BasicBlockReplicationPolicy
   }
 
 }
-
-@DeveloperApi
-class ObjectivesBasedReplicationPolicy
-  extends BlockReplicationPolicy
-    with Logging {
-  val objectives: Set[BlockReplicationObjective] = Set(
-    ReplicateToADifferentHost,
-    ReplicateBlockOutsideRack,
-    ReplicateBlockWithinRack,
-    NoTwoReplicasInSameRack
-  )
-
-  /**
-   * Method to prioritize a bunch of candidate peers of a block
-   *
-   * @param blockManagerId    Id of the current BlockManager for self identification
-   * @param peers             A list of peers of a BlockManager
-   * @param peersReplicatedTo Set of peers already replicated to
-   * @param blockId           BlockId of the block being replicated. This can be used as a source of
-   *                          randomness if needed.
-   * @param numReplicas Number of peers we need to replicate to
-   * @return A prioritized list of peers. Lower the index of a peer, higher its priority
-   */
-  override def prioritize(
-      blockManagerId: BlockManagerId,
-      peers: Seq[BlockManagerId],
-      peersReplicatedTo: mutable.HashSet[BlockManagerId],
-      blockId: BlockId,
-      numReplicas: Int): List[BlockManagerId] = {
-    val (optimalPeers, objectivesMet) = BlockReplicationOptimizer.getPeersToMeetObjectives(
-      objectives,
-      peers,
-      peersReplicatedTo,
-      blockId,
-      blockManagerId,
-      numReplicas
-    )
-    logDebug(s"BlockReplication objectives met : ${objectivesMet.mkString(", ")}")
-    logDebug(s"Optimal peers : ${optimalPeers.mkString(", ")}")
-    // pad optimal peers with random peers if we are short
-    val numRemainingPeers = numReplicas - optimalPeers.size
-    val remainingPeers = if (numRemainingPeers > 0) {
-      val r = new Random(blockId.hashCode)
-      BlockReplicationUtils.getRandomSample(
-        peers.filter(p => !optimalPeers.contains(p)),
-        numRemainingPeers,
-        r)
-    } else {
-      List.empty
-    }
-    optimalPeers.toList ++ remainingPeers
-  }
-}

core/src/main/scala/org/apache/spark/storage/BlockReplicationPolicy.scala

@@ -553,14 +553,3 @@ class BlockManagerBasicStrategyReplicationSuite extends BlockManagerReplicationB
     classOf[DummyTopologyMapper].getName)
 }
 
-class BlockManagerPrioritizerReplicationSuite extends BlockManagerReplicationBehavior {
-  override val conf: SparkConf = new SparkConf(false).set("spark.app.id", "test")
-  conf.set("spark.kryoserializer.buffer", "1m")
-  conf.set(
-    "spark.storage.replication.policy",
-    classOf[ObjectivesBasedReplicationPolicy].getName)
-  conf.set(
-    "spark.storage.replication.topologyMapper",
-    classOf[DummyTopologyMapper].getName
-  )
-}

core/src/test/scala/org/apache/spark/storage/BlockManagerReplicationSuite.scala

@@ -24,15 +24,15 @@ import org.scalatest.{BeforeAndAfter, Matchers}
 
 import org.apache.spark.{LocalSparkContext, SparkFunSuite}
 
-trait RandomBlockReplicationPolicyBehavior extends SparkFunSuite
+class RandomBlockReplicationPolicyBehavior extends SparkFunSuite
     with Matchers
     with BeforeAndAfter
     with LocalSparkContext {
 
   // Implicitly convert strings to BlockIds for test clarity.
   protected implicit def StringToBlockId(value: String): BlockId = new TestBlockId(value)
 
-  val replicationPolicy: BlockReplicationPolicy
+  val replicationPolicy: BlockReplicationPolicy = new RandomBlockReplicationPolicy
 
   val blockId = "test-block"
   /**
@@ -77,7 +77,9 @@ trait RandomBlockReplicationPolicyBehavior extends SparkFunSuite
   }
 }
 
-trait TopologyAwareBlockReplicationPolicyBehavior extends RandomBlockReplicationPolicyBehavior {
+class TopologyAwareBlockReplicationPolicyBehavior extends RandomBlockReplicationPolicyBehavior {
+  override val replicationPolicy = new BasicBlockReplicationPolicy
+
   test("All peers in the same rack") {
     val racks = Seq("/default-rack")
     val numBlockManager = 10
@@ -123,67 +125,3 @@ trait TopologyAwareBlockReplicationPolicyBehavior extends RandomBlockReplication
     }
   }
 }
-
-class RandomBlockReplicationPolicySuite extends RandomBlockReplicationPolicyBehavior {
-  override val replicationPolicy = new RandomBlockReplicationPolicy
-}
-
-class BasicBlockReplicationPolicySuite extends TopologyAwareBlockReplicationPolicyBehavior {
-  override val replicationPolicy = new BasicBlockReplicationPolicy
-}
-
-class ObjectivesBasedReplicationPolicySuite extends TopologyAwareBlockReplicationPolicyBehavior {
-  override val replicationPolicy = new ObjectivesBasedReplicationPolicy
-
-  val objectives: Set[BlockReplicationObjective] = Set(
-    ReplicateToADifferentHost,
-    ReplicateBlockOutsideRack,
-    ReplicateBlockWithinRack,
-    NoTwoReplicasInSameRack
-  )
-
-  test("peers are all in the same rack") {
-    val blockManagerIds = generateBlockManagerIds(10, List("Default-rack"))
-
-    val blockId = BlockId("test_block")
-
-    val candidateBMId = generateBlockManagerIds(1, List("Default-rack")).head
-
-    val numReplicas = 2
-
-    val (optimalPeers, objectivesMet) = BlockReplicationOptimizer.getPeersToMeetObjectives(
-      objectives,
-      blockManagerIds,
-      mutable.HashSet.empty,
-      blockId,
-      candidateBMId,
-      numReplicas)
-
-    logDebug(s"Optimal peers : ${optimalPeers}")
-    logDebug(s"Objectives met : ${objectivesMet}")
-    assert(optimalPeers.size == 1)
-    assert(objectivesMet.size == 3)
-  }
-
-  test("peers in 3 racks") {
-    val racks = List("/Rack1", "/Rack2", "/Rack3")
-    val blockManagerIds = generateBlockManagerIds(10, racks)
-    val candidateBMId = generateBlockManagerIds(1, racks).head
-    val blockId = BlockId("test_block")
-    val numReplicas = 2
-    val (optimalPeers, objectivesMet) = BlockReplicationOptimizer.getPeersToMeetObjectives(
-      objectives,
-      blockManagerIds,
-      mutable.HashSet.empty,
-      blockId,
-      candidateBMId,
-      numReplicas)
-
-    logDebug(s"Optimal peers : ${optimalPeers}")
-    logDebug(s"Objectives met : ${objectivesMet}")
-    assert(optimalPeers.size == 2)
-    assert(objectivesMet.size == 4)
-    assert(objectives.forall(_.isObjectiveMet(candidateBMId, optimalPeers)))
-  }
-
-}