KAFKA-13417; Ensure dynamic reconfigurations set old config properly

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

@@ -516,7 +516,11 @@ class DynamicBrokerConfig(private val kafkaConfig: KafkaConfig) extends Logging
     newProps ++= staticBrokerConfigs
     overrideProps(newProps, dynamicDefaultConfigs)
     overrideProps(newProps, dynamicBrokerConfigs)
-    val oldConfig = currentConfig
+
+    // We need a copy of the current config since `currentConfig` is initialized with `kafkaConfig`
+    // which means the call to `updateCurrentConfig` would end up mutating `oldConfig`.
+    val oldConfig = KafkaConfig(currentConfig.values, doLog = false)
+
     val (newConfig, brokerReconfigurablesToUpdate) = processReconfiguration(newProps, validateOnly = false)
     if (newConfig ne currentConfig) {
       currentConfig = newConfig
@@ -955,6 +959,3 @@ class DynamicListenerConfig(server: KafkaBroker) extends BrokerReconfigurable wi
     listeners.map(e => (e.listenerName, e)).toMap
 
 }
-
-
-

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

@@ -1341,7 +1341,7 @@ object KafkaConfig {
     fromProps(props, doLog)
   }
 
-  def apply(props: java.util.Map[_, _]): KafkaConfig = new KafkaConfig(props, true)
+  def apply(props: java.util.Map[_, _], doLog: Boolean = true): KafkaConfig = new KafkaConfig(props, true)
 
   private def typeOf(name: String): Option[ConfigDef.Type] = Option(configDef.configKeys.get(name)).map(_.`type`)
 

core/src/test/scala/unit/kafka/server/DynamicBrokerConfigTest.scala

@@ -20,6 +20,7 @@ package kafka.server
 import java.{lang, util}
 import java.util.Properties
 import java.util.concurrent.CompletionStage
+
 import kafka.utils.TestUtils
 import kafka.zk.KafkaZkClient
 import org.apache.kafka.common.{Endpoint, Reconfigurable}
@@ -30,6 +31,7 @@ import org.apache.kafka.server.authorizer._
 import org.easymock.EasyMock
 import org.junit.jupiter.api.Assertions._
 import org.junit.jupiter.api.Test
+import org.mockito.Mockito
 
 import scala.annotation.nowarn
 import scala.jdk.CollectionConverters._
@@ -80,6 +82,27 @@ class DynamicBrokerConfigTest {
     }
   }
 
+  @Test
+  def testUpdateThreadPool(): Unit = {
+    val origProps = TestUtils.createBrokerConfig(0, TestUtils.MockZkConnect, port = 8181)
+    origProps.put(KafkaConfig.NumIoThreadsProp, "4")
+
+    val config = KafkaConfig(origProps)
+    val serverMock = Mockito.mock(classOf[KafkaBroker])
+    val handlerPoolMock = Mockito.mock(classOf[KafkaRequestHandlerPool])
+    Mockito.when(serverMock.config).thenReturn(config)
+    Mockito.when(serverMock.dataPlaneRequestHandlerPool).thenReturn(handlerPoolMock)
+
+    config.dynamicConfig.addBrokerReconfigurable(new DynamicThreadPool(serverMock))
+
+    val updateProps = new Properties()
+    updateProps.put(KafkaConfig.NumIoThreadsProp, "8")
+
+    config.dynamicConfig.updateDefaultConfig(updateProps)
+    assertEquals(8, config.numIoThreads)
+    Mockito.verify(handlerPoolMock).resizeThreadPool(8)
+  }
+
   @nowarn("cat=deprecation")
   @Test
   def testConfigUpdateWithSomeInvalidConfigs(): Unit = {