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

Code before:
         verify(mockZkClient).propagateLogDirEvent(config.brokerId)
      }
    } finally {
      Utils.tryAll(util.Arrays.asList[Callable[Void]](
        () => {
          replicaManager.shutdown(checkpointHW = false)
          null
        },
        () => {
          try {
            logManager.shutdown()
          } catch {
            case _: Exception =>
          }
          null
        },
        () => {
          quotaManager.shutdown()
          null
        }
      ))
    }
  }

  @Test
  def testReplicaManagerDontGenerateEmptyTopicPartition(): Unit = {
    // Given
    val props = TestUtils.createBrokerConfig(1, TestUtils.MockZkConnect, logDirCount = 2)
    val config = KafkaConfig.fromProps(props)
    val logDirFiles = config.logDirs.map(new File(_))
    val logDirFailureChannel = new LogDirFailureChannel(config.logDirs.size)
    val logManager = TestUtils.createLogManager(logDirFiles, defaultConfig = new LogConfig(new Properties()), time = time)
    val mockZkClient = mock(classOf[KafkaZkClient])
    val replicaManager = new ReplicaManager(
      metrics = metrics,
      config = config,
      time = time,
      scheduler = time.scheduler,
      logManager = logManager,
      quotaManagers = quotaManager,
      metadataCache = MetadataCache.zkMetadataCache(config.brokerId, config.interBrokerProtocolVersion),
      logDirFailureChannel = logDirFailureChannel,
      alterPartitionManager = alterPartitionManager,
      threadNamePrefix = Option(this.getClass.getName),
      zkClient = Option(mockZkClient),
    )
    try {
      logManager.startup(Set.empty[String])
      replicaManager.startup()

      // When

Review comment (suggestion):
It might be easier to test this in `DescribeLogDirsRequestTest` by setting up 2 topics with 2 topic partitions each and querying for both partitions of either one of the topics.

Code after:
    assertTrue(partition.isInstanceOf[HostedPartition.Online],
      s"Expected ${topicIdPartition} to be in state: HostedPartition.Online. But was in state: ${partition}")
    val hostedPartition = partition.asInstanceOf[HostedPartition.Online]
    assertTrue(hostedPartition.partition.log.isDefined,
      s"Expected ${topicIdPartition} to have a log set in ReplicaManager, but it did not.")
    assertTrue(hostedPartition.partition.log.get.topicId.isDefined,
      s"Expected the log for ${topicIdPartition} to topic ID set in LogManager, but it did not.")
    assertEquals(topicIdPartition.topicId(), hostedPartition.partition.log.get.topicId.get)
    assertEquals(topicIdPartition.topicPartition(), hostedPartition.partition.topicPartition)
  }

  def verifyPartitionIsOffline(
    replicaManager: ReplicaManager,
    topicIdPartition: TopicIdPartition
  ): Unit = {
    val partition = replicaManager.getPartition(topicIdPartition.topicPartition())
    assertEquals(HostedPartition.None, partition, s"Expected ${topicIdPartition} to be offline, but it was: ${partition}")
  }

  @Test
  def testFullLairDuringKRaftMigration(): Unit = {
    val replicaManager = setupReplicaManagerForKRaftMigrationTest()
    try {
      val becomeLeaderRequest = LogManagerTest.createLeaderAndIsrRequestForStrayDetection(
        Seq(foo0, foo1, bar0), Seq(3, 4, 3))
      replicaManager.becomeLeaderOrFollower(1, becomeLeaderRequest, (_, _) => ())
      verifyPartitionIsOnlineAndHasId(replicaManager, foo0)
      verifyPartitionIsOnlineAndHasId(replicaManager, foo1)
      verifyPartitionIsOnlineAndHasId(replicaManager, bar0)
    } finally {
      replicaManager.shutdown(checkpointHW = false)
    }
  }

  @Test
  def testFullLairDuringKRaftMigrationRemovesOld(): Unit = {
    val replicaManager = setupReplicaManagerForKRaftMigrationTest()
    try {
      val becomeLeaderRequest1 = LogManagerTest.createLeaderAndIsrRequestForStrayDetection(
        Seq(foo0, foo1, bar0), Seq(3, 4, 3))
      replicaManager.becomeLeaderOrFollower(1, becomeLeaderRequest1, (_, _) => ())
      val becomeLeaderRequest2 = LogManagerTest.createLeaderAndIsrRequestForStrayDetection(
        Seq(bar0), Seq(3, 4, 3))
      replicaManager.becomeLeaderOrFollower(2, becomeLeaderRequest2, (_, _) => ())

      verifyPartitionIsOffline(replicaManager, foo0)
      verifyPartitionIsOffline(replicaManager, foo1)
      verifyPartitionIsOnlineAndHasId(replicaManager, bar0)
    } finally {
      replicaManager.shutdown(checkpointHW = false)
    }