tests/rptest: exercise the fetch_controller_snapshot_rpc gate

It was previously acknowledged as needing no perturbation coverage, on
the rationale that a gated broker's bootstrap is indistinguishable from
prior behavior. It is distinguishable: exercise the gate in the
unfinalized state by bringing one broker down, changing a
restart-requiring property it misses, then restarting only that broker
with the controller leader still up. While gated the broker boots on its
stale local cache, reports restart-required, and attempts no
controller-snapshot fetch -- exactly the pre-upgrade behavior, which is
what keeps the unfinalized state downgrade-safe.

The whole-cluster restart in _perturb_common cannot exercise this: with
every broker down there is no leader to fetch from and no config delta to
reconcile. Also verify the gate opens post-finalize; the active-state
fetch mechanics are covered by
ClusterConfigMultiNodeBootstrapTest.test_node_delayed_restart.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: dotnwat

tests/rptest/tests/cluster_features_test.py

@@ -1209,7 +1209,13 @@ def test_get_upgrade_status_reports_in_progress(self):
 # exercise or an explicit acknowledgement. Add a feature's name here when you
 # cover it or knowingly skip it; entries for features no longer gated by the
 # current upgrade are harmless extras, so no per-major pruning is needed.
-PERTURB_EXERCISED_FEATURES = frozenset({"tiered_cloud_topics", "shadow_link_role_sync"})
+PERTURB_EXERCISED_FEATURES = frozenset(
+    {
+        "tiered_cloud_topics",
+        "shadow_link_role_sync",
+        "fetch_controller_snapshot_rpc",
+    }
+)
 PERTURB_ACKNOWLEDGED_FEATURES = frozenset(
     {
         # Cluster-linking features, out of scope for this single-cluster test.
@@ -1218,12 +1224,6 @@ def test_get_upgrade_status_reports_in_progress(self):
         # Iceberg extended-mode topic-config gate; exercising it needs Iceberg
         # topic setup orthogonal to the finalization behavior under test.
         "iceberg_extended_mode_config",
-        # Bootstrap-time internal RPC: while gated, a restarting node simply
-        # skips the controller-snapshot fetch and falls back to its local
-        # config cache (indistinguishable from prior behavior, with no
-        # distinctive signal to assert here). Its active-state
-        # behavior is covered by ClusterConfigMultiNodeBootstrapTest.
-        "fetch_controller_snapshot_rpc",
     }
 )
 
@@ -1455,6 +1455,7 @@ def _perturb_v26_1_to_v26_2(self, phase):
             return
         self._exercise_tiered_cloud_topics()
         self._exercise_shadow_link_role_sync()
+        self._exercise_fetch_controller_snapshot_rpc()
         # The other two v26.2-gated features are not exercised by this
         # single-cluster perturbation: shadow_link_sr_api_sync and
         # batch_mirror_topic_status are both cluster-linking features
@@ -1548,6 +1549,115 @@ def _exercise_shadow_link_role_sync(self):
                 "configuring role sync should be gated while unfinalized"
             )
 
+    def _exercise_fetch_controller_snapshot_rpc(self):
+        """Sanity-check that the fetch_controller_snapshot_rpc gate keeps an
+        unfinalized upgrade downgrade-safe.
+
+        This feature is safe because its only new behavior -- a restarting
+        broker fetching a fresh controller snapshot at bootstrap (from the "great
+        config bootstrap fix") -- sits behind a single client-side is_active()
+        check in application::bootstrap_controller_view. While the upgrade is
+        unfinalized the active version is held below the feature's
+        require_version (v26_2_1), so that gate is provably closed: the feature
+        never even becomes available, let alone active. With the gate closed
+        bootstrap_controller_view returns immediately -- it issues no RPC and,
+        crucially, writes no new state (no config-cache write, no feature-table
+        snapshot applied, no wait-for-offset). The broker boots exactly as the
+        pre-upgrade binary did, from its local config cache, so rolling back to
+        the old binary finds nothing on disk it cannot read. Finalization is the
+        one-way door that opens the gate; until it is crossed the new path stays
+        inert and leaves no trace, which is precisely what makes the downgrade
+        safe.
+
+        This test is a sanity check on that safety property -- not a proof of the
+        C++ gate logic, but a check that the *observable* behavior matches it. It
+        simulates the case the feature actually targets, not just any restart:
+        bring one broker down, change a restart-requiring, non-feature-gated
+        property it will miss, then restart only that broker with the other two
+        (and therefore the controller leader) still up. Keeping the leader
+        reachable is what makes this a gate test rather than a
+        leader-reachability test: were the gate wrongly open -- e.g. a future
+        change activating the feature before finalization, or moving the fetch
+        ahead of the is_active() check -- the broker would fetch the missed value
+        before starting services and come back needing no restart (as it does
+        once finalized -- see
+        ClusterConfigMultiNodeBootstrapTest.test_node_delayed_restart). Gated, it
+        must instead boot on the stale value and still report restart-required,
+        having attempted no fetch.
+
+        NOTE: the whole-cluster in-place restart in _perturb_common does not
+        exercise this. Restarting every broker at once leaves no controller
+        leader to fetch from and changes no config, so it cannot tell the gated
+        path from the active one."""
+        assert self._feature_state("fetch_controller_snapshot_rpc") == "unavailable", (
+            "fetch_controller_snapshot_rpc should be unavailable while unfinalized"
+        )
+
+        PROPERTY = "kafka_qdc_idle_depth"  # restart-requiring, not feature-gated
+        NEW_VALUE = 77
+        node = self.redpanda.nodes[-1]
+        node_id = self.redpanda.node_id(node)
+
+        # Down one broker, then change a property it will miss. Quorum (2/3)
+        # survives, so the change commits and a controller leader stays up.
+        self.redpanda.stop_node(node)
+        try:
+            prev = self.admin.get_cluster_config_status()[0]["config_version"]
+            new_version = self.admin.patch_cluster_config(upsert={PROPERTY: NEW_VALUE})[
+                "config_version"
+            ]
+            # Guard against a silent no-op that would make the check vacuous.
+            assert new_version > prev, (
+                f"patching {PROPERTY} did not advance the config version "
+                f"({prev} -> {new_version}); is it already {NEW_VALUE}?"
+            )
+
+            # Restart the broker on the same (HEAD) binary and let it rejoin.
+            self.redpanda.start_node(node)
+            self._wait_for_cluster_settled()
+
+            def restarted_node_status():
+                for s in self.admin.get_cluster_config_status():
+                    if s["node_id"] == node_id:
+                        return s
+                return None
+
+            # Wait until the broker has replayed the controller log up to the
+            # change, so its restart-required verdict is final before we assert.
+            def caught_up():
+                st = restarted_node_status()
+                return st is not None and st["config_version"] >= new_version
+
+            wait_until(
+                caught_up,
+                timeout_sec=60,
+                backoff_sec=1,
+                err_msg="restarted broker did not catch up to the new config version",
+            )
+
+            # Gated: no bootstrap fetch was attempted. This info-level line is
+            # logged only past the feature gate; every pre-finalize boot in this
+            # test is gated, so any match on this node is a real regression.
+            assert not self.redpanda.search_log_node(
+                node, "Fetching controller snapshot from"
+            ), "a gated broker must not attempt the bootstrap controller-snapshot fetch"
+
+            # Gated: the broker bootstrapped on its stale local cache, learned of
+            # the change only via post-bootstrap log replay, and -- since we no
+            # longer promote pending needs_restart values at start -- still
+            # reports restart-required. restart=False here would mean the
+            # bootstrap fetch ran while the upgrade was unfinalized.
+            assert restarted_node_status()["restart"] is True, (
+                "a gated broker must bootstrap on its stale local cache and "
+                "report restart-required for the missed change"
+            )
+        finally:
+            # Leave no config override behind for later phases or the downgrade.
+            try:
+                self.admin.patch_cluster_config(remove=[PROPERTY])
+            except Exception as e:
+                self.logger.warn(f"cleanup: failed to reset {PROPERTY}: {e}")
+
     def _verify_tiered_cloud_topics_working(self):
         """After finalize the active version has advanced past the feature's
         require_version, so the gate opens: the feature moves from unavailable to
@@ -1593,11 +1703,27 @@ def _verify_shadow_link_role_sync_working(self):
                 f"role-sync config still gated after finalize: {e}"
             )
 
+    def _verify_fetch_controller_snapshot_rpc_working(self):
+        """After finalize the active version advances past the feature's
+        require_version and it auto-activates (available_policy::always), opening
+        the bootstrap-fetch gate. Confirm the state transition; the active-state
+        fetch mechanics (a delayed-restart broker picking up fresh config before
+        starting services, the inverse of the gated outcome asserted in
+        _exercise_fetch_controller_snapshot_rpc) are covered end to end by
+        ClusterConfigMultiNodeBootstrapTest.test_node_delayed_restart."""
+        wait_until(
+            lambda: self._feature_state("fetch_controller_snapshot_rpc") == "active",
+            timeout_sec=30,
+            backoff_sec=1,
+            err_msg="fetch_controller_snapshot_rpc did not activate after finalize",
+        )
+
     def _verify_v26_2_features_working(self):
         """After the upgrade is finalized, confirm each v26.2 feature's gate has
         opened and the feature actually works (simple per-feature predicates)."""
         self._verify_tiered_cloud_topics_working()
         self._verify_shadow_link_role_sync_working()
+        self._verify_fetch_controller_snapshot_rpc_working()
 
     def _feature_state(self, name):
         for f in self.admin.get_features()["features"]: