fix(backpressure): start before registering to dynamic configs (#2208)

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

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

@@ -511,6 +511,13 @@ class BrokerServer(
         () => lifecycleManager.resendBrokerRegistrationUnlessZkMode())
       metadataPublishers.add(brokerRegistrationTracker)
 
+      // AutoMQ inject start
+      backPressureManager = new DefaultBackPressureManager(
+        BackPressureConfig.from(config),
+        newBackPressureRegulator()
+      )
+      backPressureManager.start()
+      // AutoMQ inject end
 
       // Register parts of the broker that can be reconfigured via dynamic configs.  This needs to
       // be done before we publish the dynamic configs, so that we don't miss anything.
@@ -551,12 +558,6 @@ class BrokerServer(
       newPartitionLifecycleListeners().forEach(l => {
         _replicaManager.addPartitionLifecycleListener(l)
       })
-
-      backPressureManager = new DefaultBackPressureManager(
-        BackPressureConfig.from(config),
-        newBackPressureRegulator()
-      )
-      backPressureManager.start()
       // AutoMQ inject end
 
       // We're now ready to unfence the broker. This also allows this broker to transition