tyamashi-oss commented on code in PR #12296: URL: https://github.com/apache/kafka/pull/12296#discussion_r914719266
########## core/src/main/scala/kafka/log/LogCleaner.scala: ########## @@ -186,11 +186,18 @@ class LogCleaner(initialConfig: CleanerConfig, } /** - * Reconfigure log clean config. This simply stops current log cleaners and creates new ones. + * Reconfigure log clean config. This updates desiredRatePerSec in Throttler with logCleanerIoMaxBytesPerSecond and stops current log cleaners and creates new ones. Review Comment: Thank you. I’ve update the comment. https://github.com/apache/kafka/pull/12296/commits/f9d27f6181c3fbc1ef02d56904d58793962dedb2 ########## core/src/test/scala/unit/kafka/log/LogCleanerTest.scala: ########## @@ -1854,6 +1853,26 @@ class LogCleanerTest { } finally logCleaner.shutdown() } + @Test + def testReconfigureLogCleanerIoMaxBytesPerSecond(): Unit = { + val oldKafkaProps = TestUtils.createBrokerConfig(1, "localhost:2181") + oldKafkaProps.put(KafkaConfig.LogCleanerIoMaxBytesPerSecondProp, 10000000) + + val logCleaner = new LogCleaner(LogCleaner.cleanerConfig(new KafkaConfig(oldKafkaProps)), + logDirs = Array(TestUtils.tempDir()), + logs = new Pool[TopicPartition, UnifiedLog](), + logDirFailureChannel = new LogDirFailureChannel(1), + time = time) + + assertEquals(logCleaner.throttler.desiredRatePerSec, 10000000, "Throttler.desiredRatePerSec should be initialized with KafkaConfig.LogCleanerIoMaxBytesPerSecondProp") + + val newKafkaProps = TestUtils.createBrokerConfig(1, "localhost:2181") + newKafkaProps.put(KafkaConfig.LogCleanerIoMaxBytesPerSecondProp, 20000000) + + logCleaner.reconfigure(new KafkaConfig(oldKafkaProps), new KafkaConfig(newKafkaProps)) Review Comment: Your concern is correct. Thank you. LogCleaner.shutdown() should be called at the end of the test because kafka-log-cleaner-thread-x threads are created in LogCleaner.startup() at the end of LogCleaner.reconfigure(), and the threads continue to remain. I appended LogCleaner.shutdown() to the end of the test and also used LogCleaner with empty startup() and shutdown() implementations. The test is somewhat more white-box like according to the LogCleaner.reconfigure() implementation, but I couldn't think of any other way. Please let me know if you have any. https://github.com/apache/kafka/pull/12296/commits/e05707d5613d096df06c4e96085deb481b00a7e7 -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: jira-unsubscr...@kafka.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
