MINOR: Use typed BrokerReconfigurable in DynamicLogConfig

Author: unknowntpo

server/src/main/java/org/apache/kafka/server/config/DynamicLogConfig.java

@@ -16,7 +16,6 @@
  */
 package org.apache.kafka.server.config;
 
-import org.apache.kafka.common.config.AbstractConfig;
 import org.apache.kafka.common.config.ConfigException;
 import org.apache.kafka.config.BrokerReconfigurable;
 import org.apache.kafka.server.common.DirectoryEventHandler;
@@ -33,7 +32,7 @@
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
 
-public class DynamicLogConfig implements BrokerReconfigurable {
+public class DynamicLogConfig implements BrokerReconfigurable<AbstractKafkaConfig> {
     /**
      * The broker configurations pertaining to logs that are reconfigurable. This set contains
      * the names you would use when setting a static or dynamic broker configuration (not topic
@@ -59,21 +58,12 @@ public Set<String> reconfigurableConfigs() {
     }
 
     @Override
-    public void validateReconfiguration(AbstractConfig newConfig) {
-        AbstractKafkaConfig kafkaConfig = requireKafkaConfig(newConfig);
-        validateLogLocalRetentionMs(kafkaConfig);
-        validateLogLocalRetentionBytes(kafkaConfig);
-        validateLogRemoteCopyLagMs(kafkaConfig);
-        validateLogRemoteCopyLagBytes(kafkaConfig);
-        validateCordonedLogDirs(kafkaConfig);
-    }
-
-    private AbstractKafkaConfig requireKafkaConfig(AbstractConfig config) {
-        if (!(config instanceof AbstractKafkaConfig kafkaConfig)) {
-            throw new IllegalArgumentException("DynamicLogConfig requires AbstractKafkaConfig, got " +
-                config.getClass().getName());
-        }
-        return kafkaConfig;
+    public void validateReconfiguration(AbstractKafkaConfig newConfig) {
+        validateLogLocalRetentionMs(newConfig);
+        validateLogLocalRetentionBytes(newConfig);
+        validateLogRemoteCopyLagMs(newConfig);
+        validateLogRemoteCopyLagBytes(newConfig);
+        validateCordonedLogDirs(newConfig);
     }
 
     private void validateLogLocalRetentionMs(AbstractKafkaConfig config) {
@@ -157,14 +147,13 @@ private void updateLogsConfig(Map<String, Object> newBrokerDefaults) {
     }
 
     @Override
-    public void reconfigure(AbstractConfig oldConfig, AbstractConfig newConfig) {
-        AbstractKafkaConfig kafkaConfig = requireKafkaConfig(newConfig);
-        Map<String, Object> newBrokerDefaults = new HashMap<>(kafkaConfig.extractLogConfigMap());
+    public void reconfigure(AbstractKafkaConfig oldConfig, AbstractKafkaConfig newConfig) {
+        Map<String, Object> newBrokerDefaults = new HashMap<>(newConfig.extractLogConfigMap());
         logManager.reconfigureDefaultLogConfig(new LogConfig(newBrokerDefaults));
         updateLogsConfig(newBrokerDefaults);
 
-        logManager.updateCordonedLogDirs(Set.copyOf(kafkaConfig.cordonedLogDirs()));
-        directoryEventHandler.handleCordoned(kafkaConfig.cordonedLogDirs().stream()
+        logManager.updateCordonedLogDirs(Set.copyOf(newConfig.cordonedLogDirs()));
+        directoryEventHandler.handleCordoned(newConfig.cordonedLogDirs().stream()
             .flatMap(dir -> logManager.directoryId(dir).stream())
             .collect(Collectors.toSet()));
     }