diff --git a/core/src/main/scala/kafka/server/DynamicBrokerConfig.scala b/core/src/main/scala/kafka/server/DynamicBrokerConfig.scala index bf3d988244689..75081847e3e98 100755 --- a/core/src/main/scala/kafka/server/DynamicBrokerConfig.scala +++ b/core/src/main/scala/kafka/server/DynamicBrokerConfig.scala @@ -619,7 +619,7 @@ class DynamicLogConfig(logManager: LogManager, server: KafkaServer) extends Brok props ++= newBrokerDefaults props ++= log.config.originals.asScala.filterKeys(log.config.overriddenConfigs.contains) - val logConfig = LogConfig(props.asJava) + val logConfig = LogConfig(props.asJava, log.config.overriddenConfigs) log.updateConfig(logConfig) } } diff --git a/core/src/test/scala/integration/kafka/server/DynamicBrokerReconfigurationTest.scala b/core/src/test/scala/integration/kafka/server/DynamicBrokerReconfigurationTest.scala index f77eac9f5a6c4..e959d302a6581 100644 --- a/core/src/test/scala/integration/kafka/server/DynamicBrokerReconfigurationTest.scala +++ b/core/src/test/scala/integration/kafka/server/DynamicBrokerReconfigurationTest.scala @@ -504,6 +504,40 @@ class DynamicBrokerReconfigurationTest extends ZooKeeperTestHarness with SaslSet stopAndVerifyProduceConsume(producerThread, consumerThread) } + @Test + def testConsecutiveConfigChange(): Unit = { + val topic2 = "testtopic2" + val topicProps = new Properties + topicProps.put(KafkaConfig.MinInSyncReplicasProp, "2") + TestUtils.createTopic(zkClient, topic2, 1, replicationFactor = numServers, servers, topicProps) + var log = servers.head.logManager.getLog(new TopicPartition(topic2, 0)).getOrElse(throw new IllegalStateException("Log not found")) + assertTrue(log.config.overriddenConfigs.contains(KafkaConfig.MinInSyncReplicasProp)) + assertEquals("2", log.config.originals().get(KafkaConfig.MinInSyncReplicasProp).toString) + + val props = new Properties + props.put(KafkaConfig.MinInSyncReplicasProp, "3") + // Make a broker-default config + reconfigureServers(props, perBrokerConfig = false, (KafkaConfig.MinInSyncReplicasProp, "3")) + // Verify that all broker defaults have been updated again + servers.foreach { server => + props.asScala.foreach { case (k, v) => + assertEquals(s"Not reconfigured $k", v, server.config.originals.get(k).toString) + } + } + + log = servers.head.logManager.getLog(new TopicPartition(topic2, 0)).getOrElse(throw new IllegalStateException("Log not found")) + assertTrue(log.config.overriddenConfigs.contains(KafkaConfig.MinInSyncReplicasProp)) + assertEquals("2", log.config.originals().get(KafkaConfig.MinInSyncReplicasProp).toString) // Verify topic-level config survives + + // Make a second broker-default change + props.clear() + props.put(KafkaConfig.LogRetentionTimeMillisProp, "604800000") + reconfigureServers(props, perBrokerConfig = false, (KafkaConfig.LogRetentionTimeMillisProp, "604800000")) + log = servers.head.logManager.getLog(new TopicPartition(topic2, 0)).getOrElse(throw new IllegalStateException("Log not found")) + assertTrue(log.config.overriddenConfigs.contains(KafkaConfig.MinInSyncReplicasProp)) + assertEquals("2", log.config.originals().get(KafkaConfig.MinInSyncReplicasProp).toString) // Verify topic-level config still survives + } + @Test def testDefaultTopicConfig(): Unit = { val (producerThread, consumerThread) = startProduceConsume(retries = 0) @@ -774,7 +808,7 @@ class DynamicBrokerReconfigurationTest extends ZooKeeperTestHarness with SaslSet val partitions = (0 until numPartitions).map(i => new TopicPartition(topic, i)).filter { tp => zkClient.getLeaderForPartition(tp).contains(leaderId) } - assertTrue(s"Partitons not found with leader $leaderId", partitions.nonEmpty) + assertTrue(s"Partitions not found with leader $leaderId", partitions.nonEmpty) partitions.foreach { tp => (1 to 2).foreach { i => val replicaFetcherManager = servers(i).replicaManager.replicaFetcherManager