(kafka) branch KAFKA-16649 created (now f435287ac63)

[cm...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

cmccabe pushed a change to branch KAFKA-16649
in repository https://gitbox.apache.org/repos/asf/kafka.git


      at f435287ac63 Fixes

This branch includes the following new commits:

     new f435287ac63 Fixes

The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails.  The revisions
listed as "add" were already present in the repository and have only
been added to this reference.

(kafka) 01/01: Fixes

[cm...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

cmccabe pushed a commit to branch KAFKA-16649
in repository https://gitbox.apache.org/repos/asf/kafka.git

commit f435287ac6315f9bacd41c7862559b42cbd8f395
Author: Colin P. McCabe <cm...@apache.org>
AuthorDate: Wed May 1 14:25:18 2024 -0700

    Fixes
---
 core/src/main/scala/kafka/server/DynamicBrokerConfig.scala        | 8 +++++++-
 .../test/scala/integration/kafka/server/KRaftClusterTest.scala    | 4 ++--
 2 files changed, 9 insertions(+), 3 deletions(-)

diff --git a/core/src/main/scala/kafka/server/DynamicBrokerConfig.scala b/core/src/main/scala/kafka/server/DynamicBrokerConfig.scala
index 924bb0f660e..d2d518f1dd2 100755
--- a/core/src/main/scala/kafka/server/DynamicBrokerConfig.scala
+++ b/core/src/main/scala/kafka/server/DynamicBrokerConfig.scala
@@ -221,7 +221,13 @@ class DynamicBrokerConfig(private val kafkaConfig: KafkaConfig) extends Logging
   // collections, while another thread is iterating over them.
   private[server] val reconfigurables = new CopyOnWriteArrayList[Reconfigurable]()
   private val brokerReconfigurables = new CopyOnWriteArrayList[BrokerReconfigurable]()
+
+  /**
+   * The DynamicBrokerConfig lock which prevents concurrent changes to dynamic configuration.
+   * (It does not prevent new reconfigurables from being registered or unregistered, though.)
+   */
   private val lock = new ReentrantReadWriteLock
+
   private var metricsReceiverPluginOpt: Option[ClientMetricsReceiverPlugin] = _
   private var currentConfig: KafkaConfig = _
   private val dynamicConfigPasswordEncoder = if (kafkaConfig.processRoles.isEmpty) {
@@ -317,7 +323,7 @@ class DynamicBrokerConfig(private val kafkaConfig: KafkaConfig) extends Logging
     reconfigurables.remove(reconfigurable)
   }
 
-  private def verifyReconfigurableConfigs(configNames: Set[String]): Unit = CoreUtils.inWriteLock(lock) {
+  private def verifyReconfigurableConfigs(configNames: Set[String]): Unit = {
     val nonDynamic = configNames.filter(DynamicConfig.Broker.nonDynamicProps.contains)
     require(nonDynamic.isEmpty, s"Reconfigurable contains non-dynamic configs $nonDynamic")
   }
diff --git a/core/src/test/scala/integration/kafka/server/KRaftClusterTest.scala b/core/src/test/scala/integration/kafka/server/KRaftClusterTest.scala
index 83182d54443..d59d8859562 100644
--- a/core/src/test/scala/integration/kafka/server/KRaftClusterTest.scala
+++ b/core/src/test/scala/integration/kafka/server/KRaftClusterTest.scala
@@ -1548,7 +1548,7 @@ class KRaftClusterTest {
       new TestKitNodes.Builder().
         setNumBrokerNodes(1).
         setNumControllerNodes(1).build()).
-      setConfigProp(KafkaConfig.NumNetworkThreadsProp, "4").
+      setConfigProp(KafkaConfig.NumNetworkThreadsProp, "2").
       build()
     try {
       cluster.format()
@@ -1559,7 +1559,7 @@ class KRaftClusterTest {
         admin.incrementalAlterConfigs(
           Collections.singletonMap(new ConfigResource(Type.BROKER, ""),
             Collections.singletonList(new AlterConfigOp(
-              new ConfigEntry(KafkaConfig.NumNetworkThreadsProp, "2"), OpType.SET)))).all().get()
+              new ConfigEntry(KafkaConfig.NumNetworkThreadsProp, "3"), OpType.SET)))).all().get()
         val newTopic = Collections.singletonList(new NewTopic("test-topic", 1, 1.toShort))
         val createTopicResult = admin.createTopics(newTopic)
         createTopicResult.all().get()