refactor(controller): add method ControllerServer#reconfigurables (#2344)

Signed-off-by: Ning Yu <[email protected]>

Author: Chillax-0v0

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

@@ -22,7 +22,6 @@ import com.automq.stream.s3.metadata.ObjectUtils
 import kafka.autobalancer.AutoBalancerManager
 import kafka.autobalancer.services.AutoBalancerService
 import kafka.controller.streamaspect.client.{Context, StreamClientFactoryProxy}
-
 import kafka.migration.MigrationPropagator
 import kafka.network.{DataPlaneAcceptor, SocketServer}
 import kafka.raft.KafkaRaftManager
@@ -38,7 +37,7 @@ import org.apache.kafka.common.network.ListenerName
 import org.apache.kafka.common.security.scram.internals.ScramMechanism
 import org.apache.kafka.common.security.token.delegation.internals.DelegationTokenCache
 import org.apache.kafka.common.utils.LogContext
-import org.apache.kafka.common.{ClusterResource, Endpoint, Uuid}
+import org.apache.kafka.common.{ClusterResource, Endpoint, Reconfigurable, Uuid}
 import org.apache.kafka.controller.metrics.{ControllerMetadataMetricsPublisher, QuorumControllerMetrics}
 import org.apache.kafka.controller.{QuorumController, QuorumControllerExtension, QuorumFeatures}
 import org.apache.kafka.image.publisher.{ControllerRegistrationsPublisher, MetadataPublisher}
@@ -584,5 +583,12 @@ class ControllerServer(
   protected def replicaPlacer(): ReplicaPlacer = {
     new StripedReplicaPlacer(new Random())
   }
+
+  // return a list of all reconfigurable objects
+  def reconfigurables(): java.util.List[Reconfigurable] = {
+    java.util.List.of(
+      autoBalancerManager
+    )
+  }
   // AutoMQ for Kafka inject end
 }

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

@@ -302,7 +302,9 @@ class DynamicBrokerConfig(private val kafkaConfig: KafkaConfig) extends Logging
       addReconfigurable(new DynamicMetricsReporters(kafkaConfig.nodeId, controller.config, controller.metrics, controller.clusterId))
     }
     addReconfigurable(new DynamicClientQuotaCallback(controller.quotaManagers, controller.config))
-    addReconfigurable(controller.autoBalancerManager)
+    // AutoMQ inject start
+    controller.reconfigurables().asScala.foreach(addReconfigurable)
+    // AutoMQ inject end
     addBrokerReconfigurable(new ControllerDynamicThreadPool(controller))
     // TODO: addBrokerReconfigurable(new DynamicListenerConfig(controller))
     addBrokerReconfigurable(controller.socketServer)