Repository: apache/kafka
File: core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala (Scala)

Code before:
    val tpId = new TopicIdPartition(topicId, tp)

    val props = TestUtils.createBrokerConfig(localId)
    val path1 = TestUtils.tempRelativeDir("data").getAbsolutePath
    val path2 = TestUtils.tempRelativeDir("data2").getAbsolutePath
    val path3 = TestUtils.tempRelativeDir("data3").getAbsolutePath
    props.put("log.dirs", Seq(path1, path2, path3).mkString(","))
    val config = KafkaConfig.fromProps(props)
    val mockLogMgr = TestUtils.createLogManager(config.logDirs.asScala.map(new File(_)), cleanerConfig = new CleanerConfig(true))
    mockLogMgr.startup(Set())
    val replicaManager = new ReplicaManager(
      metrics = metrics,
      config = config,
      time = time,
      scheduler = new MockScheduler(time),
      logManager = mockLogMgr,
      quotaManagers = quotaManager,
      metadataCache = metadataCache,
      logDirFailureChannel = new LogDirFailureChannel(config.logDirs.size),
      alterPartitionManager = alterPartitionManager,
      addPartitionsToTxnManager = Some(addPartitionsToTxnManager))

    try {
      val spiedPartition = spy(Partition(tpId, time, replicaManager))

      val blockPromotion = new AtomicBoolean(false)
      doAnswer { invocation =>
        if (blockPromotion.get) {
          false
        } else {
          invocation.callRealMethod()
        }
      }.when(spiedPartition).maybeReplaceCurrentWithFutureReplica()

      replicaManager.addOnlinePartition(tp, spiedPartition)

      val leaderDelta = topicsCreateDelta(localId, isStartIdLeader = true, partitions = List(0, 1), List.empty, topic, topicIds(topic))
      val leaderImage = imageFromTopics(leaderDelta.apply())
      replicaManager.applyDelta(leaderDelta, leaderImage)

      // Move the replica to the second log directory.
      val partition = replicaManager.getPartitionOrException(tp)
      val firstLogDir = partition.log.get.dir.getParentFile
      val newReplicaFolder = replicaManager.logManager.liveLogDirs.filterNot(_ == firstLogDir).head
      replicaManager.alterReplicaLogDirs(Map(tp -> newReplicaFolder.getAbsolutePath))

      // Prevent promotion of future replica
      blockPromotion.set(true)

      // Make sure the future log is created with the correct topic ID.
      val futureLog = replicaManager.futureLocalLogOrException(tp)

Review comment (suggestion):
will the following changes be sufficient?

```scala
      val newReplicaFolder = replicaManager.logManager.liveLogDirs.filterNot(_ == firstLogDir).head
      // Prevent promotion of future replica
      doReturn(false).when(spiedPartition).maybeReplaceCurrentWithFutureReplica()

      replicaManager.alterReplicaLogDirs(Map(tp -> newReplicaFolder.getAbsolutePath))
```

Code after:
    val tpId = new TopicIdPartition(topicId, tp)

    val props = TestUtils.createBrokerConfig(localId)
    val path1 = TestUtils.tempRelativeDir("data").getAbsolutePath
    val path2 = TestUtils.tempRelativeDir("data2").getAbsolutePath
    val path3 = TestUtils.tempRelativeDir("data3").getAbsolutePath
    props.put("log.dirs", Seq(path1, path2, path3).mkString(","))
    val config = KafkaConfig.fromProps(props)
    val mockLogMgr = TestUtils.createLogManager(config.logDirs.asScala.map(new File(_)), cleanerConfig = new CleanerConfig(true))
    mockLogMgr.startup(Set())
    val replicaManager = new ReplicaManager(
      metrics = metrics,
      config = config,
      time = time,
      scheduler = new MockScheduler(time),
      logManager = mockLogMgr,
      quotaManagers = quotaManager,
      metadataCache = metadataCache,
      logDirFailureChannel = new LogDirFailureChannel(config.logDirs.size),
      alterPartitionManager = alterPartitionManager,
      addPartitionsToTxnManager = Some(addPartitionsToTxnManager))

    try {
      val spiedPartition = spy(Partition(tpId, time, replicaManager))

      // Prevent promotion of future replica
      val blockPromotion = new AtomicBoolean(true)
      doAnswer { invocation =>
        if (blockPromotion.compareAndSet(true, false)) {
          false
        } else {
          invocation.callRealMethod()
        }
      }.when(spiedPartition).maybeReplaceCurrentWithFutureReplica()

      replicaManager.addOnlinePartition(tp, spiedPartition)

      val leaderDelta = topicsCreateDelta(localId, isStartIdLeader = true, partitions = List(0, 1), List.empty, topic, topicIds(topic))
      val leaderImage = imageFromTopics(leaderDelta.apply())
      replicaManager.applyDelta(leaderDelta, leaderImage)

      // Move the replica to the second log directory.
      val partition = replicaManager.getPartitionOrException(tp)
      val firstLogDir = partition.log.get.dir.getParentFile
      val newReplicaFolder = replicaManager.logManager.liveLogDirs.filterNot(_ == firstLogDir).head
      replicaManager.alterReplicaLogDirs(Map(tp -> newReplicaFolder.getAbsolutePath))

      // Make sure the future log is created with the correct topic ID.
      val futureLog = replicaManager.futureLocalLogOrException(tp)
      assertEquals(Optional.of(topicId), futureLog.topicId)