Repository: apache/kafka
File: core/src/test/scala/integration/kafka/server/DynamicBrokerReconfigurationTest.scala (Scala)

Code before:
    assertEquals(pollingInterval, this.pollingInterval)
  }

  def verifyMetricValue(name: String, group: String): Unit = {
    val matchingMetrics = kafkaMetrics.filter(metric => metric.metricName.name == name && metric.metricName.group == group)
    assertTrue(matchingMetrics.nonEmpty, "Metric not found")
    val total = matchingMetrics.foldLeft(0.0)((total, metric) => total + metric.metricValue.asInstanceOf[Double])
    assertTrue(total > 0.0, "Invalid metric value " + total + " for name " + name + " , group " + group)
  }
}

object TestNumReplicaFetcherMetricsReporter {
  val testReporters = new ConcurrentLinkedQueue[TestNumReplicaFetcherMetricsReporter]()
  val configuredBrokers = mutable.Set[Int]()

  def waitForReporters(count: Int): List[TestNumReplicaFetcherMetricsReporter] = {
    TestUtils.waitUntilTrue(() => testReporters.size == count, msg = "Metrics reporters size not matched. Expected: " + count + ", actual: " + testReporters.size())

    val reporters = testReporters.asScala.toList
    TestUtils.waitUntilTrue(() => reporters.forall(_.configureCount == 1), msg = "Metrics reporters not configured")
    reporters
  }
}


class TestNumReplicaFetcherMetricsReporter extends MetricsReporter with Reconfigurable with Closeable with ClusterResourceListener {
  import TestNumReplicaFetcherMetricsReporter._
  @volatile var initializeCount = 0
  @volatile var contextChangeCount = 0
  @volatile var configureCount = 0
  @volatile var reconfigureCount = 0
  @volatile var closeCount = 0
  @volatile var clusterUpdateCount = 0
  @volatile var numFetchers: Int = 1
  testReporters.add(this)

  override def contextChange(metricsContext: MetricsContext): Unit = {
    contextChangeCount += 1
  }

  override def init(metrics: util.List[KafkaMetric]): Unit = {
    assertTrue(contextChangeCount > 0, "contextChange must be called before init")
    initializeCount += 1
  }

  override def configure(configs: util.Map[String, _]): Unit = {
    configuredBrokers += configs.get(ServerConfigs.BROKER_ID_CONFIG).toString.toInt
    configureCount += 1
    numFetchers = configs.get(ReplicationConfigs.NUM_REPLICA_FETCHERS_CONFIG).toString.toInt
  }


Review comment (suggestion):
The reporter's functionality is excessive. All we need to verify is that is can "see" the latest configurations at startup. Therefore, only `reconfigureCount` is relevant for the new test case

Code after:
    assertEquals(pollingInterval, this.pollingInterval)
  }

  def verifyMetricValue(name: String, group: String): Unit = {
    val matchingMetrics = kafkaMetrics.filter(metric => metric.metricName.name == name && metric.metricName.group == group)
    assertTrue(matchingMetrics.nonEmpty, "Metric not found")
    val total = matchingMetrics.foldLeft(0.0)((total, metric) => total + metric.metricValue.asInstanceOf[Double])
    assertTrue(total > 0.0, "Invalid metric value " + total + " for name " + name + " , group " + group)
  }
}

object TestNumReplicaFetcherMetricsReporter {
  val testReporters = new ConcurrentLinkedQueue[TestNumReplicaFetcherMetricsReporter]()

  def waitForReporters(count: Int): List[TestNumReplicaFetcherMetricsReporter] = {
    TestUtils.waitUntilTrue(() => testReporters.size == count, msg = "Metrics reporters size not matched. Expected: " + count + ", actual: " + testReporters.size())

    val reporters = testReporters.asScala.toList
    TestUtils.waitUntilTrue(() => reporters.forall(_.configureCount == 1), msg = "Metrics reporters not configured")
    reporters
  }
}


class TestNumReplicaFetcherMetricsReporter extends MetricsReporter {
  import TestNumReplicaFetcherMetricsReporter._
  @volatile var configureCount = 0
  @volatile var reconfigureCount = 0
  @volatile var numFetchers: Int = 1
  testReporters.add(this)

  override def init(metrics: util.List[KafkaMetric]): Unit = {
  }

  override def configure(configs: util.Map[String, _]): Unit = {
    configureCount += 1
    numFetchers = configs.get(ReplicationConfigs.NUM_REPLICA_FETCHERS_CONFIG).toString.toInt
  }

  override def metricChange(metric: KafkaMetric): Unit = {
  }

  override def metricRemoval(metric: KafkaMetric): Unit = {
  }

  override def reconfigurableConfigs(): util.Set[String] = {
    util.Set.of(ReplicationConfigs.NUM_REPLICA_FETCHERS_CONFIG)
  }

  override def validateReconfiguration(configs: util.Map[String, _]): Unit = {
    val numFetchers = configs.get(ReplicationConfigs.NUM_REPLICA_FETCHERS_CONFIG).toString.toInt