You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by ch...@apache.org on 2026/08/11 03:11:49 UTC
(kafka) branch 4.4 updated: KAFKA-18628 Add node.id to the metrics reporter config override (#23097)
This is an automated email from the ASF dual-hosted git repository.
chia7712 pushed a commit to branch 4.4
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/4.4 by this push:
new d2e840e6e92 KAFKA-18628 Add node.id to the metrics reporter config override (#23097)
d2e840e6e92 is described below
commit d2e840e6e9250fa44bf0a7f5220b28f6f455422f
Author: Ming-Yen Chung <mi...@gmail.com>
AuthorDate: Tue Aug 11 11:07:42 2026 +0800
KAFKA-18628 Add node.id to the metrics reporter config override (#23097)
`DynamicMetricReporterState` overrides only `broker.id` when configuring
metrics reporters. Since `broker.id` will no longer be passed to plugins
from 5.0 (KIP-1232), pass `node.id` alongside it.
This also fixes an inconsistency: reconfigurables are handed
`KafkaConfig#valuesFromThisConfig`, so a reporter recreated by a dynamic
update saw `node.id` as an `Integer` while `broker.id` was a `String`.
Both are now passed as strings, matching what reporters see at startup.
Raised in
https://github.com/apache/kafka/pull/22977#issuecomment-5167693531.
Reviewers: Ken Huang <s7...@gmail.com>, Chia-Ping Tsai
<ch...@gmail.com>
---
.../scala/kafka/server/DynamicBrokerConfig.scala | 8 +++-
.../kafka/server/DynamicBrokerConfigTest.scala | 44 ++++++++++++++++++++++
2 files changed, 50 insertions(+), 2 deletions(-)
diff --git a/core/src/main/scala/kafka/server/DynamicBrokerConfig.scala b/core/src/main/scala/kafka/server/DynamicBrokerConfig.scala
index 82081d5ec6c..f7a31a28827 100755
--- a/core/src/main/scala/kafka/server/DynamicBrokerConfig.scala
+++ b/core/src/main/scala/kafka/server/DynamicBrokerConfig.scala
@@ -36,7 +36,7 @@ import org.apache.kafka.common.utils.internals.BufferSupplier
import org.apache.kafka.common.utils.Utils
import org.apache.kafka.common.utils.internals.ConfigUtils
import org.apache.kafka.network.SocketServer
-import org.apache.kafka.raft.KafkaRaftClient
+import org.apache.kafka.raft.{KRaftConfigs, KafkaRaftClient}
import org.apache.kafka.server.{DynamicThreadPool, ProcessRole}
import org.apache.kafka.server.common.{ApiMessageAndVersion, DirectoryEventHandler}
import org.apache.kafka.server.config.{BrokerReconfigurable => JBrokerReconfigurable, DynamicConfig, DynamicProducerStateManagerConfig, ServerConfigs, ServerLogConfigs, DynamicBrokerConfig => JDynamicBrokerConfig}
@@ -741,7 +741,11 @@ class DynamicMetricsReporters(brokerId: Int, config: KafkaConfig, metrics: Metri
class DynamicMetricReporterState(brokerId: Int, config: KafkaConfig, metrics: Metrics, clusterId: String) {
private[server] val dynamicConfig = config.dynamicConfig
- private val propsOverride = Map[String, AnyRef](ServerConfigs.BROKER_ID_CONFIG -> brokerId.toString)
+ // broker.id will no longer be passed from 5.0 (KIP-1232).
+ private val propsOverride = Map[String, AnyRef](
+ ServerConfigs.BROKER_ID_CONFIG -> brokerId.toString,
+ KRaftConfigs.NODE_ID_CONFIG -> brokerId.toString
+ )
private[server] val currentReporters = mutable.Map[String, MetricsReporter]()
createReporters(config, clusterId, metricsReporterClasses(dynamicConfig.currentKafkaConfig.values()).asJava,
Collections.emptyMap[String, Object])
diff --git a/core/src/test/scala/unit/kafka/server/DynamicBrokerConfigTest.scala b/core/src/test/scala/unit/kafka/server/DynamicBrokerConfigTest.scala
index 08c1e6737f8..ae912fcaa47 100755
--- a/core/src/test/scala/unit/kafka/server/DynamicBrokerConfigTest.scala
+++ b/core/src/test/scala/unit/kafka/server/DynamicBrokerConfigTest.scala
@@ -699,6 +699,40 @@ class DynamicBrokerConfigTest {
assertTrue(m.currentReporters.isEmpty)
}
+ @Test
+ def testMetricReportersAreConfiguredWithBothIdConfigs(): Unit = {
+ val nodeId = 0
+ val reporterName = classOf[TestConfigCapturingReporter].getName
+ // createBrokerConfig only sets node.id, so broker.id is only visible through the synonym mechanism
+ val origProps = TestUtils.createBrokerConfig(nodeId)
+ origProps.put(MetricConfigs.METRIC_REPORTER_CLASSES_CONFIG, reporterName)
+
+ val config = KafkaConfig(origProps)
+ config.dynamicConfig.initialize(None)
+ val m = new DynamicMetricsReporters(nodeId, config, mock(classOf[Metrics]), "clusterId")
+ config.dynamicConfig.addReconfigurable(m)
+
+ def updateReporters(reporterNames: String): Unit = {
+ val props = new Properties()
+ props.put(MetricConfigs.METRIC_REPORTER_CLASSES_CONFIG, reporterNames)
+ config.dynamicConfig.updateDefaultConfig(props)
+ }
+
+ def assertBothIdsArePassedAsStrings(): Unit = {
+ val configs = m.currentReporters(reporterName).asInstanceOf[TestConfigCapturingReporter].configs
+ assertEquals(nodeId.toString, configs.get(KRaftConfigs.NODE_ID_CONFIG))
+ assertEquals(nodeId.toString, configs.get(ServerConfigs.BROKER_ID_CONFIG))
+ }
+
+ assertBothIdsArePassedAsStrings()
+
+ // a dynamic update recreates the reporter with the parsed config values, where the ids are integers
+ updateReporters("")
+ assertTrue(m.currentReporters.isEmpty)
+ updateReporters(reporterName)
+ assertBothIdsArePassedAsStrings()
+ }
+
@Test
def testDynamicLogLocalRetentionMsConfig(): Unit = {
val props = TestUtils.createBrokerConfig(0, port = 8181)
@@ -1336,6 +1370,16 @@ class TestDynamicThreadPool extends BrokerReconfigurable {
}
}
+class TestConfigCapturingReporter extends MetricsReporter {
+ var configs: util.Map[String, _] = _
+
+ override def configure(configs: util.Map[String, _]): Unit = this.configs = configs
+ override def init(metrics: util.List[KafkaMetric]): Unit = {}
+ override def metricChange(metric: KafkaMetric): Unit = {}
+ override def metricRemoval(metric: KafkaMetric): Unit = {}
+ override def close(): Unit = {}
+}
+
class TestExporterOnly extends MetricsReporter with ClientTelemetryExporterProvider {
override def configure(configs: util.Map[String, _]): Unit = {}
override def init(metrics: util.List[KafkaMetric]): Unit = {}