KAFKA-19642: Replace dynamicPerBrokerConfigs with dynamicDefaultConfigs #20405

jim0987795064 · 2025-08-25T13:59:30Z

Changes: Replace misused dynamicPerBrokerConfigs with
dynamicDefaultConfigs
Reasons: KRaft servers don't handle the cluser-level configs in
starting

from: https://github.com/apache/kafka/pull/18949/files#r2296809389

Reviewers: Jun Rao junrao@gmail.com, Jhen-Yung Hsu
jhenyunghsu@gmail.com, PoAn Yang payang@apache.org, Chia-Ping Tsai
chia7712@gmail.com

junrao

@jim0987795064 : Thanks for the PR. A couple of comments. Also, could we add a test case?

junrao · 2025-08-25T16:45:19Z

core/src/main/scala/kafka/server/DynamicBrokerConfig.scala

@@ -231,15 +230,15 @@ object DynamicBrokerConfig {
                    if (configRecord.resourceName().isEmpty) {
                      putOrRemoveIfNull(dynamicDefaultConfigs, configRecord.name(), configRecord.value())
                    } else if (configRecord.resourceName() == config.brokerId.toString) {
-                      putOrRemoveIfNull(dynamicPerBrokerConfigs, configRecord.name(), configRecord.value())
+                      putOrRemoveIfNull(dynamicDefaultConfigs, configRecord.name(), configRecord.value())


Hmm, this doesn't look right. If configRecord.resourceName() == config.brokerId.toString, it's a broker level config and it seems that we should use dynamicPerBrokerConfigs.

junrao · 2025-08-25T16:45:59Z

core/src/main/scala/kafka/server/DynamicBrokerConfig.scala

-          configHandler.processConfigChanges("", dynamicPerBrokerConfigs)
-          configHandler.processConfigChanges(config.brokerId.toString, dynamicPerBrokerConfigs)
+          configHandler.processConfigChanges("", dynamicDefaultConfigs)
+          configHandler.processConfigChanges(config.brokerId.toString, dynamicDefaultConfigs)


It seems that we should continue to use dynamicPerBrokerConfigs here?

@junrao
Thanks for pointing this out. I've corrected the misused dynamicDefaultConfigs and now use dynamicPerBrokerConfigs for broker-scoped updates.

jim0987795064 · 2025-08-25T17:22:05Z

@jim0987795064 : Thanks for the PR. A couple of comments. Also, could we add a test case?

Thanks for the feedback. I'm still working on adding the test case and will update the PR once it's ready.

chia7712 · 2025-08-25T19:38:29Z

Thanks for the feedback. I'm still working on adding the test case and will update the PR once it's ready.

Perhaps we could enhance testServersCanStartWithInvalidStaticConfigsAndValidDynamicConfigs to cover this case.

add a custom MetricsReporter to the newBroker. the MetricsReporter must rely on a dynamic server configuration, such as num.replica.fetchers
add num.replica.fetchers to the cluster-level config
if readDynamicBrokerConfigsFromSnapshot works correctly, the custom MetricsReporter should not be reconfigured after a restart.

Signed-off-by: PoAn Yang <payang@apache.org>

chia7712 · 2025-08-26T08:24:37Z

core/src/test/scala/integration/kafka/server/DynamicBrokerReconfigurationTest.scala

@@ -1599,31 +1622,37 @@ class TestMetricsReporter extends MetricsReporter with Reconfigurable with Close
  }

  override def reconfigurableConfigs(): util.Set[String] = {
-    util.Set.of(PollingIntervalProp)
+    util.Set.of(PollingIntervalProp, ReplicationConfigs.NUM_REPLICA_FETCHERS_CONFIG)


@FrankYang0529 Thanks for your help. Could you please consider creating an individual metrics reporter for the new test? Including too many test cases in TestMetricsReporter could make it harder to understand

Add another TestNumReplicaFetcherMetricsReporter for it.

Please revert the change of TestMetricsReporter

Yunyung

Thanks for the PR. One nitpick,

Yunyung · 2025-08-26T08:46:19Z

core/src/test/scala/integration/kafka/server/DynamicBrokerReconfigurationTest.scala

@@ -1568,6 +1589,7 @@ class TestMetricsReporter extends MetricsReporter with Reconfigurable with Close
  @volatile var closeCount = 0
  @volatile var clusterUpdateCount = 0
  @volatile var pollingInterval: Int = -1
+  @volatile var numFetchers: Int = 1


Maybe changing it to = 0 makes it clearer that it must be changed statically/dynamically.

Please ignore this.
For this reason: https://issues.apache.org/jira/browse/KAFKA-19645 , it is probably more reasonable to keep it as 1 instead of setting invalid value.

Yunyung · 2025-08-26T08:47:22Z

core/src/test/scala/integration/kafka/server/DynamicBrokerReconfigurationTest.scala

  }

  override def close(): Unit = {
    closeCount += 1
  }

-  def verifyState(reconfigureCount: Int, deleteCount: Int, pollingInterval: Int): Unit = {
+  def verifyState(reconfigureCount: Int, deleteCount: Int, pollingInterval: Int, numFetcher: Int = 1): Unit = {


Suggested change

def verifyState(reconfigureCount: Int, deleteCount: Int, pollingInterval: Int, numFetcher: Int = 1): Unit = {

def verifyState(reconfigureCount: Int, deleteCount: Int, pollingInterval: Int, numFetcher: Int): Unit = {

Yunyung · 2025-08-26T08:48:34Z

core/src/test/scala/integration/kafka/server/DynamicBrokerReconfigurationTest.scala

@@ -947,6 +947,7 @@ class DynamicBrokerReconfigurationTest extends QuorumTestHarness with SaslSetup
    // Add a new metrics reporter
    val newProps = new Properties
    newProps.put(TestMetricsReporter.PollingIntervalProp, "100")
+    newProps.put(ReplicationConfigs.NUM_REPLICA_FETCHERS_CONFIG, "1")


Should we change all verifyState calls in this method to numFetcher = 1? It's easier to understand, IMO.
Like reporters.foreach(_.verifyState(reconfigureCount = 0, deleteCount = 0, pollingInterval = 100, numFetcher=1))

Signed-off-by: PoAn Yang <payang@apache.org>

core/src/test/scala/integration/kafka/server/DynamicBrokerReconfigurationTest.scala

Signed-off-by: PoAn Yang <payang@apache.org>

junrao

@jim0987795064 : Thanks for the updated PR. Just a minor comment.

junrao · 2025-08-26T17:30:03Z

core/src/test/scala/integration/kafka/server/DynamicBrokerReconfigurationTest.scala

@@ -1634,6 +1655,71 @@ class TestMetricsReporter extends MetricsReporter with Reconfigurable with Close
  }
 }

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


Is this being used?

chia7712 · 2025-08-26T18:58:04Z

core/src/test/scala/integration/kafka/server/DynamicBrokerReconfigurationTest.scala

+}
+
+
+class TestNumReplicaFetcherMetricsReporter extends MetricsReporter with Reconfigurable with Closeable with ClusterResourceListener {


class TestNumReplicaFetcherMetricsReporter extends MetricsReporter

chia7712 · 2025-08-26T18:58:33Z

core/src/test/scala/integration/kafka/server/DynamicBrokerReconfigurationTest.scala

+  @volatile var numFetchers: Int = 1
+  testReporters.add(this)
+
+  override def contextChange(metricsContext: MetricsContext): Unit = {


we don't need to override the method with empty body

chia7712 · 2025-08-26T18:58:48Z

core/src/test/scala/integration/kafka/server/DynamicBrokerReconfigurationTest.scala

+  override def metricRemoval(metric: KafkaMetric): Unit = {
+  }
+
+  override def onUpdate(clusterResource: ClusterResource): Unit = {


Signed-off-by: PoAn Yang <payang@apache.org>

chia7712

LGTM

#20405) - **Changes**: Replace misused dynamicPerBrokerConfigs with dynamicDefaultConfigs - **Reasons**: KRaft servers don't handle the cluser-level configs in starting from: https://github.com/apache/kafka/pull/18949/files#r2296809389 Reviewers: Jun Rao <junrao@gmail.com>, Jhen-Yung Hsu <jhenyunghsu@gmail.com>, PoAn Yang <payang@apache.org>, Chia-Ping Tsai <chia7712@gmail.com> --------- Co-authored-by: PoAn Yang <payang@apache.org>

Replace dynamicPerBrokerConfigs with dynamicDefaultConfigs

daeb1dd

junrao reviewed Aug 25, 2025

View reviewed changes

Use per-broker configs for broker-scoped updates correctly

3aab8c6

chia7712 added the ci-approved label Aug 25, 2025

feat: add test for readDynamicBrokerConfigsFromSnapshot

b15e1c0

Signed-off-by: PoAn Yang <payang@apache.org>

FrankYang0529 requested a review from chia7712 August 26, 2025 06:29

chia7712 reviewed Aug 26, 2025

View reviewed changes

Yunyung reviewed Aug 26, 2025

View reviewed changes

address comment

8696bdf

Signed-off-by: PoAn Yang <payang@apache.org>

chia7712 reviewed Aug 26, 2025

View reviewed changes

core/src/test/scala/integration/kafka/server/DynamicBrokerReconfigurationTest.scala Outdated Show resolved Hide resolved

address comment

6cf7f20

Signed-off-by: PoAn Yang <payang@apache.org>

junrao reviewed Aug 26, 2025

View reviewed changes

chia7712 reviewed Aug 26, 2025

View reviewed changes

address comment

8513d4e

Signed-off-by: PoAn Yang <payang@apache.org>

chia7712 approved these changes Aug 27, 2025

View reviewed changes

chia7712 merged commit 8de88db into apache:4.1 Aug 27, 2025
16 of 20 checks passed

	def verifyState(reconfigureCount: Int, deleteCount: Int, pollingInterval: Int, numFetcher: Int = 1): Unit = {
	def verifyState(reconfigureCount: Int, deleteCount: Int, pollingInterval: Int, numFetcher: Int): Unit = {

		}


		class TestNumReplicaFetcherMetricsReporter extends MetricsReporter with Reconfigurable with Closeable with ClusterResourceListener {

KAFKA-19642: Replace dynamicPerBrokerConfigs with dynamicDefaultConfigs #20405

KAFKA-19642: Replace dynamicPerBrokerConfigs with dynamicDefaultConfigs #20405

Conversation

jim0987795064 commented Aug 25, 2025 • edited by chia7712 Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

junrao left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

jim0987795064 Aug 25, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

jim0987795064 commented Aug 25, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

chia7712 commented Aug 25, 2025

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Yunyung left a comment

Choose a reason for hiding this comment

Uh oh!

Yunyung Aug 26, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Yunyung Aug 26, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

junrao left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

chia7712 left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

jim0987795064 commented Aug 25, 2025 •

edited by chia7712

Loading

jim0987795064 Aug 25, 2025 •

edited

Loading

jim0987795064 commented Aug 25, 2025 •

edited

Loading

Yunyung Aug 26, 2025 •

edited

Loading

Yunyung Aug 26, 2025 •

edited

Loading