[CORE-16775] cluster_link: unsupported-feature policy for Schema Registry config sync

src/v/cluster_link/schema_registry_sync/http_source_reader.cc

@@ -325,8 +325,7 @@ http_source_reader::read_mode(ppsr::context_subject sub, ss::abort_source& as) {
     co_return narrowed;
 }
 
-ss::future<source_result<std::optional<ppsr::compatibility_level>>>
-http_source_reader::read_config(
+ss::future<source_result<source_config_read>> http_source_reader::read_config(
   ppsr::context_subject sub, ss::abort_source& as) {
     auto units = co_await ss::get_units(_inflight, 1, as);
     auto client = co_await ensure_client(as);
@@ -342,26 +341,32 @@ http_source_reader::read_config(
                      sub, rtc, ppsr::default_to_global::no);
     if (!res.has_value()) {
         if (std::holds_alternative<rc::subject_config_not_found>(res.error())) {
-            co_return std::optional<ppsr::compatibility_level>{std::nullopt};
+            co_return source_config_read{.compatibility = std::nullopt};
         }
         co_return std::unexpected(to_source_error(std::move(res.error())));
     }
-    auto narrowed = narrow_compat(res.value().level);
-    if (!narrowed.has_value()) {
-        co_return std::unexpected(
-          source_error{
-            .kind = source_error_kind::operation_failed,
-            .message = fmt::format(
-              "source compatibility level '{}' of {} is not supported by the "
-              "destination",
-              res.value().raw,
-              sub)});
+    // A governance-only config (no compatibility level) is "no override"; its
+    // unsupported fields still reach the policy. The Config API documents the
+    // subject-level compatibility as optional ("the compatibility, if any"):
+    // https://docs.confluent.io/platform/current/schema-registry/develop/api.html#config
+    std::optional<ppsr::compatibility_level> narrowed;
+    if (res.value().level.has_value()) {
+        narrowed = narrow_compat(*res.value().level);
+        if (!narrowed.has_value()) {
+            co_return std::unexpected(
+              source_error{
+                .kind = source_error_kind::operation_failed,
+                .message = fmt::format(
+                  "source compatibility level '{}' of {} is not supported by "
+                  "the destination",
+                  res.value().raw,
+                  sub)});
+        }
     }
-    // Unsupported config fields (config_info::unknown_fields, e.g.
-    // defaultRuleSet or compatibilityGroup) are ignored; honoring
-    // unsupported_schema_feature_policy is future work, as it is for the
-    // schema-body path in read_subject_version.
-    co_return narrowed;
+    // Carry the unsupported config fields through for the sync's policy.
+    co_return source_config_read{
+      .compatibility = narrowed,
+      .unsupported = std::move(res.value().unsupported)};
 }
 
 ss::future<> http_source_reader::stop() {

src/v/cluster_link/schema_registry_sync/http_source_reader.h

@@ -83,7 +83,7 @@ class http_source_reader final : public source_reader {
     ss::future<source_result<std::optional<ppsr::mode>>>
     read_mode(ppsr::context_subject, ss::abort_source&) override;
 
-    ss::future<source_result<std::optional<ppsr::compatibility_level>>>
+    ss::future<source_result<source_config_read>>
     read_config(ppsr::context_subject, ss::abort_source&) override;
 
     ss::future<> stop() override;

src/v/cluster_link/schema_registry_sync/mirroring_task.cc

@@ -28,6 +28,8 @@
 #include <seastar/coroutine/as_future.hh>
 #include <seastar/util/defer.hh>
 
+#include <fmt/ranges.h>
+
 #include <utility>
 
 namespace cluster_link::schema_registry_sync {
@@ -337,8 +339,54 @@ ss::future<uint64_t> mirroring_task::purge_destination_only_versions(
     co_return purged;
 }
 
+bool mirroring_task::fail_if_contains_unsupported(
+  const ppsr::context_subject& target,
+  model::schema_registry_sync_config::unsupported_feature_policy feature_policy,
+  const chunked_vector<ppsr::unsupported_feature>& unsupported) {
+    using policy_t
+      = model::schema_registry_sync_config::unsupported_feature_policy;
+    if (feature_policy != policy_t::fail || unsupported.empty()) {
+        return false;
+    }
+    // Per-item failure (as on the schema-body path): count it, log the
+    // offending fields, and let the caller skip this subject's config sync
+    // while the rest of the work continues.
+    record_error(
+      fmt::format(
+        "{} config carries {} unsupported feature(s) the destination cannot "
+        "store; failing it under the FAIL policy: {}",
+        target,
+        unsupported.size(),
+        fmt::join(unsupported, ", ")));
+    return true;
+}
+
+void mirroring_task::count_if_contains_unsupported_removed(
+  const ppsr::context_subject& target,
+  model::schema_registry_sync_config::unsupported_feature_policy feature_policy,
+  const chunked_vector<ppsr::unsupported_feature>& unsupported) {
+    using policy_t
+      = model::schema_registry_sync_config::unsupported_feature_policy;
+    if (feature_policy != policy_t::remove || unsupported.empty()) {
+        return;
+    }
+    // Only compatibilityLevel is synced, so the unsupported config fields are
+    // already dropped; log and count them.
+    vlog(
+      logger().info,
+      "Removed {} unsupported config feature(s) from {}: {}",
+      unsupported.size(),
+      target,
+      fmt::join(unsupported, ", "));
+    _status.current_sync->summary.unsupported_features_removed
+      += unsupported.size();
+    _status.totals_since_task_start.unsupported_features_removed
+      += unsupported.size();
+}
+
 ss::future<> mirroring_task::sync_mode_and_config(
   const ppsr::context_subject& target,
+  model::schema_registry_sync_config::unsupported_feature_policy feature_policy,
   ss::abort_source& as,
   std::optional<source_error>& unavailable) {
     // A peer fiber already hit source_unavailable; skip the remaining work.
@@ -402,9 +450,16 @@ ss::future<> mirroring_task::sync_mode_and_config(
         co_return;
     }
 
+    auto& cfg = config.value();
+    // FAIL rejects before the write; REMOVE accounting runs after it lands.
+    // Policy diagnostics name the source subject, write errors the destination.
+    if (fail_if_contains_unsupported(target, feature_policy, cfg.unsupported)) {
+        co_return;
+    }
+
     auto write = co_await ss::coroutine::as_future(
-      config.value().has_value()
-        ? _destination->write_config(dest_target, *config.value())
+      cfg.compatibility.has_value()
+        ? _destination->write_config(dest_target, *cfg.compatibility)
         : _destination->delete_config(dest_target));
     if (write.failed()) {
         auto ex = write.get_exception();
@@ -419,6 +474,12 @@ ss::future<> mirroring_task::sync_mode_and_config(
         ++_status.current_sync->summary.compatibility_configs_changed;
         ++_status.totals_since_task_start.compatibility_configs_changed;
     }
+    // Counted per completed sync, even when the write was a no-op: the drop
+    // recurs on every re-read, and a governance-only config (a no-op delete)
+    // must not be silently ignored. Mirrors FAIL's per-sync errors; the schema
+    // path counts once because its projection lands durably.
+    count_if_contains_unsupported_removed(
+      target, feature_policy, cfg.unsupported);
 }
 
 void mirroring_task::record_error(std::string_view what) {
@@ -704,7 +765,8 @@ ss::future<task::state_transition> mirroring_task::full_source_sync(
       mode_config_targets,
       std::max<size_t>(1, limits.parallelism),
       [&](const ppsr::context_subject& target) {
-          return sync_mode_and_config(target, as, mc_unavailable);
+          return sync_mode_and_config(
+            target, feature_policy, as, mc_unavailable);
       });
     if (mc_unavailable.has_value()) {
         co_return make_unavailable(mc_unavailable->message);

src/v/cluster_link/schema_registry_sync/mirroring_task.h

@@ -192,9 +192,31 @@ class mirroring_task : public task {
     /// override when it has one, deletes the destination override otherwise.
     ss::future<> sync_mode_and_config(
       const ppsr::context_subject& target,
+      model::schema_registry_sync_config::unsupported_feature_policy
+        feature_policy,
       ss::abort_source& as,
       std::optional<source_error>& unavailable);
 
+    /// Under FAIL, records unsupported config fields as a per-item error and
+    /// returns true so the caller skips the config write; a no-op (false)
+    /// otherwise. Config-path analogue of the reconciler's helper.
+    bool fail_if_contains_unsupported(
+      const ppsr::context_subject& target,
+      model::schema_registry_sync_config::unsupported_feature_policy
+        feature_policy,
+      const chunked_vector<ppsr::unsupported_feature>& unsupported);
+
+    /// Under REMOVE, logs the unsupported config fields and counts them in
+    /// `unsupported_features_removed`; a no-op otherwise. Called once per
+    /// completed config sync -- a static source config re-counts every full
+    /// sync, mirroring FAIL's per-sync errors -- but not after a failed write,
+    /// which is counted as an error instead.
+    void count_if_contains_unsupported_removed(
+      const ppsr::context_subject& target,
+      model::schema_registry_sync_config::unsupported_feature_policy
+        feature_policy,
+      const chunked_vector<ppsr::unsupported_feature>& unsupported);
+
     // Requires a sync in progress (`current_sync` engaged).
     void record_error(std::string_view what);
 

src/v/cluster_link/schema_registry_sync/source_reader.h

@@ -48,6 +48,14 @@ struct source_error {
 template<typename T>
 using source_result = std::expected<T, source_error>;
 
+/// A config read from the source: the compatibility-level override (nullopt
+/// when the source has none) plus any unsupported config fields, for the
+/// caller's unsupported-feature policy.
+struct source_config_read {
+    std::optional<ppsr::compatibility_level> compatibility;
+    chunked_vector<ppsr::unsupported_feature> unsupported;
+};
+
 /// \brief Abstraction over a source Schema Registry, scoped to one link.
 ///
 /// Reads are split into discovery (list subjects/versions) and fetch (read a
@@ -90,8 +98,9 @@ class source_reader {
     virtual ss::future<source_result<std::optional<ppsr::mode>>>
     read_mode(ppsr::context_subject, ss::abort_source&) = 0;
 
-    /// As read_mode, for the compatibility-level override.
-    virtual ss::future<source_result<std::optional<ppsr::compatibility_level>>>
+    /// As read_mode, for the compatibility-level override plus any unsupported
+    /// config fields.
+    virtual ss::future<source_result<source_config_read>>
     read_config(ppsr::context_subject, ss::abort_source&) = 0;
 
     /// Releases any resources the reader holds (e.g. an HTTP transport). Called

src/v/cluster_link/schema_registry_sync/tests/http_source_reader_test.cc

@@ -434,8 +434,10 @@ TEST(http_source_reader, read_config_narrows_and_classifies) {
                      .get();
         reader.stop().get();
         ASSERT_TRUE(res.has_value());
-        ASSERT_TRUE(res->has_value());
-        EXPECT_EQ(**res, pps::compatibility_level::full_transitive);
+        ASSERT_TRUE(res->compatibility.has_value());
+        EXPECT_EQ(
+          *res->compatibility, pps::compatibility_level::full_transitive);
+        EXPECT_TRUE(res->unsupported.empty());
     }
     {
         auto reader = reader_over([](mock_client& m) {
@@ -449,7 +451,7 @@ TEST(http_source_reader, read_config_narrows_and_classifies) {
                      .get();
         reader.stop().get();
         ASSERT_TRUE(res.has_value());
-        EXPECT_FALSE(res->has_value());
+        EXPECT_FALSE(res->compatibility.has_value());
     }
     {
         auto reader = reader_over([](mock_client& m) {
@@ -465,6 +467,26 @@ TEST(http_source_reader, read_config_narrows_and_classifies) {
         ASSERT_FALSE(res.has_value());
         EXPECT_EQ(res.error().kind, srs::source_error_kind::operation_failed);
     }
+    {
+        // Governance-only subject config: no compatibility level, only an
+        // unsupported field. The read succeeds ("no override") and carries the
+        // field for the policy instead of failing the parse.
+        auto reader = reader_over([](mock_client& m) {
+            EXPECT_CALL(m, request_and_collect_response(_, _, _))
+              .WillOnce(respond(
+                bh::status::ok,
+                R"({"compatibilityGroup":"app.major.version"})"));
+        });
+        ss::abort_source as;
+        auto res = reader
+                     .read_config(pps::context_subject::unqualified("s1"), as)
+                     .get();
+        reader.stop().get();
+        ASSERT_TRUE(res.has_value());
+        EXPECT_FALSE(res->compatibility.has_value());
+        ASSERT_EQ(res->unsupported.size(), size_t{1});
+        EXPECT_EQ(res->unsupported[0].json_pointer, "/compatibilityGroup");
+    }
 }
 
 // A null config (not in API mode) or an unparseable URL yields an unavailable