cluster_link: gate shadow-link role sync behind a feature flag

src/v/features/feature_table.cc

@@ -49,6 +49,8 @@ std::string_view to_string_view(feature f) {
         return "batch_mirror_topic_status";
     case feature::shadow_link_sr_api_sync:
         return "shadow_link_sr_api_sync";
+    case feature::shadow_link_role_sync:
+        return "shadow_link_role_sync";
     case feature::iceberg_extended_mode_config:
         return "iceberg_extended_mode_config";
     case feature::coordinated_compaction:

src/v/features/feature_table.h

@@ -62,6 +62,7 @@ enum class feature : std::uint64_t {
     fetch_controller_snapshot_rpc = 1ULL << 18U,
     node_isolation = 1ULL << 19U,
     group_offset_retention = 1ULL << 20U,
+    shadow_link_role_sync = 1ULL << 21U,
     membership_change_controller_cmds = 1ULL << 22U,
     controller_snapshots = 1ULL << 23U,
     cloud_storage_manifest_format_v2 = 1ULL << 24U,
@@ -574,6 +575,12 @@ inline constexpr std::array feature_schema{
     feature::shadow_link_sr_api_sync,
     feature_spec::available_policy::always,
     feature_spec::prepare_policy::always},
+  feature_spec{
+    release_version::v26_2_1,
+    "shadow_link_role_sync",
+    feature::shadow_link_role_sync,
+    feature_spec::available_policy::always,
+    feature_spec::prepare_policy::always},
   feature_spec{
     release_version::v26_2_1,
     "iceberg_extended_mode_config",

src/v/features/tests/feature_table_test.cc

@@ -89,6 +89,12 @@ TEST_F(FeatureTableTest, feature_table_strings) {
     ASSERT_EQ(to_string_view(feature::audit_logging), "audit_logging");
 }
 
+TEST(feature_table_string_test, shadow_link_role_sync) {
+    EXPECT_EQ(
+      features::to_string_view(features::feature::shadow_link_role_sync),
+      std::string_view{"shadow_link_role_sync"});
+}
+
 /**
  * Check that the test feature shows up when environment variable
  * is set (via the fixture, in this case)

src/v/kafka/server/handlers/api_versions.cc

@@ -8,6 +8,7 @@
 // by the Apache License, Version 2.0
 
 #include "features/feature_table.h"
+#include "kafka/protocol/describe_redpanda_roles.h"
 #include "kafka/protocol/types.h"
 #include "kafka/server/handlers/handlers.h"
 #include "kafka/server/request_context.h"
@@ -110,6 +111,19 @@ void topic_id_api_version_limiter(api_versions_response& r) {
     }
 }
 
+void remove_unavailable_reserved_apis(
+  api_versions_response& r, const features::feature_table& ft) {
+    if (ft.is_active(features::feature::shadow_link_role_sync)) {
+        return;
+    }
+
+    auto to_remove = std::ranges::remove(
+      r.data.api_keys,
+      describe_redpanda_roles_api::key,
+      &api_versions_response_key::api_key);
+    r.data.api_keys.erase_to_end(to_remove.begin());
+}
+
 api_versions_response api_versions_handler::handle_raw(request_context& ctx) {
     // Unlike other request types, we handle ApiVersion requests
     // with higher versions than supported. We treat such a request
@@ -147,12 +161,16 @@ api_versions_response api_versions_handler::handle_raw(request_context& ctx) {
     }
 
     {
-        // If feature::topic_ids is not active, limit reported API versions
         const auto& features = ctx.feature_table().local();
+
+        // If feature::topic_ids is not active, limit reported API versions
         if (unlikely(!features.is_active(features::feature::topic_ids_api))) {
             topic_id_api_version_limiter(r);
         }
+
+        remove_unavailable_reserved_apis(r, features);
     }
+
     return r;
 }
 

src/v/kafka/server/handlers/api_versions.h

@@ -13,6 +13,10 @@
 #include "kafka/protocol/api_versions.h"
 #include "kafka/server/handlers/handler.h"
 
+namespace features {
+class feature_table;
+} // namespace features
+
 namespace kafka {
 
 struct api_versions_handler
@@ -33,4 +37,7 @@ struct api_versions_handler
 
 chunked_vector<api_versions_response_key> get_supported_apis();
 
+void remove_unavailable_reserved_apis(
+  api_versions_response& r, const features::feature_table& ft);
+
 } // namespace kafka

src/v/kafka/server/handlers/describe_redpanda_roles.cc

@@ -11,6 +11,7 @@
 #include "kafka/server/handlers/describe_redpanda_roles.h"
 
 #include "container/chunked_hash_map.h"
+#include "features/feature_table.h"
 #include "kafka/protocol/errors.h"
 #include "kafka/server/handlers/details/roles.h"
 #include "kafka/server/request_context.h"
@@ -32,11 +33,20 @@ ss::future<response_ptr> describe_redpanda_roles_handler::handle(
     request.decode(ctx.reader(), ctx.header().version);
     log_request(ctx.header(), request);
 
+    describe_redpanda_roles_response resp;
+
+    if (!ctx.feature_table().local().is_active(
+          features::feature::shadow_link_role_sync)) {
+        resp.data.error_code = error_code::unsupported_version;
+        resp.data.error_message
+          = "DescribeRedpandaRoles is not available until the cluster is "
+            "fully upgraded";
+        co_return co_await ctx.respond(std::move(resp));
+    }
+
     auto authz = ctx.authorized(
       security::acl_operation::describe, security::default_cluster_name);
 
-    describe_redpanda_roles_response resp;
-
     if (!ctx.audit()) {
         resp.data.error_code = error_code::broker_not_available;
         resp.data.error_message = "Broker not available - audit system failure";

src/v/kafka/server/tests/BUILD

@@ -450,7 +450,9 @@ redpanda_cc_btest(
         "//src/v/cluster:node_status_backend",
         "//src/v/cluster:self_test",
         "//src/v/cluster:topic_metrics_watcher",
+        "//src/v/features",
         "//src/v/kafka/protocol",
+        "//src/v/kafka/protocol:describe_redpanda_roles",
         "//src/v/kafka/server",
         "//src/v/redpanda/tests:fixture",
         "//src/v/test_utils:seastar_boost",

src/v/kafka/server/tests/api_versions_test.cc

@@ -7,6 +7,8 @@
 // the Business Source License, use of this software will be governed
 // by the Apache License, Version 2.0
 
+#include "features/feature_table.h"
+#include "kafka/protocol/describe_redpanda_roles.h"
 #include "kafka/protocol/types.h"
 #include "kafka/server/handlers/api_versions.h"
 #include "redpanda/tests/fixture.h"
@@ -96,3 +98,31 @@ FIXTURE_TEST(reserved_range_apis_advertised, redpanda_thread_fixture) {
         return a.api_key == kafka::redpanda_api_key_base;
     }));
 }
+
+SEASTAR_THREAD_TEST_CASE(reserved_api_gated_by_feature) {
+    auto make_resp = [] {
+        kafka::api_versions_response r;
+        r.data.api_keys.push_back(
+          kafka::api_versions_response_key{
+            kafka::describe_redpanda_roles_api::key,
+            kafka::api_version{0},
+            kafka::api_version{0}});
+        return r;
+    };
+    const auto has_key = [](const kafka::api_versions_response& r) {
+        return std::ranges::any_of(r.data.api_keys, [](const auto& a) {
+            return a.api_key == kafka::describe_redpanda_roles_api::key;
+        });
+    };
+
+    features::feature_table inactive;
+    auto r1 = make_resp();
+    kafka::remove_unavailable_reserved_apis(r1, inactive);
+    BOOST_CHECK(!has_key(r1));
+
+    features::feature_table active;
+    active.testing_activate_all();
+    auto r2 = make_resp();
+    kafka::remove_unavailable_reserved_apis(r2, active);
+    BOOST_CHECK(has_key(r2));
+}

src/v/redpanda/admin/services/shadow_link/shadow_link.cc

@@ -35,6 +35,19 @@ void check_sr_api_sync_supported(
           "cluster is fully upgraded");
     }
 }
+
+void check_role_sync_supported(
+  const cluster_link::model::role_sync_config& cfg,
+  const features::feature_table& feature_table) {
+    if (cfg.role_name_filters.empty()) {
+        return;
+    }
+    if (!feature_table.is_active(features::feature::shadow_link_role_sync)) {
+        throw serde::pb::rpc::failed_precondition_exception(
+          "Role sync cannot be configured until the cluster is fully "
+          "upgraded");
+    }
+}
 } // namespace
 
 shadow_link_service_impl::shadow_link_service_impl(
@@ -67,6 +80,8 @@ shadow_link_service_impl::create_shadow_link(
     auto md = convert_create_to_metadata(std::move(req));
     check_sr_api_sync_supported(
       md.configuration.schema_registry_sync_cfg, _feature_table->local());
+    check_role_sync_supported(
+      md.configuration.role_sync_cfg, _feature_table->local());
     auto get_resp = _service->local().get_cluster_link(md.name);
     if (get_resp.has_value()) {
         throw serde::pb::rpc::already_exists_exception(
@@ -196,6 +211,8 @@ shadow_link_service_impl::update_shadow_link(
       std::move(req), std::move(current_link));
     check_sr_api_sync_supported(
       update_cmd.link_config.schema_registry_sync_cfg, _feature_table->local());
+    check_role_sync_supported(
+      update_cmd.link_config.role_sync_cfg, _feature_table->local());
 
     auto updated_md = handle_error(
       co_await _service->local().update_cluster_link(

tests/rptest/tests/cluster_features_test.py

@@ -11,12 +11,16 @@
 import time
 
 from connectrpc.errors import ConnectError, ConnectErrorCode
+import google.protobuf.duration_pb2 as duration_pb2
 from ducktape.errors import TimeoutError as DucktapeTimeoutError
 from ducktape.mark import parametrize
 from ducktape.utils.util import wait_until
 from requests.exceptions import HTTPError
 
-from rptest.clients.admin.proto.redpanda.core.admin.v2 import features_pb2
+from rptest.clients.admin.proto.redpanda.core.admin.v2 import (
+    features_pb2,
+    shadow_link_pb2,
+)
 from rptest.clients.admin.v2 import Admin as AdminV2
 from rptest.clients.kafka_cli_tools import KafkaCliTools
 from rptest.clients.rpk import RpkException, RpkTool
@@ -1205,7 +1209,7 @@ 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"})
+PERTURB_EXERCISED_FEATURES = frozenset({"tiered_cloud_topics", "shadow_link_role_sync"})
 PERTURB_ACKNOWLEDGED_FEATURES = frozenset(
     {
         # Cluster-linking features, out of scope for this single-cluster test.
@@ -1245,6 +1249,12 @@ class ManualFinalizationUpgradeTest(FeaturesTestBase):
     status/finalize RPCs are only invoked once every node is on HEAD.
     """
 
+    # DescribeRedpandaRoles occupies the reserved Redpanda Kafka API key range
+    # (>= 15000); api_versions.cc strips it from ApiVersions until the
+    # shadow_link_role_sync feature is active.
+    DESCRIBE_REDPANDA_ROLES_API_KEY = 15000
+    _ROLE_SYNC_GATE_MESSAGE = "Role sync cannot be configured"
+
     def __init__(self, *args, **kwargs):
         super().__init__(*args, num_brokers=3, **kwargs)
         self.admin_v2 = AdminV2(self.redpanda)
@@ -1444,6 +1454,7 @@ def _perturb_v26_1_to_v26_2(self, phase):
         if "upgraded" not in phase:
             return
         self._exercise_tiered_cloud_topics()
+        self._exercise_shadow_link_role_sync()
         # 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
@@ -1481,6 +1492,62 @@ def _exercise_tiered_cloud_topics(self):
                 "creating a tiered_cloud topic should be gated while unfinalized"
             )
 
+    def _validate_role_sync_config(self):
+        """Issue a validate-only CreateShadowLink with role sync configured to
+        exercise the config gate (check_role_sync_supported in shadow_link.cc).
+        role_name_filters must be non-empty -- it's the field the gate keys on;
+        callers interpret the gated vs ungated result."""
+        client = self.admin_v2.shadow_link()
+        req = shadow_link_pb2.CreateShadowLinkRequest(validate_only=True)
+        req.shadow_link.name = "perturb-role-sync"
+        req.shadow_link.configurations.client_options.bootstrap_servers.extend(
+            self.redpanda.brokers().split(",")
+        )
+        req.shadow_link.configurations.role_sync_options.CopyFrom(
+            shadow_link_pb2.RoleSyncOptions(
+                interval=duration_pb2.Duration(seconds=1),
+                role_name_filters=[
+                    shadow_link_pb2.NameFilter(
+                        pattern_type=shadow_link_pb2.PATTERN_TYPE_PREFIX,
+                        filter_type=shadow_link_pb2.FILTER_TYPE_INCLUDE,
+                        name="synced-",
+                    )
+                ],
+            )
+        )
+        self._call_with_leader_retry(lambda: client.create_shadow_link(req=req))
+
+    def _exercise_shadow_link_role_sync(self):
+        """shadow_link_role_sync gates two surfaces while the upgrade is
+        unfinalized: the DescribeRedpandaRoles Kafka API is not advertised in
+        ApiVersions, and configuring role sync on a shadow link is refused.
+        Exercise both."""
+        assert self._feature_state("shadow_link_role_sync") == "unavailable", (
+            "shadow_link_role_sync should be unavailable while unfinalized"
+        )
+        # Wire gate: "(<key>)" appears only when the broker advertises the key,
+        # which the client renders as UNKNOWN(<key>); its absence means it's gated.
+        api_versions = KafkaCliTools(self.redpanda).get_api_versions()
+        assert f"({self.DESCRIBE_REDPANDA_ROLES_API_KEY})" not in api_versions, (
+            "DescribeRedpandaRoles should not be advertised while unfinalized:\n"
+            f"{api_versions}"
+        )
+        # Config gate: confirm it's the role-sync gate and not an unrelated
+        # precondition failure by checking both the error code and the message.
+        try:
+            self._validate_role_sync_config()
+        except ConnectError as e:
+            assert e.code == ConnectErrorCode.FAILED_PRECONDITION, (
+                f"role-sync config should be gated by a precondition, got {e}"
+            )
+            assert self._ROLE_SYNC_GATE_MESSAGE in str(e), (
+                f"role-sync config rejected, but not via the feature gate: {e}"
+            )
+        else:
+            raise AssertionError(
+                "configuring role sync should be gated while unfinalized"
+            )
+
     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
@@ -1502,10 +1569,35 @@ def _verify_tiered_cloud_topics_working(self):
             err_msg="tiered_cloud_topics did not activate after being enabled",
         )
 
+    def _verify_shadow_link_role_sync_working(self):
+        """After finalize both gates open: shadow_link_role_sync auto-activates
+        (available_policy::always, so no explicit enable), DescribeRedpandaRoles is
+        advertised, and the role-sync config gate no longer rejects. Any
+        non-gate outcome of the validate call is acceptable -- the connection test
+        may pass or fail, but it must not be the feature precondition."""
+        wait_until(
+            lambda: self._feature_state("shadow_link_role_sync") == "active",
+            timeout_sec=30,
+            backoff_sec=1,
+            err_msg="shadow_link_role_sync did not activate after finalize",
+        )
+        api_versions = KafkaCliTools(self.redpanda).get_api_versions()
+        assert f"({self.DESCRIBE_REDPANDA_ROLES_API_KEY})" in api_versions, (
+            "DescribeRedpandaRoles should be advertised after finalize:\n"
+            f"{api_versions}"
+        )
+        try:
+            self._validate_role_sync_config()
+        except ConnectError as e:
+            assert self._ROLE_SYNC_GATE_MESSAGE not in str(e), (
+                f"role-sync config still gated after finalize: {e}"
+            )
+
     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()
 
     def _feature_state(self, name):
         for f in self.admin.get_features()["features"]: