MINOR: Add generic config type to BrokerReconfigurable

Author: unknowntpo

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

@@ -229,7 +229,7 @@ class DynamicBrokerConfig(private val kafkaConfig: KafkaConfig) extends Logging
     reconfigurables.add(reconfigurable)
   }
 
-  def addBrokerReconfigurable(reconfigurable: config.BrokerReconfigurable): Unit = {
+  def addBrokerReconfigurable(reconfigurable: config.BrokerReconfigurable[_ >: KafkaConfig]): Unit = {
     verifyReconfigurableConfigs(reconfigurable.reconfigurableConfigs)
     brokerReconfigurables.add(new BrokerReconfigurable {
       override def reconfigurableConfigs: util.Set[String] = reconfigurable.reconfigurableConfigs

server-common/src/main/java/org/apache/kafka/config/BrokerReconfigurable.java

@@ -32,8 +32,10 @@
  *   <li>Validating the new configuration before applying it via {@link #validateReconfiguration(AbstractConfig)}</li>
  *   <li>Applying the new configuration via {@link #reconfigure(AbstractConfig, AbstractConfig)}</li>
  * </ol>
+ *
+ * @param <T> the configuration type used by the reconfigurable component
  */
-public interface BrokerReconfigurable {
+public interface BrokerReconfigurable<T extends AbstractConfig> {
     /**
      * Returns the set of configuration keys that can be dynamically reconfigured.
      *
@@ -53,7 +55,7 @@ public interface BrokerReconfigurable {
      *
      * @param newConfig the new configuration to validate
      */
-    void validateReconfiguration(AbstractConfig newConfig);
+    void validateReconfiguration(T newConfig);
 
     /**
      * Applies the new configuration.
@@ -63,5 +65,5 @@ public interface BrokerReconfigurable {
      * @param oldConfig the previous configuration
      * @param newConfig the new configuration to apply
      */
-    void reconfigure(AbstractConfig oldConfig, AbstractConfig newConfig);
+    void reconfigure(T oldConfig, T newConfig);
 }

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

@@ -27,7 +27,7 @@
 
 import java.util.Set;
 
-public class DynamicProducerStateManagerConfig implements BrokerReconfigurable {
+public class DynamicProducerStateManagerConfig implements BrokerReconfigurable<AbstractConfig> {
     private final Logger log = LoggerFactory.getLogger(DynamicProducerStateManagerConfig.class);
     private final ProducerStateManagerConfig producerStateManagerConfig;
 

storage/src/main/java/org/apache/kafka/storage/internals/log/LogCleaner.java

@@ -95,7 +95,7 @@
  *    tombstone deletion.</li>
  * </ol>
  */
-public class LogCleaner implements BrokerReconfigurable {
+public class LogCleaner implements BrokerReconfigurable<AbstractConfig> {
     private static final Logger LOG = LoggerFactory.getLogger(LogCleaner.class);
 
     public static final Set<String> RECONFIGURABLE_CONFIGS = Set.of(