From 43b03f35049efa5510a40a6a46bb2bf78bf4750a Mon Sep 17 00:00:00 2001
From: Jon Bringhurst <jbringhu@jbringhu-mn3034.linkedin.biz>
Date: Fri, 11 Sep 2026 11:08:27 -0700
Subject: [PATCH 1/2] kafka: make bridge configuration metrics an explicit
 opt-in

---
 .github/workflows/li-ci.yml                   |   2 +-
 .../main/scala/kafka/server/KafkaConfig.scala |   6 +
 .../server/LiProtocolBridgeMetrics.scala      | 107 +++++++++---------
 .../server/LiProtocolBridgeMetricsTest.scala  |  55 ++++++++-
 tests/bin/README.li-bridge.md                 |   4 +
 tests/bin/li_bridge_contract.py               |   4 +-
 tests/bin/li_bridge_mixed_cluster_smoke.py    |   5 +-
 tests/unit/li_bridge_preflight_test.py        |  18 +++
 8 files changed, 143 insertions(+), 58 deletions(-)

diff --git a/.github/workflows/li-ci.yml b/.github/workflows/li-ci.yml
index 8e79a825e9d86..11d56ae73b249 100644
--- a/.github/workflows/li-ci.yml
+++ b/.github/workflows/li-ci.yml
@@ -229,7 +229,7 @@ jobs:
     env:
       # Checkout fixes the companion source for this run. Record its commit and
       # archive hash; release CI separately requires approved commit hashes.
-      LI_BRIDGE_LEGACY_REF: ${{ vars.LI_BRIDGE_LEGACY_REF || '3.0-li-bridge/topic-identity-recovery' }}
+      LI_BRIDGE_LEGACY_REF: ${{ vars.LI_BRIDGE_LEGACY_REF || '3.0-li-bridge/config-metrics-gate' }}
     steps:
       - uses: actions/checkout@v4
         with:
diff --git a/core/src/main/scala/kafka/server/KafkaConfig.scala b/core/src/main/scala/kafka/server/KafkaConfig.scala
index 688eda5bce74e..5c1290502a1c4 100755
--- a/core/src/main/scala/kafka/server/KafkaConfig.scala
+++ b/core/src/main/scala/kafka/server/KafkaConfig.scala
@@ -61,6 +61,7 @@ import scala.collection.{Map, Seq}
 object KafkaConfig {
 
   val LiProtocolBridgeModeEnableProp = "li.protocol.bridge.mode.enable"
+  val LiProtocolBridgeConfigMetricsEnableProp = "li.protocol.bridge.config.metrics.enable"
   val LiProtocolBridgeTopicDeletionStateCleanupEnableProp =
     "li.protocol.bridge.topic.deletion.state.cleanup.enable"
   val LiProtocolBridgeFollowerRecoveryEnableProp = "li.protocol.bridge.follower.recovery.enable"
@@ -104,6 +105,7 @@ object KafkaConfig {
 
   val LiProtocolBridgeEnableProps: Seq[String] = Seq(
     LiProtocolBridgeModeEnableProp,
+    LiProtocolBridgeConfigMetricsEnableProp,
     LiProtocolBridgeTopicDeletionStateCleanupEnableProp,
     LiProtocolBridgeFollowerRecoveryEnableProp,
     LiProtocolBridgeRecommendedElectionEnableProp,
@@ -280,6 +282,8 @@ object KafkaConfig {
   val configDef = new ConfigDef(AbstractKafkaConfig.CONFIG_DEF)
     .define(LiProtocolBridgeModeEnableProp, ConfigDef.Type.BOOLEAN, false,
       ConfigDef.Importance.HIGH, LiProtocolBridgeModeEnableDoc)
+    .define(LiProtocolBridgeConfigMetricsEnableProp, ConfigDef.Type.BOOLEAN, false,
+      ConfigDef.Importance.LOW, "Register bridge configuration gauges on ZooKeeper brokers. Requires a broker restart.")
     .define(LiProtocolBridgeTopicDeletionStateCleanupEnableProp, ConfigDef.Type.BOOLEAN, false,
       ConfigDef.Importance.HIGH, "Clear stale topic deletion state and reconcile the metadata cache " +
         "from the first full update of each ZooKeeper controller epoch. Enable on every broker together.")
@@ -483,6 +487,8 @@ class KafkaConfig private(doLog: Boolean, val props: util.Map[_, _])
   @volatile private var currentConfig = this
   val processRoles: Set[ProcessRole] = parseProcessRoles()
   def liProtocolBridgeModeEnable: Boolean = getBoolean(KafkaConfig.LiProtocolBridgeModeEnableProp)
+  def liProtocolBridgeConfigMetricsActive: Boolean =
+    processRoles.isEmpty && getBoolean(KafkaConfig.LiProtocolBridgeConfigMetricsEnableProp)
   def liProtocolBridgeTopicDeletionStateCleanupActive: Boolean =
     processRoles.isEmpty && getBoolean(KafkaConfig.LiProtocolBridgeTopicDeletionStateCleanupEnableProp)
   def liProtocolBridgeFollowerRecoveryEnable: Boolean =
diff --git a/core/src/main/scala/kafka/server/LiProtocolBridgeMetrics.scala b/core/src/main/scala/kafka/server/LiProtocolBridgeMetrics.scala
index 27a91e6b0a026..42f9370394059 100644
--- a/core/src/main/scala/kafka/server/LiProtocolBridgeMetrics.scala
+++ b/core/src/main/scala/kafka/server/LiProtocolBridgeMetrics.scala
@@ -22,6 +22,7 @@ import scala.jdk.CollectionConverters._
 
 object LiProtocolBridgeMetrics {
   val ModeEnabled = "ModeEnabled"
+  val ConfigMetricsEnabled = "ConfigMetricsEnabled"
   val TopicDeletionStateCleanupEnabled = "TopicDeletionStateCleanupEnabled"
   val FollowerRecoveryEnabled = "FollowerRecoveryEnabled"
   val RecommendedLeaderElectionEnabled = "RecommendedLeaderElectionEnabled"
@@ -46,7 +47,7 @@ object LiProtocolBridgeMetrics {
   val LeaderTransferEnabled = "LeaderTransferEnabled"
   val LegacyRequestMetricsEnabled = "LegacyRequestMetricsEnabled"
   val LogTruncationMetricsEnabled = "LogTruncationMetricsEnabled"
-  val MetricNames: Seq[String] = Seq(ModeEnabled, TopicDeletionStateCleanupEnabled, FollowerRecoveryEnabled,
+  val MetricNames: Seq[String] = Seq(ModeEnabled, ConfigMetricsEnabled, TopicDeletionStateCleanupEnabled, FollowerRecoveryEnabled,
     RecommendedLeaderElectionEnabled, ExcludePartitionsEnabled, MoveControllerEnabled,
     ShutdownSafetyOverrideEnabled, PreferredControllerEnabled, FederatedTopicsEnabled,
     RackIdMapperEnabled, ZookeeperPaginationEnabled, DynamicTopicDeletionEnabled,
@@ -64,60 +65,64 @@ class LiProtocolBridgeMetrics(config: KafkaConfig) extends AutoCloseable {
   private val metricsGroup = new KafkaMetricsGroup(this.getClass)
   private val tags = Map("broker-id" -> config.brokerId.toString).asJava
 
-  metricsGroup.newGauge(ModeEnabled, () => enabled(config.liProtocolBridgeModeActive), tags)
-  metricsGroup.newGauge(TopicDeletionStateCleanupEnabled,
-    () => enabled(config.liProtocolBridgeTopicDeletionStateCleanupActive), tags)
-  metricsGroup.newGauge(FollowerRecoveryEnabled,
-    () => enabled(config.liProtocolBridgeFollowerRecoveryActive), tags)
-  metricsGroup.newGauge(RecommendedLeaderElectionEnabled,
-    () => enabled(config.liProtocolBridgeRecommendedElectionActive), tags)
-  metricsGroup.newGauge(ExcludePartitionsEnabled,
-    () => enabled(config.liProtocolBridgeExcludePartitionsActive), tags)
-  metricsGroup.newGauge(MoveControllerEnabled,
-    () => enabled(config.liProtocolBridgeMoveControllerActive), tags)
-  metricsGroup.newGauge(ShutdownSafetyOverrideEnabled,
-    () => enabled(config.liProtocolBridgeShutdownSafetyOverrideActive), tags)
-  metricsGroup.newGauge(PreferredControllerEnabled,
-    () => enabled(config.liProtocolBridgePreferredControllerActive), tags)
-  metricsGroup.newGauge(FederatedTopicsEnabled,
-    () => enabled(config.liProtocolBridgeFederatedTopicsActive), tags)
-  metricsGroup.newGauge(RackIdMapperEnabled,
-    () => enabled(config.liProtocolBridgeRackIdMapperActive), tags)
-  metricsGroup.newGauge(ZookeeperPaginationEnabled,
-    () => enabled(config.liZookeeperPaginationEnable), tags)
-  metricsGroup.newGauge(DynamicTopicDeletionEnabled,
-    () => enabled(config.liProtocolBridgeDynamicTopicDeletionActive), tags)
-  metricsGroup.newGauge(ControllerInitializationThreads,
-    () => config.liNumControllerInitThreads, tags)
-  metricsGroup.newGauge(ProduceRequestInstrumentationEnabled,
-    () => enabled(config.liProtocolBridgeProduceRequestInstrumentationActive), tags)
-  metricsGroup.newGauge(RequestMetricBucketsEnabled,
-    () => enabled(config.liProtocolBridgeRequestMetricBucketsActive), tags)
-  metricsGroup.newGauge(RequestChannelWatchdogEnabled,
-    () => enabled(config.liProtocolBridgeRequestChannelWatchdogActive), tags)
-  metricsGroup.newGauge(MinimumLogRollEnabled,
-    () => enabled(config.liProtocolBridgeMinimumLogRollActive), tags)
-  metricsGroup.newGauge(ReassignmentCancellationSafetyEnabled,
-    () => enabled(config.liProtocolBridgeReassignmentCancellationSafetyActive), tags)
-  metricsGroup.newGauge(ListOffsetsInstrumentationEnabled,
-    () => enabled(config.liProtocolBridgeListOffsetsInstrumentationActive), tags)
-  metricsGroup.newGauge(StaticDefaultQuotasEnabled,
-    () => enabled(config.liProtocolBridgeStaticDefaultQuotasActive), tags)
-  metricsGroup.newGauge(ReplicaRequestTimeoutEnabled,
-    () => enabled(config.liProtocolBridgeReplicaRequestTimeoutActive), tags)
-  metricsGroup.newGauge(OffsetsTopicConfigEnabled,
-    () => enabled(config.liProtocolBridgeOffsetsTopicConfigActive), tags)
-  metricsGroup.newGauge(LeaderTransferEnabled,
-    () => enabled(config.liProtocolBridgeLeaderTransferActive), tags)
-  metricsGroup.newGauge(LegacyRequestMetricsEnabled,
-    () => enabled(config.liProtocolBridgeLegacyRequestMetricsActive), tags)
-  metricsGroup.newGauge(LogTruncationMetricsEnabled,
-    () => enabled(config.liProtocolBridgeLogTruncationMetricsActive), tags)
+  private val registrationEnabled = config.liProtocolBridgeConfigMetricsActive
+  if (registrationEnabled) {
+    metricsGroup.newGauge(ConfigMetricsEnabled, () => 1, tags)
+    metricsGroup.newGauge(ModeEnabled, () => enabled(config.liProtocolBridgeModeActive), tags)
+    metricsGroup.newGauge(TopicDeletionStateCleanupEnabled,
+      () => enabled(config.liProtocolBridgeTopicDeletionStateCleanupActive), tags)
+    metricsGroup.newGauge(FollowerRecoveryEnabled,
+      () => enabled(config.liProtocolBridgeFollowerRecoveryActive), tags)
+    metricsGroup.newGauge(RecommendedLeaderElectionEnabled,
+      () => enabled(config.liProtocolBridgeRecommendedElectionActive), tags)
+    metricsGroup.newGauge(ExcludePartitionsEnabled,
+      () => enabled(config.liProtocolBridgeExcludePartitionsActive), tags)
+    metricsGroup.newGauge(MoveControllerEnabled,
+      () => enabled(config.liProtocolBridgeMoveControllerActive), tags)
+    metricsGroup.newGauge(ShutdownSafetyOverrideEnabled,
+      () => enabled(config.liProtocolBridgeShutdownSafetyOverrideActive), tags)
+    metricsGroup.newGauge(PreferredControllerEnabled,
+      () => enabled(config.liProtocolBridgePreferredControllerActive), tags)
+    metricsGroup.newGauge(FederatedTopicsEnabled,
+      () => enabled(config.liProtocolBridgeFederatedTopicsActive), tags)
+    metricsGroup.newGauge(RackIdMapperEnabled,
+      () => enabled(config.liProtocolBridgeRackIdMapperActive), tags)
+    metricsGroup.newGauge(ZookeeperPaginationEnabled,
+      () => enabled(config.liZookeeperPaginationEnable), tags)
+    metricsGroup.newGauge(DynamicTopicDeletionEnabled,
+      () => enabled(config.liProtocolBridgeDynamicTopicDeletionActive), tags)
+    metricsGroup.newGauge(ControllerInitializationThreads,
+      () => config.liNumControllerInitThreads, tags)
+    metricsGroup.newGauge(ProduceRequestInstrumentationEnabled,
+      () => enabled(config.liProtocolBridgeProduceRequestInstrumentationActive), tags)
+    metricsGroup.newGauge(RequestMetricBucketsEnabled,
+      () => enabled(config.liProtocolBridgeRequestMetricBucketsActive), tags)
+    metricsGroup.newGauge(RequestChannelWatchdogEnabled,
+      () => enabled(config.liProtocolBridgeRequestChannelWatchdogActive), tags)
+    metricsGroup.newGauge(MinimumLogRollEnabled,
+      () => enabled(config.liProtocolBridgeMinimumLogRollActive), tags)
+    metricsGroup.newGauge(ReassignmentCancellationSafetyEnabled,
+      () => enabled(config.liProtocolBridgeReassignmentCancellationSafetyActive), tags)
+    metricsGroup.newGauge(ListOffsetsInstrumentationEnabled,
+      () => enabled(config.liProtocolBridgeListOffsetsInstrumentationActive), tags)
+    metricsGroup.newGauge(StaticDefaultQuotasEnabled,
+      () => enabled(config.liProtocolBridgeStaticDefaultQuotasActive), tags)
+    metricsGroup.newGauge(ReplicaRequestTimeoutEnabled,
+      () => enabled(config.liProtocolBridgeReplicaRequestTimeoutActive), tags)
+    metricsGroup.newGauge(OffsetsTopicConfigEnabled,
+      () => enabled(config.liProtocolBridgeOffsetsTopicConfigActive), tags)
+    metricsGroup.newGauge(LeaderTransferEnabled,
+      () => enabled(config.liProtocolBridgeLeaderTransferActive), tags)
+    metricsGroup.newGauge(LegacyRequestMetricsEnabled,
+      () => enabled(config.liProtocolBridgeLegacyRequestMetricsActive), tags)
+    metricsGroup.newGauge(LogTruncationMetricsEnabled,
+      () => enabled(config.liProtocolBridgeLogTruncationMetricsActive), tags)
+  }
 
   private def enabled(value: Boolean): Int = if (value) 1 else 0
 
   override def close(): Unit = {
-    MetricNames.foreach { name =>
+    if (registrationEnabled) MetricNames.foreach { name =>
       metricsGroup.removeMetric(name, tags)
     }
   }
diff --git a/core/src/test/scala/unit/kafka/server/LiProtocolBridgeMetricsTest.scala b/core/src/test/scala/unit/kafka/server/LiProtocolBridgeMetricsTest.scala
index fe946cc031e7a..b894d7754342d 100644
--- a/core/src/test/scala/unit/kafka/server/LiProtocolBridgeMetricsTest.scala
+++ b/core/src/test/scala/unit/kafka/server/LiProtocolBridgeMetricsTest.scala
@@ -20,7 +20,7 @@ import com.yammer.metrics.core.Gauge
 import kafka.utils.TestUtils
 import org.apache.kafka.server.config.ReplicationConfigs
 import org.apache.kafka.server.metrics.KafkaYammerMetrics
-import org.junit.jupiter.api.Assertions.{assertEquals, assertTrue}
+import org.junit.jupiter.api.Assertions.{assertEquals, assertFalse, assertTrue}
 import org.junit.jupiter.api.Test
 
 import java.util.Properties
@@ -28,11 +28,58 @@ import scala.jdk.CollectionConverters._
 
 class LiProtocolBridgeMetricsTest {
 
+  @Test
+  def testNoMetricsWithoutOptIn(): Unit = {
+    val brokerId = 988
+    val config = KafkaConfig(TestUtils.createBrokerConfig(brokerId, TestUtils.MockZkConnect))
+    val metrics = new LiProtocolBridgeMetrics(config)
+    try assertTrue(metricValues(brokerId).isEmpty)
+    finally metrics.close()
+  }
+
+  @Test
+  def testGateRequiresRestartAndDisabledClosePreservesOtherMetrics(): Unit = {
+    val brokerId = 989
+    val props = TestUtils.createBrokerConfig(brokerId, TestUtils.MockZkConnect)
+    props.put(KafkaConfig.LiProtocolBridgeConfigMetricsEnableProp, "true")
+    val config = KafkaConfig(props)
+    config.dynamicConfig.initialize(None, None)
+    val metrics = new LiProtocolBridgeMetrics(config)
+    try {
+      assertFalse(DynamicBrokerConfig.AllDynamicConfigs.contains(KafkaConfig.LiProtocolBridgeConfigMetricsEnableProp))
+      val update = new Properties
+      update.put(KafkaConfig.LiProtocolBridgeConfigMetricsEnableProp, "false")
+      config.dynamicConfig.updateDefaultConfig(update)
+      assertTrue(config.liProtocolBridgeConfigMetricsActive)
+      props.remove(KafkaConfig.LiProtocolBridgeConfigMetricsEnableProp)
+      new LiProtocolBridgeMetrics(KafkaConfig(props)).close()
+      assertEquals(LiProtocolBridgeMetrics.MetricNames.toSet, metricValues(brokerId).keySet)
+      assertEquals(1, metricValues(brokerId)(LiProtocolBridgeMetrics.ConfigMetricsEnabled))
+    } finally metrics.close()
+    assertTrue(metricValues(brokerId).isEmpty)
+  }
+
+  @Test
+  def testKRaftDoesNotRegisterBridgeMetrics(): Unit = {
+    val props = new Properties
+    Map("broker.id" -> "990", "node.id" -> "990", "process.roles" -> "broker,controller",
+      "controller.quorum.voters" -> "990@localhost:19093", "controller.listener.names" -> "CONTROLLER",
+      "listeners" -> "PLAINTEXT://localhost:19092,CONTROLLER://localhost:19093",
+      "listener.security.protocol.map" -> "PLAINTEXT:PLAINTEXT,CONTROLLER:PLAINTEXT",
+      KafkaConfig.LiProtocolBridgeConfigMetricsEnableProp -> "true").foreach { case (key, value) => props.put(key, value) }
+    val config = KafkaConfig(props)
+    assertFalse(config.liProtocolBridgeConfigMetricsActive)
+    val metrics = new LiProtocolBridgeMetrics(config)
+    try assertTrue(metricValues(990).isEmpty)
+    finally metrics.close()
+  }
+
   @Test
   def testMetricsFollowDynamicFlags(): Unit = {
     val brokerId = 987
     val brokerProps = TestUtils.createBrokerConfig(brokerId, TestUtils.MockZkConnect)
     brokerProps.put(ReplicationConfigs.INTER_BROKER_PROTOCOL_VERSION_CONFIG, "3.0")
+    brokerProps.put(KafkaConfig.LiProtocolBridgeConfigMetricsEnableProp, "true")
     val config = KafkaConfig(brokerProps)
     config.dynamicConfig.initialize(None, None)
     val metrics = new LiProtocolBridgeMetrics(config)
@@ -40,8 +87,10 @@ class LiProtocolBridgeMetricsTest {
     try {
       val initialValues = metricValues(brokerId)
       assertEquals(1, initialValues(LiProtocolBridgeMetrics.ControllerInitializationThreads))
-      assertTrue(initialValues.filterNot(_._1 == LiProtocolBridgeMetrics.ControllerInitializationThreads)
-        .values.forall(_ == 0))
+      assertEquals(1, initialValues(LiProtocolBridgeMetrics.ConfigMetricsEnabled))
+      assertTrue(initialValues.filterNot { case (name, _) =>
+        name == LiProtocolBridgeMetrics.ControllerInitializationThreads || name == LiProtocolBridgeMetrics.ConfigMetricsEnabled
+      }.values.forall(_ == 0))
 
       val props = new Properties
       Seq(
diff --git a/tests/bin/README.li-bridge.md b/tests/bin/README.li-bridge.md
index 6a1271148b4ea..a0958f7c58d99 100644
--- a/tests/bin/README.li-bridge.md
+++ b/tests/bin/README.li-bridge.md
@@ -51,6 +51,10 @@ The wrapper's declared Kafka and Scala versions must match the 3.9 archive. Ever
 
 Source compilation and focused source tests still run. `BRIDGE_VERIFY_FULL=1` adds the complete clients, server, and storage suites. The mixed-process test runs the retained archives.
 
+## Diagnostic metrics opt-in
+
+Both broker generations default `li.protocol.bridge.config.metrics.enable` to false. Set it to true in each broker's startup configuration to register the bridge-state gauges. The setting is ZooKeeper-only and requires a restart. It does not enable protocol or data-path behavior; those flags remain separate. The process runner enables it explicitly, and preflight requires it in every migration phase, including dormant mode. A disabled instance does not remove another instance's registered gauges during cleanup.
+
 ## Evidence and resume
 
 Archives are copied under `EVIDENCE_DIR/archives` with content-addressed names. `archive-30.json` and `archive-39.json` record versions, source metadata, and jar hashes. `wrapper-artifacts.json` records the wrapper's resolved Kafka files.
diff --git a/tests/bin/li_bridge_contract.py b/tests/bin/li_bridge_contract.py
index 64fd161b854ed..e6f5aa595ebf5 100644
--- a/tests/bin/li_bridge_contract.py
+++ b/tests/bin/li_bridge_contract.py
@@ -27,10 +27,12 @@
 SCENARIO_REVISION = 4
 MODE = "li.protocol.bridge.mode.enable"
 TOPIC_CLEANUP = "li.protocol.bridge.topic.deletion.state.cleanup.enable"
+CONFIG_METRICS = "li.protocol.bridge.config.metrics.enable"
 # The Kafka registry and metric-name set are checked against this table by tests.
 # (config suffix, effective metric, retained in the minimum compatibility profile)
 FEATURES = (
     ("mode", "ModeEnabled", True),
+    ("config.metrics", "ConfigMetricsEnabled", True),
     ("topic.deletion.state.cleanup", "TopicDeletionStateCleanupEnabled", True),
     ("follower.recovery", "FollowerRecoveryEnabled", True),
     ("recommended.leader.election", "RecommendedLeaderElectionEnabled", True),
@@ -88,7 +90,7 @@ def required_gates(phase, generation, all_gates=False):
     if generation == "3.0":
         # The dormant binary may still have every new flag off. Enable cleanup before
         # entering the common protocol, then keep it enabled through the native bake.
-        return () if phase == "dormant" else (TOPIC_CLEANUP,)
+        return (CONFIG_METRICS,) if phase == "dormant" else (CONFIG_METRICS, TOPIC_CLEANUP)
     gates = BRIDGE_GATES if all_gates else MIXED_REQUIRED_GATES
     return tuple(gate for gate in gates if gate != MODE)
 
diff --git a/tests/bin/li_bridge_mixed_cluster_smoke.py b/tests/bin/li_bridge_mixed_cluster_smoke.py
index 50f260e980aae..6a5199c00e313 100644
--- a/tests/bin/li_bridge_mixed_cluster_smoke.py
+++ b/tests/bin/li_bridge_mixed_cluster_smoke.py
@@ -37,7 +37,7 @@
 from pathlib import Path
 
 from li_bridge_artifacts import file_sha256, snapshot_archive
-from li_bridge_contract import CONTRACT_VERSION, MODE, PHASES, TOPIC_CLEANUP, scenario_spec
+from li_bridge_contract import CONFIG_METRICS, CONTRACT_VERSION, MODE, PHASES, TOPIC_CLEANUP, scenario_spec
 
 SCRIPT_DIR = Path(__file__).resolve().parent
 
@@ -206,7 +206,8 @@ def start_broker(self, identifier, generation, mode=True, ibp="3.0"):
             "transaction.state.log.min.isr": 1, "controlled.shutdown.enable": "true",
             "controlled.shutdown.max.retries": 3, "controlled.shutdown.retry.backoff.ms": 1000,
             "delete.topic.enable": "true", "log.segment.delete.delay.ms": 100,
-            "inter.broker.protocol.version": ibp, MODE: str(mode).lower(), TOPIC_CLEANUP: "true",
+            "inter.broker.protocol.version": ibp, MODE: str(mode).lower(),
+            CONFIG_METRICS: "true", TOPIC_CLEANUP: "true",
             "remote.log.storage.system.enable": "false", "li.drop.corrupted.files.enable": "false",
             "li.leader.election.on.corruption.wait.ms": 0,
         }
diff --git a/tests/unit/li_bridge_preflight_test.py b/tests/unit/li_bridge_preflight_test.py
index ee339368b85d2..837f331bafe85 100644
--- a/tests/unit/li_bridge_preflight_test.py
+++ b/tests/unit/li_bridge_preflight_test.py
@@ -106,12 +106,30 @@ def test_require_all_gates_reports_missing_operational_gates(self):
             Path("broker.properties"), self.mixed_properties(), "mixed", True)
         self.assertEqual(len(PREFLIGHT.BRIDGE_GATES) - len(PREFLIGHT.MIXED_REQUIRED_GATES), len(issues))
 
+    def test_config_metrics_opt_in_is_required_in_every_phase(self):
+        gate = "li.protocol.bridge.config.metrics.enable"
+        for phase, (ibp, mode, generations) in PREFLIGHT.PHASES.items():
+            for generation in generations:
+                for value in (None, "false"):
+                    with self.subTest(phase=phase, generation=generation, value=value):
+                        properties = {name: "true" for name in PREFLIGHT.BRIDGE_GATES}
+                        properties.update({"broker.id": "0", "inter.broker.protocol.version": ibp,
+                                           PREFLIGHT.MODE: str(mode).lower()})
+                        self.assertEqual([], PREFLIGHT.inspect_config(Path("broker"), properties, phase, True, generation)[0])
+                        if value is None:
+                            properties.pop(gate)
+                        else:
+                            properties[gate] = value
+                        issues, _ = PREFLIGHT.inspect_config(Path("broker"), properties, phase, True, generation)
+                        self.assertTrue(any(gate in issue for issue in issues))
+
     def test_legacy_broker_requires_bridge_mode_and_topic_cleanup(self):
         properties = {
             "broker.id": "0",
             "inter.broker.protocol.version": "3.0",
             "li.protocol.bridge.mode.enable": "true",
             "li.protocol.bridge.topic.deletion.state.cleanup.enable": "true",
+            "li.protocol.bridge.config.metrics.enable": "true",
         }
         issues, details = PREFLIGHT.inspect_config(
             Path("legacy.properties"), properties, "mixed", True, "3.0")

From e80460f79782b3965d87b68179e29f52f9e82f9b Mon Sep 17 00:00:00 2001
From: Jon Bringhurst <jbringhu@jbringhu-mn3034.linkedin.biz>
Date: Fri, 11 Sep 2026 11:17:36 -0700
Subject: [PATCH 2/2] docs: record the diagnostic opt-in and completed
 revision-4 process run

---
 docs/ops/li-bridge-review-comments.md |  5 ++--
 docs/ops/li-bridge-review.md          | 36 ++++++++++++++++++---------
 docs/ops/li-bridge-upgrade.md         | 17 +++++++------
 3 files changed, 37 insertions(+), 21 deletions(-)

diff --git a/docs/ops/li-bridge-review-comments.md b/docs/ops/li-bridge-review-comments.md
index f3a92e89f5015..83891b6ad7168 100644
--- a/docs/ops/li-bridge-review-comments.md
+++ b/docs/ops/li-bridge-review-comments.md
@@ -26,7 +26,8 @@ All 62 original threads have replies with published source decisions. A fresh Gr
 | [F18: offline replica misses deletion](https://github.com/linkedin/kafka/pull/576#issuecomment-5629453741) | Paired complete-image recovery in 583/584; unassigned/leaderless/incremental tests and exact records after promotion. | Assignment does not establish topic identity; also require F19. |
 | [F19: recreated topic already assigned to returning replica](https://github.com/linkedin/kafka/pull/584#issuecomment-5631079569) | Paired identity recovery in 585/586; missing/zero IDs, errors, retries, current/future copies and mixed-batch tests. [Qualification update](https://github.com/linkedin/kafka/pull/586#issuecomment-5638005151). | All four revision-4 record checks passed, but full final-source and wrapper qualification remain open. |
 | [F20: rotated protocol logs omitted](https://github.com/linkedin/kafka/pull/584#issuecomment-5638004839) | PR 584 retains and scans hourly rotations; 586 is restacked on it. New positive and negative tests fail before the fix and pass afterward. | The previous failed CI job stays failed. Re-run the corrected collector. |
-| [F21: churn exits during controller movement](https://github.com/linkedin/kafka/pull/586#issuecomment-5638182322) | PR 584 fixes the workload retry policy without changing the upstream broker response. Tests with both client archives require controller retries and reject data/auth/record errors; the process setup runs them for both generations. | All 71 Python tests pass. Full migration and wrapper qualification remain required. |
+| [F21: churn exits during controller movement](https://github.com/linkedin/kafka/pull/586#issuecomment-5638182322) | PR 584 fixes the workload retry policy without changing the upstream broker response. [Test and code update](https://github.com/linkedin/kafka/pull/584#issuecomment-5638482746). | The complete revision-4 process run and audit pass; final F22/wrapper qualification remains required. |
+| F22: bridge-state MBeans register by default | Paired [588](https://github.com/linkedin/kafka/pull/588)/[589](https://github.com/linkedin/kafka/pull/589) add a default-off diagnostics flag. Tests cover disabled registration, enabled readings, restart scope, KRaft and cleanup. | All 72 Python tests pass. The wrapper mapping still needs matching-jar qualification. |
 
 ## PR 541
 
@@ -108,7 +109,7 @@ All 62 original threads have replies with published source decisions. A fresh Gr
 | [7](https://github.com/linkedin/kafka/pull/551#discussion_r3927365100) | Synchronize close and clear the metrics map. | RequestChannel.Metrics.close uses the same monitor as apply. | [reply](https://github.com/linkedin/kafka/pull/551#discussion_r3984230263) |
 | [8](https://github.com/linkedin/kafka/pull/551#discussion_r3927365140) | Prefix the watchdog histogram with the request-channel metric prefix. | RequestChannel and RequestChannelWatchdogTest. | [reply](https://github.com/linkedin/kafka/pull/551#discussion_r3984230448) |
 | [9](https://github.com/linkedin/kafka/pull/551#discussion_r3927365177) | Derive the watchdog check interval from the configured timeout. | KafkaServerTest.testRequestChannelWatchdogIntervalTracksConfiguredTimeout. | [reply](https://github.com/linkedin/kafka/pull/551#discussion_r3984230674) |
-| [10](https://github.com/linkedin/kafka/pull/551#discussion_r3927365245) | Remove the hard-coded count. Check uniqueness and compare the actual registry with the shared Python contract. | LiProtocolBridgeConfigTest; preflight/scenario registry-consistency tests; 23 gates are currently registered. | [reply](https://github.com/linkedin/kafka/pull/551#discussion_r3984230900) |
+| [10](https://github.com/linkedin/kafka/pull/551#discussion_r3927365245) | Remove the hard-coded count. Check uniqueness and compare the actual registry with the shared Python contract. | LiProtocolBridgeConfigTest; preflight/scenario registry-consistency tests; 24 gates are currently registered. | [reply](https://github.com/linkedin/kafka/pull/551#discussion_r3984230900) |
 | [11](https://github.com/linkedin/kafka/pull/551#discussion_r3927475761) | Cache the total topic-name length and update it with topic membership. | ControllerContextTest.testTopicNameLengthTotalTracksTopicChanges; controller gauge reads the cached total. | [reply](https://github.com/linkedin/kafka/pull/551#discussion_r3984231122) |
 
 ## PR 552
diff --git a/docs/ops/li-bridge-review.md b/docs/ops/li-bridge-review.md
index 3133019b1af46..b1b45dcbc08f2 100644
--- a/docs/ops/li-bridge-review.md
+++ b/docs/ops/li-bridge-review.md
@@ -19,15 +19,15 @@ limitations under the License.
 
 ## Current verdict
 
-**Do not deploy this candidate. The coverage audit found two offline topic-name-reuse failures after an earlier full verifier passed.** Both paired repairs are published in 583–586, but final qualification remains incomplete. The current inventory has 37 open PRs, all below 1,000 changed lines at readback. An earlier green bundle or CI run does not cover a scenario it never exercised.
+**Do not deploy this candidate. The coverage audit found two offline topic-name-reuse failures after an earlier full verifier passed.** Both paired repairs are published in 583–586, but final qualification remains incomplete. The current inventory has 39 open PRs. The complete revision-4 process run now passes, including all four offline-reuse record checks. The later diagnostic-metrics opt-in and matching wrapper still need final qualification. An earlier green bundle or CI run does not cover a scenario it never exercised.
 
 The findings below record what the initial review and later tests found. Instructions in an original finding describe the repair that was needed; use the current disposition and evidence sections for status. This is not a line-by-line approval of every Kafka change.
 
 ### Reviewed revisions
 
 - Workspace plan: `LI-3.0-TO-3.9-ROLLING-UPGRADE-PLAN.md`. Canonical runbook: `docs/ops/li-bridge-upgrade.md`.
-- Current 3.9 behavior and tools: PR 586, `d67d7b28bc379416d7c6266558a933b772b08460`. This documentation follow-up is based on that revision.
-- Current 3.0 source: PR 585, `1a5d02403bd97758f5ac3f8ddc776d0ad528405b`.
+- Current 3.9 behavior change: PR 589, `43b03f35049efa5510a40a6a46bb2bf78bf4750a`; its documentation update follows that commit.
+- Current 3.0 source: PR 588, `ec940efaacdebf38b82c786e61ac46e91eecd5da`.
 - CI: PR 558, `3e799b08ea`; PR 559, `a86214e2da`.
 - Wrapper: `1a9ecccf`, including the ACL test fix `6ddf2a87`.
 
@@ -271,7 +271,15 @@ The revision-4 run stopped after the native client checkpoint when topic creatio
 
 A focused test against both real client archives reproduces that failure. It also exposes a second gap: `KafkaStorageException` and `CorruptRecordException` inherit `RetriableException`, so the old broad policy could silently retry data errors. PR 584's helper now explicitly retries controller transitions but rejects storage and corrupt-record errors. Authorization, malformed requests/configuration, oversized records and unexpected failures remain fatal. Broker response codes and client libraries are unchanged.
 
-The positive/negative classifier test passes against both archives and is run during process setup for both generations. All 71 Python tests pass, including a test that checks this setup wiring. Record comparisons, retry delays, phase-progress checks and deadlines are unchanged. The earlier failed run stays failed; a complete rerun is still required.
+The positive/negative classifier test passes against both archives and is run during process setup for both generations. All 71 Python tests pass, including a test that checks this setup wiring. Record comparisons, retry delays, phase-progress checks and deadlines are unchanged. The complete rerun now passes in `/tmp/li-scenario-4-churn-fixed`. The earlier failed run stays failed. Final qualification must also cover the later F22 change and matching wrapper.
+
+### F22 — P2: Bridge-state MBeans register without an opt-in
+
+Both generations constructed `LiProtocolBridgeMetrics` and registered new MBeans even with bridge behavior flags disabled. Diagnostics are not exempt from the requirement that upgrade behavior be config-gated.
+
+PRs 588/589 add `li.protocol.bridge.config.metrics.enable`, default false, ZooKeeper-only and restart-scoped. Enabled diagnostics still report disabled behavior flags without activating them. The existing constructors remain available. A disabled instance does not remove an enabled instance's gauges during cleanup.
+
+The default-off test failed on both previous implementations and passes with the repair. Tests also cover enabled readings, dynamic behavior flags, ignored live updates to the restart-only setting, KRaft exclusion and cleanup. The selected 3.0 and 3.9 suites passed 22 and 39 tests respectively. All 72 Python tests pass, including missing/false opt-in rejection in every migration phase. The process profile enables diagnostics explicitly. The wrapper mapping and its new negative test still need qualification with matching staged jars.
 
 ## PR dispositions and dependency audit
 
@@ -285,6 +293,7 @@ Every PR below has a distinct migration or CI purpose. Keep these scopes, but do
 | [577](https://github.com/linkedin/kafka/pull/577) | 3.0 unhosted-log cleanup — storage/controller |
 | [583](https://github.com/linkedin/kafka/pull/583) | 3.0 complete-image log recovery — storage/controller |
 | [585](https://github.com/linkedin/kafka/pull/585) | 3.0 topic-identity validation — storage/controller |
+| [588](https://github.com/linkedin/kafka/pull/588) | 3.0 diagnostic-metrics opt-in — observability |
 | [558](https://github.com/linkedin/kafka/pull/558) | 3.9 CI/publication — release engineering |
 | [543](https://github.com/linkedin/kafka/pull/543) | 3.9 outbound bridge — protocol/controller |
 | [544](https://github.com/linkedin/kafka/pull/544) | old wire/client/recovery compatibility — protocol/replication |
@@ -316,6 +325,7 @@ Every PR below has a distinct migration or CI purpose. Keep these scopes, but do
 | [584](https://github.com/linkedin/kafka/pull/584) | 3.9 complete-image recovery, offline-reuse tests and log retention — storage/verification |
 | [586](https://github.com/linkedin/kafka/pull/586) | 3.9 topic identity and scenario-revision-4 qualification — storage/verification |
 | [587](https://github.com/linkedin/kafka/pull/587) | current findings, evidence limits and completion checklist — operations/review |
+| [589](https://github.com/linkedin/kafka/pull/589) | 3.9 diagnostic-metrics opt-in and admission checks — observability/verification |
 
 Merge CI 558/559 into their own release branches first. Then retarget the upgrade stack bottoms as described in the runbook. Never force a Git dependency between the 3.0 and 3.9 CI branches. When reordering again, change PR bases before pushing a head that becomes an ancestor of its former base; GitHub can otherwise auto-close and delete that branch.
 
@@ -331,22 +341,22 @@ This prompt-to-artifact checklist separates observed results from open requireme
 
 | Requirement | Artifact and verification surface | Evidence / open work |
 |---|---|---|
-| Review the named plan and every open public PR | `LI-3.0-TO-3.9-ROLLING-UPGRADE-PLAN.md`; canonical runbook; GitHub inventory | 37 open PRs, including this documentation follow-up, are listed in both tables. Recheck the exact PR set after any further publication. |
-| Explain scope and dependencies | PR responsibility table, heads/bases, stack membership | Separate protocol, handlers, storage metrics, runner, auditor, verifier and release-gate layers. Stack 582 has 30 upgrade PRs; stack 581 has five. CI 558/559 remain on independent release histories. No release branch was merged. |
-| Apply the requested label | GitHub labels | All 35 upgrade PRs have `kafka-upgrade-august-2026`; CI 558/559 do not. |
-| Keep PRs below 1,000 changed lines, preferably near 500 | Additions plus deletions, not file length | All 37 diffs passed the limit. The largest established diffs are 981 and 963 lines. This update stays separate from the original 863-line docs PR. |
+| Review the named plan and every open public PR | `LI-3.0-TO-3.9-ROLLING-UPGRADE-PLAN.md`; canonical runbook; GitHub inventory | 39 open PRs are listed in both tables. Recheck the exact PR set after any further publication. |
+| Explain scope and dependencies | PR responsibility table, heads/bases, stack membership | Separate protocol, handlers, storage metrics, runner, auditor, verifier and release-gate layers. Stack 582 has 31 upgrade PRs; stack 581 has six. CI 558/559 remain on independent release histories. No release branch was merged. |
+| Apply the requested label | GitHub labels | All 37 upgrade PRs have `kafka-upgrade-august-2026`; CI 558/559 do not. |
+| Keep PRs below 1,000 changed lines, preferably near 500 | Additions plus deletions, not file length | Established diffs passed the limit; recheck the new metrics follow-ups after publication. The largest established diffs are 981 and 963 lines. |
 | Use plain, direct English | Plan, review, comment replies and workflow comments | Final wording/link review remains required. Historical findings are not current deployment instructions. |
-| Gate every Kafka behavior change | `KafkaConfig`, `DynamicBrokerConfig`, runtime call sites, metrics, wrapper mapping | 23 default-off 3.9 gates. F18/F19 use the cleanup gate; retry repair uses bridge mode. Dedicated disabled/activation tests pass. The final per-change call-site audit remains open. |
+| Gate every Kafka behavior change | `KafkaConfig`, `DynamicBrokerConfig`, runtime call sites, metrics, wrapper mapping | 24 default-off 3.9 gates. F18/F19 use the cleanup gate; ISR retry repair uses bridge mode; F22 separately gates diagnostic registration. Dedicated disabled/activation tests pass. The final per-change call-site audit remains open. |
 | Select symmetric v2/v5/v1 control | Schemas, controller selectors, wire fixtures and retained logs | Both generations have fixtures and real-process coverage. F20 makes rotated log checks fail closed too. Final-source process qualification remains required. |
 | Fence activation and preserve callbacks | `RequestSendThreadBridgeTest` | Blocked dequeue, sustained queue and admitted-deletion callback tests pass. Controller restart remains mandatory. |
-| Enforce all six phases and unchanged clients | `li_bridge_contract.py`, preflight, persistent client/Streams/Connect and private-API helpers | 71 Python tests pass. Native control stays at IBP 3.0 before the separate IBP roll. One 3.0 archive is not the deployed client/tool floor. |
+| Enforce all six phases and unchanged clients | `li_bridge_contract.py`, preflight, persistent client/Streams/Connect and private-API helpers | 72 Python tests pass. Native control stays at IBP 3.0 before the separate IBP roll. One 3.0 archive is not the deployed client/tool floor. |
 | Prove persisted rollback and recovery | Process runner, record helper, timings and JUnit | Historical full run covers canary/all-3.9 rollback, cancellation, crashes and truncation. Repeat against the final source; registration or ISR alone is not proof. |
 | Prove deletion and name reuse | `TopicDeletionManager`, `ZkMetadataCache`, `ReplicaManager`, `BridgeTopicIdentity` | Gated F12/F13/F15/F17/F18/F19 repairs have tests. Revision 4 requires four post-promotion record checks. They passed in the current scoped checks, but do not replace complete qualification. |
 | Handle version-changing ISR retries | `BridgeAlterPartitionRetryTest` | One queued builder crosses versions 3/1 and activation without reusing mutated request data. Earlier failed logs remain failed evidence. |
 | Collect real runtime/configuration/state | Live inventory, runtime probe and negative-input tests | Disposable-cluster collection passes. Production binaries, settings, state dispositions, owners and client/tool floor remain required inputs. |
 | Qualify pagination with the loaded runtime | `KafkaZkClient`, five vendor-client tests and release runtime probe | Startup rejects an unsupported client. Vendor tests pass. The actual deployed client/Jute/server pairing remains a release gate. |
 | Follow Google shell style, including comments | Four wrappers, all extracted workflow Bash blocks, ShellCheck, shfmt, syntax/length checks | The refreshed audit checks exact reviewed source revisions: 54 changed-workflow blocks and four wrappers pass ShellCheck, shfmt, syntax, no-tab and 80-column checks. Unmodified upstream Docker workflows are outside these PRs; the broader diagnostic results are retained separately. |
-| Preserve wrapper/API compatibility | Factory mapping tests, ACL tests, complete wrapper suite and jar comparison | Historical 132-test suite passed with matching jars and stable hashes. Current dependency-spec refresh is blocked on internal network/VPN; no TTL or artifact-identity bypass is allowed. |
+| Preserve wrapper/API compatibility | Factory mapping tests, ACL tests, complete wrapper suite and jar comparison | Historical 132-test suite passed with matching jars and stable hashes. The required Mint refresh has now produced a fresh dependency spec. Test the new metrics mapping against matching staged jars; no TTL or artifact-identity bypass is allowed. |
 | Qualify real archives and reject incomplete evidence | `verify_li_bridge.sh`, `audit_li_bridge_evidence.py`, `verify_li_bridge_release.sh` and negative fixtures | Earlier full verifier passed, but predates F18/F19. Final full-source verification and release-guard inputs/negative behavior remain open. |
 | Address every review comment with evidence | Comment ledger, source/test decisions, GraphQL readback | All original 62 replies verified across 37 PRs; no new review threads. Later issue findings have published fixes and explicit qualification limits. Re-fetch after final publication. |
 | Keep the three documents consistent | Workspace files and `docs/ops/li-bridge-{upgrade,review,review-comments}.md` | This follow-up synchronizes the records. Verify relative links and actual PR/source/evidence state before calling the review complete. |
@@ -373,7 +383,9 @@ Later evidence supersedes the inventory and coverage limits of those historical
 - `/tmp/li-churn-retry-before-{3.0,3.9}.log`: real client error classes expose the missing controller retry and incorrectly retryable storage/corruption errors. `/tmp/li-churn-retry-after-{3.0,3.9}.log` and `/tmp/li-churn-retry-restacked-python.log` pass with the repair.
 - `/tmp/li-review-readback-result.json`: 37 PRs, 62 original threads, no missing/mismatched replies and no new threads at that readback.
 
-No complete bundle yet covers the final identity repair, collector and wrapper together. Failed runs remain failed. A source, archive or scenario change must be checked against the fingerprint before reuse.
+The complete revision-4 process run `/tmp/li-scenario-4-churn-fixed` passed on clean `d8f255f8a4` / `1a5d02403b`, with all four name-reuse checks, unchanged source and an issue-free process audit. It includes F20/F21 but predates F22. The real `mint --no-metrics dependency create-dependency-spec --detect-variant --overwrite` command has now refreshed the wrapper metadata successfully. An invocation without `--overwrite` returned success without refreshing the expired file; that no-op was not accepted as freshness evidence.
+
+No complete bundle yet covers the final metrics opt-in, both binaries and wrapper together. Failed runs remain failed. A source, archive or scenario change must be checked against the fingerprint before reuse.
 
 ### Inputs still required before production
 
diff --git a/docs/ops/li-bridge-upgrade.md b/docs/ops/li-bridge-upgrade.md
index f9deb20cdd1b5..d0ab9013736c3 100644
--- a/docs/ops/li-bridge-upgrade.md
+++ b/docs/ops/li-bridge-upgrade.md
@@ -23,7 +23,7 @@ limitations under the License.
 
 The implementation base is Apache **3.9.2** with the reviewed LI bridge stack. Pin the final internal `3.9.2.N`, matching `3.0.1.N`, wrapper commit, archive checksums, JDKs and ZooKeeper runtime in the release record. A maintenance-baseline change requires a new qualification run; do not substitute a newer tag during rollout.
 
-The current implementation is the split stack through `3.9-li-bridge/topic-identity-recovery`, with the companion `3.0-li-bridge/topic-identity-recovery` branch. It is not the closed aggregate PR 542. The canonical mergeable runbook is `docs/ops/li-bridge-upgrade.md`; the workspace copy is `LI-3.0-TO-3.9-ROLLING-UPGRADE-PLAN.md`. The `3.9-li-bridge/review-refresh` branch updates the documentation after the behavior fixes. Historical experiments are evidence, not current acceptance criteria.
+The current implementation is the split stack through `3.9-li-bridge/config-metrics-gate`, with the companion `3.0-li-bridge/config-metrics-gate` branch. It is not the closed aggregate PR 542. The canonical mergeable runbook is `docs/ops/li-bridge-upgrade.md`; the workspace copy is `LI-3.0-TO-3.9-ROLLING-UPGRADE-PLAN.md`. The `3.9-li-bridge/review-refresh` branch updates the documentation after the behavior fixes. Historical experiments are evidence, not current acceptance criteria.
 
 **Clients do not change.** The supported producer, consumer, transactional client, Streams application, Connect worker, LI AdminClient and operational-tool artifacts/configuration must remain unchanged across every phase. Discovery must name their deployed version floor and owners. One 3.0 test archive is not proof for every externally deployed client.
 
@@ -70,13 +70,14 @@ For a phase change, supply `--previous-phase` to preflight and retain both repor
 
 ## Feature contract and lifecycle
 
-Kafka's 23 LI compatibility gates default false. The wrapper explicitly enables the production profile. With `--require-all-gates`, every gate below except the phase-dependent mode gate must stay true on 3.9 **including native and final-IBP phases**. Turning bridge mode off is not permission to turn off operational compatibility.
+Kafka's 24 LI compatibility gates default false. The wrapper explicitly enables the production profile. With `--require-all-gates`, every gate below except the phase-dependent mode gate must stay true on 3.9 **including native and final-IBP phases**. Turning bridge mode off is not permission to turn off operational compatibility.
 
-All names below have the prefix `li.protocol.bridge.` and suffix `.enable`. Effective gauges are under `kafka.server:type=LiProtocolBridgeMetrics,broker-id=<id>`.
+All names below have the prefix `li.protocol.bridge.` and suffix `.enable`. Effective gauges are under `kafka.server:type=LiProtocolBridgeMetrics,broker-id=<id>`. Set `config.metrics` to true in each broker's startup configuration before entering the dormant qualification phase. This opt-in registers the diagnostic gauges; it does not enable protocol or data-path behavior.
 
 | Setting suffix | Effective gauge | Change scope | Owner / disposition |
 |---|---|---|---|
 | `mode` | `ModeEnabled` | cluster dynamic; controller restart fence | Protocol: temporary control versions and version-changing ISR retries |
+| `config.metrics` | `ConfigMetricsEnabled` | restart | Observability: explicit registration of bridge-state MBeans on both generations |
 | `topic.deletion.state.cleanup` | `TopicDeletionStateCleanupEnabled` | cluster dynamic; enable on all brokers before controller restart | Controller: clear deletion blocks and replace stale metadata on a new controller epoch |
 | `follower.recovery` | `FollowerRecoveryEnabled` | cluster dynamic | Replication: retain until old recovery callers are retired |
 | `recommended.leader.election` | `RecommendedLeaderElectionEnabled` | cluster dynamic | Controller: retain old election type 2 |
@@ -102,7 +103,7 @@ All names below have the prefix `li.protocol.bridge.` and suffix `.enable`. Effe
 
 Per-broker dynamic overrides of cluster-dynamic compatibility gates are rejected. The smoke profile intentionally enables only wire/tool compatibility and cancellation safety; it is not the complete production wrapper profile. Wrapper qualification and live admission use the full profile.
 
-Two additional settings are not members of the 23-gate bundle: `li.zookeeper.pagination.enable` and `li.num.controller.init.threads`. Record their effective values and `ZookeeperPaginationEnabled` / `ControllerInitializationThreads` gauges. The documented source profile enables pagination and uses ten controller-init threads; confirm live values.
+Two additional settings are not members of the 24-gate bundle: `li.zookeeper.pagination.enable` and `li.num.controller.init.threads`. Record their effective values and `ZookeeperPaginationEnabled` / `ControllerInitializationThreads` gauges. The documented source profile enables pagination and uses ten controller-init threads; confirm live values.
 
 ### Generation-specific behavior
 
@@ -235,7 +236,7 @@ Automatically stop for unexpected control versions, post-fence API 1001 traffic,
 
 ## PR inventory and merge order
 
-All 37 open public PRs are covered below. Upgrade PRs carry `kafka-upgrade-august-2026`; CI foundations 558 and 559 do not. All current diffs are below 1,000 changed lines. These checks do not grant approval to deploy.
+All 39 open public PRs are covered below. Upgrade PRs carry `kafka-upgrade-august-2026`; CI foundations 558 and 559 do not. All current diffs are below 1,000 changed lines. These checks do not grant approval to deploy.
 
 Closed PRs 542 and 555 are superseded. GitHub automatically closed 563 and 564 during the dependency reorder because their new heads were contained in their former base branches. No release branch was merged. Their restored, separate reviews are 579 and 578.
 
@@ -247,6 +248,7 @@ Closed PRs 542 and 555 are superseded. GitHub automatically closed 563 and 564 d
 | [577](https://github.com/linkedin/kafka/pull/577) | 3.0 unhosted-log cleanup — storage/controller |
 | [583](https://github.com/linkedin/kafka/pull/583) | 3.0 complete-image log recovery — storage/controller |
 | [585](https://github.com/linkedin/kafka/pull/585) | 3.0 topic-identity validation — storage/controller |
+| [588](https://github.com/linkedin/kafka/pull/588) | 3.0 diagnostic-metrics opt-in — observability |
 | [558](https://github.com/linkedin/kafka/pull/558) | 3.9 CI/publication — release engineering |
 | [543](https://github.com/linkedin/kafka/pull/543) | 3.9 outbound bridge — protocol/controller |
 | [544](https://github.com/linkedin/kafka/pull/544) | old wire/client/recovery compatibility — protocol/replication |
@@ -278,12 +280,13 @@ Closed PRs 542 and 555 are superseded. GitHub automatically closed 563 and 564 d
 | [584](https://github.com/linkedin/kafka/pull/584) | 3.9 complete-image recovery and offline-reuse tests — storage/verification |
 | [586](https://github.com/linkedin/kafka/pull/586) | 3.9 topic identity and scenario-revision-4 qualification — storage/verification |
 | [587](https://github.com/linkedin/kafka/pull/587) | current findings, evidence limits and completion checklist — operations/review |
+| [589](https://github.com/linkedin/kafka/pull/589) | 3.9 diagnostic-metrics opt-in and admission checks — observability/verification |
 
 Merge 558 into `3.9-li` and 559 into `3.0-li` first. They have different release bases, so do not put them in one dependent Git stack. Rebase/retarget 575 to `3.0-li` and 543 to `3.9-li`; do not merge feature work into temporary CI branches.
 
-The GitHub stack rooted at PR 575 has the 3.0 order **575 → 541 → 577 → 583 → 585**. The stack rooted at PR 543 has the 3.9 order:
+The GitHub stack rooted at PR 575 has the 3.0 order **575 → 541 → 577 → 583 → 585 → 588**. The stack rooted at PR 543 has the 3.9 order:
 
-**543 → 544 → 565 → 545 → 546 → 547 → 548 → 560 → 549 → 550 → 561 → 566 → 551 → 567 → 576 → 568 → 578 → 579 → 552 → 569 → 570 → 571 → 553 → 572 → 554 → 573 → 574 → 584 → 586 → 587**.
+**543 → 544 → 565 → 545 → 546 → 547 → 548 → 560 → 549 → 550 → 561 → 566 → 551 → 567 → 576 → 568 → 578 → 579 → 552 → 569 → 570 → 571 → 553 → 572 → 554 → 573 → 574 → 584 → 586 → 587 → 589**.
 
 Retarget remaining layers after each independent merge. The wrapper branch contains the ACL test repair (`6ddf2a87`) and cleanup mapping/tests (`1a9ecccf`); its source suite passes 132 tests. Add the approved wrapper/dependency/security PR and named deployment-gate owner to the release record.