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

Code before:
    override protected def latestEpoch(topicPartition: TopicPartition): Optional[Integer] = Optional.of(0)

    override protected def logStartOffset(topicPartition: TopicPartition): Long = 1

    override protected def logEndOffset(topicPartition: TopicPartition): Long = 1

    override protected def endOffsetForEpoch(topicPartition: TopicPartition, epoch: Int): Optional[OffsetAndEpoch] = Optional.of(new OffsetAndEpoch(1, 0))
  }

  @Test
  def testMetricsClassName(): Unit = {
    val config = mock(classOf[KafkaConfig])
    val replicaManager = mock(classOf[ReplicaManager])
    val quotaManager = mock(classOf[ReplicationQuotaManager])
    val brokerTopicStats   = new BrokerTopicStats()
    val directoryEventHandler = DirectoryEventHandler.NOOP
    val metrics = new Metrics()
    val time = new MockTime()
    val metadataVersionSupplier = () => MetadataVersion.LATEST_PRODUCTION
    val brokerEpochSupplier = () => 1L

    val replicaAlterLogDirsManager = new ReplicaAlterLogDirsManager(config, replicaManager, quotaManager, brokerTopicStats, directoryEventHandler)
    val replicaFetcherManager = new ReplicaFetcherManager(config, replicaManager, metrics, time, quotaManager, metadataVersionSupplier, brokerEpochSupplier)

    assertEquals("ReplicaAlterLogDirsManager", replicaAlterLogDirsManager.getMetricsClassName)
    assertEquals("ReplicaFetcherManager", replicaFetcherManager.getMetricsClassName)
  }
}


Review comment (suggestion):
Maybe you can use `KafkaYammerMetrics.defaultRegistry()` to get the target metrics.

Code after:
    override protected def latestEpoch(topicPartition: TopicPartition): Optional[Integer] = Optional.of(0)

    override protected def logStartOffset(topicPartition: TopicPartition): Long = 1

    override protected def logEndOffset(topicPartition: TopicPartition): Long = 1

    override protected def endOffsetForEpoch(topicPartition: TopicPartition, epoch: Int): Optional[OffsetAndEpoch] = Optional.of(new OffsetAndEpoch(1, 0))
  }

  @Test
  def testMetricsClassName(): Unit = {
    val registry = KafkaYammerMetrics.defaultRegistry()
    val config = mock(classOf[KafkaConfig])
    val replicaManager = mock(classOf[ReplicaManager])
    val quotaManager = mock(classOf[ReplicationQuotaManager])
    val brokerTopicStats   = new BrokerTopicStats()
    val directoryEventHandler = DirectoryEventHandler.NOOP
    val metrics = new Metrics()
    val time = new MockTime()
    val metadataVersionSupplier = () => MetadataVersion.LATEST_PRODUCTION
    val brokerEpochSupplier = () => 1L

    val _ = new ReplicaAlterLogDirsManager(config, replicaManager, quotaManager, brokerTopicStats, directoryEventHandler)
    val _ = new ReplicaFetcherManager(config, replicaManager, metrics, time, quotaManager, metadataVersionSupplier, brokerEpochSupplier)
    val existReplicaAlterLogDirsManager = registry.allMetrics.entrySet().stream().filter(metric => metric.getKey.getType == "ReplicaAlterLogDirsManager").findFirst()
    val existReplicaFetcherManager = registry.allMetrics.entrySet().stream().filter(metric => metric.getKey.getType == "ReplicaFetcherManager").findFirst()
    assertTrue(existReplicaAlterLogDirsManager.isPresent)
    assertTrue(existReplicaFetcherManager.isPresent)
  }
}