diff --git a/src/v/cluster_link/schema_registry_sync/http_source_reader.cc b/src/v/cluster_link/schema_registry_sync/http_source_reader.cc
index 631e6be04df5d..df9e0104f6b6d 100644
--- a/src/v/cluster_link/schema_registry_sync/http_source_reader.cc
+++ b/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() {
diff --git a/src/v/cluster_link/schema_registry_sync/http_source_reader.h b/src/v/cluster_link/schema_registry_sync/http_source_reader.h
index 358a159267dbe..7c490ab98171f 100644
--- a/src/v/cluster_link/schema_registry_sync/http_source_reader.h
+++ b/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;
diff --git a/src/v/cluster_link/schema_registry_sync/mirroring_task.cc b/src/v/cluster_link/schema_registry_sync/mirroring_task.cc
index a26ae5195639e..202ca73c679be 100644
--- a/src/v/cluster_link/schema_registry_sync/mirroring_task.cc
+++ b/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);
diff --git a/src/v/cluster_link/schema_registry_sync/mirroring_task.h b/src/v/cluster_link/schema_registry_sync/mirroring_task.h
index 7053493809f38..0fda87dfa7a09 100644
--- a/src/v/cluster_link/schema_registry_sync/mirroring_task.h
+++ b/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);
 
diff --git a/src/v/cluster_link/schema_registry_sync/source_reader.h b/src/v/cluster_link/schema_registry_sync/source_reader.h
index 77e114acb44a2..ee9a40b086945 100644
--- a/src/v/cluster_link/schema_registry_sync/source_reader.h
+++ b/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
diff --git a/src/v/cluster_link/schema_registry_sync/tests/http_source_reader_test.cc b/src/v/cluster_link/schema_registry_sync/tests/http_source_reader_test.cc
index 4a5a1859d0d68..b1ab1e1322ecc 100644
--- a/src/v/cluster_link/schema_registry_sync/tests/http_source_reader_test.cc
+++ b/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
diff --git a/src/v/cluster_link/schema_registry_sync/tests/mirroring_task_test.cc b/src/v/cluster_link/schema_registry_sync/tests/mirroring_task_test.cc
index 690f8b77e09b9..090ead2068bf9 100644
--- a/src/v/cluster_link/schema_registry_sync/tests/mirroring_task_test.cc
+++ b/src/v/cluster_link/schema_registry_sync/tests/mirroring_task_test.cc
@@ -308,6 +308,154 @@ TEST_F(mirroring_task_test, fail_policy_counts_unsupported_and_syncs_rest) {
     EXPECT_GE(index_of(_registry.get_all(), "b"), 0);
 }
 
+TEST_F(mirroring_task_test, remove_policy_counts_unsupported_config) {
+    // The config path applies the same policy as the schema path. Under REMOVE,
+    // an unsupported config field (e.g. defaultRuleSet) is counted and logged;
+    // only the supported projection (compatibilityLevel) is synced.
+    auto a = ppsr::context_subject::unqualified("a");
+    _source_state.add(a, 1);
+    _source_state.configs.emplace(a, ppsr::compatibility_level::full);
+    _source_state.set_config_unsupported(
+      a, {{.json_pointer = "/defaultRuleSet"}});
+
+    auto metadata = get_default_metadata();
+    metadata.configuration.schema_registry_sync_cfg.api_mode()->feature_policy
+      = model::schema_registry_sync_config::unsupported_feature_policy::remove;
+
+    lead_schema_registry();
+    fixture()->upsert_link(std::move(metadata)).get();
+
+    auto status = wait_for_sync_status([](const auto& s) {
+                      return s.last_full_sync.has_value()
+                             && !s.current_sync.has_value();
+                  }).get();
+    ASSERT_TRUE(status.has_value());
+    EXPECT_EQ(status->last_full_sync->unsupported_features_removed, 1);
+    EXPECT_EQ(status->totals_since_task_start.unsupported_features_removed, 1);
+    // The compatibility level is still synced under REMOVE.
+    EXPECT_EQ(status->last_full_sync->compatibility_configs_changed, 1);
+    ASSERT_TRUE(_registry.configs().contains(a));
+    EXPECT_EQ(_registry.configs().at(a), ppsr::compatibility_level::full);
+    EXPECT_EQ(status->last_full_sync->errors, 0);
+
+    // A second full sync re-reads and re-drops the same fields: the count is
+    // per completed sync (mirroring FAIL's per-sync errors), so the total
+    // advances even though the config write itself is a no-op.
+    const auto first_start = status->last_full_sync->start_time;
+    auto metadata2 = get_default_metadata();
+    metadata2.configuration.schema_registry_sync_cfg.api_mode()->feature_policy
+      = model::schema_registry_sync_config::unsupported_feature_policy::remove;
+    fixture()->upsert_link(std::move(metadata2)).get();
+    auto second = wait_for_sync_status([&](const auto& s) {
+                      return s.last_full_sync.has_value()
+                             && s.last_full_sync->start_time != first_start
+                             && !s.current_sync.has_value();
+                  }).get();
+    ASSERT_TRUE(second.has_value());
+    EXPECT_EQ(second->totals_since_task_start.unsupported_features_removed, 2);
+    EXPECT_EQ(second->last_full_sync->unsupported_features_removed, 1);
+    EXPECT_EQ(second->totals_since_task_start.errors, 0);
+}
+
+TEST_F(mirroring_task_test, remove_policy_counts_unsupported_config_no_level) {
+    // A governance-only source config (unsupported fields, no compatibility
+    // override): the destination write is a no-op delete, but the dropped
+    // fields are still counted -- REMOVE must not ignore them silently.
+    auto a = ppsr::context_subject::unqualified("a");
+    _source_state.add(a, 1);
+    _source_state.set_config_unsupported(
+      a, {{.json_pointer = "/compatibilityGroup"}});
+
+    auto metadata = get_default_metadata();
+    metadata.configuration.schema_registry_sync_cfg.api_mode()->feature_policy
+      = model::schema_registry_sync_config::unsupported_feature_policy::remove;
+
+    lead_schema_registry();
+    fixture()->upsert_link(std::move(metadata)).get();
+
+    auto status = wait_for_sync_status([](const auto& s) {
+                      return s.last_full_sync.has_value()
+                             && !s.current_sync.has_value();
+                  }).get();
+    ASSERT_TRUE(status.has_value());
+    EXPECT_EQ(status->last_full_sync->unsupported_features_removed, 1);
+    EXPECT_EQ(status->last_full_sync->errors, 0);
+    // No compatibility override lands on the destination.
+    EXPECT_FALSE(_registry.configs().contains(a));
+}
+
+TEST_F(mirroring_task_test, fail_policy_skips_unsupported_config) {
+    // Under FAIL, an unsupported config field is a counted per-item error and
+    // the subject's config is not synced; the rest of the sync proceeds and the
+    // task stays active (the clean schema still imports).
+    auto a = ppsr::context_subject::unqualified("a");
+    _source_state.add(a, 1);
+    _source_state.configs.emplace(a, ppsr::compatibility_level::full);
+    _source_state.set_config_unsupported(
+      a, {{.json_pointer = "/defaultRuleSet"}});
+
+    auto metadata = get_default_metadata();
+    metadata.configuration.schema_registry_sync_cfg.api_mode()->feature_policy
+      = model::schema_registry_sync_config::unsupported_feature_policy::fail;
+
+    lead_schema_registry();
+    fixture()->upsert_link(std::move(metadata)).get();
+
+    auto status = wait_for_sync_status([](const auto& s) {
+                      return s.last_full_sync.has_value()
+                             && !s.current_sync.has_value();
+                  }).get();
+    ASSERT_TRUE(status.has_value());
+    EXPECT_EQ(status->last_full_sync->errors, 1);
+    EXPECT_EQ(status->last_full_sync->unsupported_features_removed, 0);
+    // The config write is skipped: no compatibility change, no destination
+    // config.
+    EXPECT_EQ(status->last_full_sync->compatibility_configs_changed, 0);
+    EXPECT_FALSE(_registry.configs().contains(a));
+    // The clean schema itself still imports.
+    EXPECT_GE(index_of(_registry.get_all(), "a"), 0);
+}
+
+// Fixture whose destination wraps `_registry` so a test can make write_config
+// fail, exercising the REMOVE no-count-on-failed-write path the plain fake
+// cannot.
+class mirroring_task_config_write_failure_test : public mirroring_task_test {
+protected:
+    schema::registry* dest() override { return &_failing_config; }
+    failing_config_registry _failing_config{&_registry};
+};
+
+TEST_F(
+  mirroring_task_config_write_failure_test,
+  remove_policy_does_not_count_unsupported_config_on_write_failure) {
+    // REMOVE counts a removed config feature only after the config write lands;
+    // a write that fails is a per-item error and must not report the feature as
+    // removed (mirrors the schema-body path's no-count-on-failed-import).
+    auto a = ppsr::context_subject::unqualified("a");
+    _source_state.add(a, 1);
+    _source_state.configs.emplace(a, ppsr::compatibility_level::full);
+    _source_state.set_config_unsupported(
+      a, {{.json_pointer = "/defaultRuleSet"}});
+    _failing_config.fail_config(
+      a, ppsr::error_code::schema_invalid, "config write rejected");
+
+    auto metadata = get_default_metadata();
+    metadata.configuration.schema_registry_sync_cfg.api_mode()->feature_policy
+      = model::schema_registry_sync_config::unsupported_feature_policy::remove;
+
+    lead_schema_registry();
+    fixture()->upsert_link(std::move(metadata)).get();
+
+    auto status = wait_for_sync_status([](const auto& s) {
+                      return s.last_full_sync.has_value()
+                             && !s.current_sync.has_value();
+                  }).get();
+    ASSERT_TRUE(status.has_value());
+    // The failed write is counted as an error, not as a removed feature.
+    EXPECT_EQ(status->last_full_sync->errors, 1);
+    EXPECT_EQ(status->last_full_sync->unsupported_features_removed, 0);
+}
+
 TEST_F(mirroring_task_test, source_unavailable_then_recovers) {
     _source_state.add(ppsr::context_subject::unqualified("orders-value"), 1);
     _source_state.list_subjects_error = srs::source_error{
diff --git a/src/v/cluster_link/schema_registry_sync/tests/sr_sync_test_fixtures.h b/src/v/cluster_link/schema_registry_sync/tests/sr_sync_test_fixtures.h
index 047df6d3971d9..02ea47952246a 100644
--- a/src/v/cluster_link/schema_registry_sync/tests/sr_sync_test_fixtures.h
+++ b/src/v/cluster_link/schema_registry_sync/tests/sr_sync_test_fixtures.h
@@ -132,6 +132,19 @@ struct fake_source_state {
     chunked_hash_map<ppsr::context_subject, srs::source_error>
       read_config_errors;
 
+    // Unsupported config fields to attach to a target's read_config, letting a
+    // test drive the config-path unsupported-feature policy.
+    chunked_hash_map<
+      ppsr::context_subject,
+      chunked_vector<ppsr::unsupported_feature>>
+      config_unsupported;
+
+    void set_config_unsupported(
+      const ppsr::context_subject& sub,
+      chunked_vector<ppsr::unsupported_feature> features) {
+        config_unsupported[sub] = std::move(features);
+    }
+
     void fail_read(
       const ppsr::context_subject& sub,
       int32_t version,
@@ -303,17 +316,23 @@ class fake_source_reader final : public srs::source_reader {
         co_return std::optional<ppsr::mode>{std::nullopt};
     }
 
-    ss::future<srs::source_result<std::optional<ppsr::compatibility_level>>>
+    ss::future<srs::source_result<srs::source_config_read>>
     read_config(ppsr::context_subject sub, ss::abort_source&) override {
         if (
           auto it = _state->read_config_errors.find(sub);
           it != _state->read_config_errors.end()) {
             co_return std::unexpected(it->second);
         }
+        srs::source_config_read out;
         if (auto it = _state->configs.find(sub); it != _state->configs.end()) {
-            co_return std::optional<ppsr::compatibility_level>{it->second};
+            out.compatibility = it->second;
         }
-        co_return std::optional<ppsr::compatibility_level>{std::nullopt};
+        if (
+          auto it = _state->config_unsupported.find(sub);
+          it != _state->config_unsupported.end()) {
+            out.unsupported = it->second.copy();
+        }
+        co_return out;
     }
 
 private:
@@ -540,6 +559,33 @@ class failing_import_registry final : public delegating_registry {
     chunked_hash_map<ppsr::subject_version, ppsr::error_info> _failures;
 };
 
+// Wraps a destination registry so a test can make write_config fail for a
+// subject, exercising the REMOVE "no count on a failed config write" path the
+// plain fake (which never rejects a config write) cannot.
+class failing_config_registry final : public delegating_registry {
+public:
+    using delegating_registry::delegating_registry;
+
+    void fail_config(
+      const ppsr::context_subject& sub,
+      ppsr::error_code code,
+      ss::sstring message) {
+        _failures.emplace(sub, ppsr::error_info{code, std::move(message)});
+    }
+
+    ss::future<bool> write_config(
+      ppsr::context_subject sub, ppsr::compatibility_level c) override {
+        if (auto it = _failures.find(sub); it != _failures.end()) {
+            return ss::make_exception_future<bool>(
+              ppsr::as_exception(it->second));
+        }
+        return _inner->write_config(std::move(sub), c);
+    }
+
+private:
+    chunked_hash_map<ppsr::context_subject, ppsr::error_info> _failures;
+};
+
 // Wraps a destination registry, modeling the real Schema Registry's refusal to
 // permanently delete a version still referenced by a live one (the in-memory
 // fake never rejects a delete). Lets a test drive the hard-delete retry loop: a
diff --git a/src/v/cluster_link/schema_registry_sync/unavailable_source_reader.cc b/src/v/cluster_link/schema_registry_sync/unavailable_source_reader.cc
index b5292a532bcc8..cff63d1fff6c7 100644
--- a/src/v/cluster_link/schema_registry_sync/unavailable_source_reader.cc
+++ b/src/v/cluster_link/schema_registry_sync/unavailable_source_reader.cc
@@ -50,7 +50,7 @@ unavailable_source_reader::read_mode(ppsr::context_subject, ss::abort_source&) {
     co_return std::unexpected(unavailable());
 }
 
-ss::future<source_result<std::optional<ppsr::compatibility_level>>>
+ss::future<source_result<source_config_read>>
 unavailable_source_reader::read_config(
   ppsr::context_subject, ss::abort_source&) {
     co_return std::unexpected(unavailable());
diff --git a/src/v/cluster_link/schema_registry_sync/unavailable_source_reader.h b/src/v/cluster_link/schema_registry_sync/unavailable_source_reader.h
index ad747f04ee775..1650f7093beae 100644
--- a/src/v/cluster_link/schema_registry_sync/unavailable_source_reader.h
+++ b/src/v/cluster_link/schema_registry_sync/unavailable_source_reader.h
@@ -49,7 +49,7 @@ class unavailable_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;
 
 private:
diff --git a/src/v/pandaproxy/schema_registry/rest_client/client.h b/src/v/pandaproxy/schema_registry/rest_client/client.h
index 2f430661e47df..96e012eed657f 100644
--- a/src/v/pandaproxy/schema_registry/rest_client/client.h
+++ b/src/v/pandaproxy/schema_registry/rest_client/client.h
@@ -149,7 +149,7 @@ class client {
     /// registry_compatibility_level::unknown with the verbatim string in
     /// config_info::raw. Any other top-level config fields present (validation
     /// flags, metadata, rule sets — which Redpanda's own server does not emit)
-    /// are named in config_info::unknown_fields rather than modeled. The
+    /// are surfaced in config_info::unsupported rather than modeled. The
     /// `defaultToGlobal` query parameter is omitted: on the subject-less global
     /// endpoint it has no observable effect.
     ss::future<std::expected<config_info, domain_error>>
diff --git a/src/v/pandaproxy/schema_registry/rest_client/config.h b/src/v/pandaproxy/schema_registry/rest_client/config.h
index a853a3bdc73e7..1789611a03294 100644
--- a/src/v/pandaproxy/schema_registry/rest_client/config.h
+++ b/src/v/pandaproxy/schema_registry/rest_client/config.h
@@ -13,10 +13,12 @@
 #include "base/format_to.h"
 #include "base/seastarx.h"
 #include "container/chunked_vector.h"
+#include "pandaproxy/schema_registry/types.h"
 #include "strings/string_switch.h"
 
 #include <seastar/core/sstring.hh>
 
+#include <optional>
 #include <string_view>
 
 namespace pandaproxy::schema_registry::rest_client {
@@ -99,22 +101,17 @@ registry_compatibility_level_from_wire(std::string_view sv) {
 
 /// The result of `GET /config`: the registry's global configuration.
 ///
-/// Only `compatibilityLevel` is modeled — as an open enum in \ref level, with
-/// the verbatim wire string kept in \ref raw so a value mapped to unknown is
-/// not lost. The endpoint may carry many other optional fields (validation
-/// flags, metadata, rule sets) that this client does not model; the names of
-/// any such top-level fields present are recorded in \ref unknown_fields, so a
-/// caller can tell config content was dropped without this client having to
-/// model it. This serves the same intent as source_schema_read::unsupported on
-/// the schema-fetch path, but is a simpler representation: bare top-level field
-/// names only, not the structured JSON pointer + type that path records.
-/// Redpanda's own server emits only `compatibilityLevel`, so unknown_fields is
-/// empty against it; a third-party Confluent-compatible registry may populate
-/// it.
+/// Only `compatibilityLevel` is modeled, as an open enum in \ref level with
+/// the verbatim wire string kept in \ref raw. `level` is optional: a subject
+/// config may carry only governance fields (e.g. `compatibilityGroup`). Every
+/// other non-null top-level field is recorded in \ref unsupported as a JSON
+/// pointer + type (mirroring source_schema_read::unsupported) for the
+/// caller's unsupported-feature policy; a null value is treated as absent.
+/// Redpanda's own server emits only `compatibilityLevel`.
 struct config_info {
-    registry_compatibility_level level;
+    std::optional<registry_compatibility_level> level;
     ss::sstring raw;
-    chunked_vector<ss::sstring> unknown_fields;
+    chunked_vector<unsupported_feature> unsupported;
 };
 
 } // namespace pandaproxy::schema_registry::rest_client
diff --git a/src/v/pandaproxy/schema_registry/rest_client/parse.cc b/src/v/pandaproxy/schema_registry/rest_client/parse.cc
index a62cbf4edceaf..c6471c9d59d4b 100644
--- a/src/v/pandaproxy/schema_registry/rest_client/parse.cc
+++ b/src/v/pandaproxy/schema_registry/rest_client/parse.cc
@@ -280,7 +280,7 @@ ss::future<std::expected<config_info, parse_error>> parse_config(iobuf body) {
 
         std::optional<registry_compatibility_level> level;
         ss::sstring raw;
-        chunked_vector<ss::sstring> unknown_fields;
+        chunked_vector<unsupported_feature> unsupported;
         while (co_await p.next()) {
             if (p.token() == token::end_object) {
                 // The body is exactly one JSON object: reject any trailing
@@ -291,17 +291,13 @@ ss::future<std::expected<config_info, parse_error>> parse_config(iobuf body) {
                       parse_error{
                         .reason = "trailing content after config object"});
                 }
-                if (!level.has_value()) {
-                    // compatibilityLevel is the one field a config response is
-                    // documented to always carry.
-                    co_return std::unexpected(
-                      parse_error{
-                        .reason = "missing compatibilityLevel field"});
-                }
+                // An absent compatibilityLevel is not an error: a subject
+                // config may carry only governance fields, which must still
+                // reach the unsupported-feature policy.
                 co_return config_info{
-                  .level = *level,
+                  .level = level,
                   .raw = std::move(raw),
-                  .unknown_fields = std::move(unknown_fields)};
+                  .unsupported = std::move(unsupported)};
             }
             if (p.token() != token::key) {
                 co_return std::unexpected(
@@ -313,6 +309,14 @@ ss::future<std::expected<config_info, parse_error>> parse_config(iobuf body) {
                   parse_error{.reason = "truncated JSON after key"});
             }
             if (key == "compatibilityLevel") {
+                if (p.token() == token::value_null) {
+                    // Null means absent, as for every other field; duplicate
+                    // keys are last-wins, so clear any earlier value.
+                    level.reset();
+                    raw = {};
+                    co_await p.skip_value();
+                    continue;
+                }
                 if (p.token() != token::value_string) {
                     co_return std::unexpected(
                       parse_error{
@@ -324,9 +328,14 @@ ss::future<std::expected<config_info, parse_error>> parse_config(iobuf body) {
                 raw = p.value_string().linearize_to_string();
                 level = registry_compatibility_level_from_wire(raw);
             } else {
-                // Any other top-level field is unmodeled: record its name so a
-                // caller can tell config was dropped, then skip its value.
-                unknown_fields.push_back(std::move(key));
+                // Any other non-null top-level field is unmodeled: recorded as
+                // a JSON pointer + type for the policy; null means absent.
+                if (p.token() != token::value_null) {
+                    unsupported.push_back(
+                      unsupported_feature{
+                        .json_pointer = ssx::sformat("/{}", key),
+                        .json_type = json_type_name(p.token())});
+                }
                 co_await p.skip_value();
             }
         }
diff --git a/src/v/pandaproxy/schema_registry/rest_client/parse.h b/src/v/pandaproxy/schema_registry/rest_client/parse.h
index 184e429ab6047..7b0e9e693110e 100644
--- a/src/v/pandaproxy/schema_registry/rest_client/parse.h
+++ b/src/v/pandaproxy/schema_registry/rest_client/parse.h
@@ -83,17 +83,12 @@ ss::future<std::expected<mode_info, parse_error>> parse_mode(iobuf body);
 
 /// Parse the body of a `GET /config` response into a config_info.
 ///
-/// The body is a JSON object. Only `compatibilityLevel` (a string) is modeled,
-/// as an open enum (see config.h) with the verbatim wire string kept in
-/// config_info::raw. Every other top-level field is unmodeled: its name is
-/// recorded in config_info::unknown_fields and its value skipped, so a caller
-/// can tell config content was dropped without this client modeling the rich
-/// object. As with parse_mode the shape is strict — a non-object body, a
-/// missing or non-string `compatibilityLevel`, or trailing content after the
-/// object yields a parse_error — while the compatibilityLevel value is open: an
-/// unrecognized string maps to registry_compatibility_level::unknown rather
-/// than being rejected. The function does not throw: malformed input is
-/// reported via the returned std::expected.
+/// The body is a JSON object. A missing or null `compatibilityLevel` parses
+/// as absent; an unrecognized string maps to the open enum's `unknown`. Every
+/// other non-null top-level field is recorded in config_info::unsupported and
+/// its value skipped. Otherwise strict: a non-object body, a non-string
+/// non-null level, or trailing content yields a parse_error. Does not throw;
+/// malformed input is reported via the returned std::expected.
 ss::future<std::expected<config_info, parse_error>> parse_config(iobuf body);
 
 /// Parse the body of a `GET /subjects/{subject}/versions` response into a list
diff --git a/src/v/pandaproxy/schema_registry/rest_client/tests/client_integration_test.cc b/src/v/pandaproxy/schema_registry/rest_client/tests/client_integration_test.cc
index 88f07363d2eb2..5501e857df1fa 100644
--- a/src/v/pandaproxy/schema_registry/rest_client/tests/client_integration_test.cc
+++ b/src/v/pandaproxy/schema_registry/rest_client/tests/client_integration_test.cc
@@ -231,12 +231,12 @@ FIXTURE_TEST(sr_rest_client_integration, pandaproxy_test_fixture) {
     {
         // No global config has been set, so the registry reports its built-in
         // default. The real server emits {"compatibilityLevel":"BACKWARD"} and
-        // nothing else, so unknown_fields is empty.
+        // nothing else, so unsupported is empty.
         auto res = sut.get_config(rtc).get();
         BOOST_REQUIRE(res.has_value());
         BOOST_REQUIRE(res->level == rc::registry_compatibility_level::backward);
         BOOST_REQUIRE_EQUAL(res->raw, "BACKWARD");
-        BOOST_REQUIRE(res->unknown_fields.empty());
+        BOOST_REQUIRE(res->unsupported.empty());
     }
 
     info(
diff --git a/src/v/pandaproxy/schema_registry/rest_client/tests/client_test.cc b/src/v/pandaproxy/schema_registry/rest_client/tests/client_test.cc
index 944238961f0ea..ab1fa8047ed9f 100644
--- a/src/v/pandaproxy/schema_registry/rest_client/tests/client_test.cc
+++ b/src/v/pandaproxy/schema_registry/rest_client/tests/client_test.cc
@@ -1027,7 +1027,7 @@ TEST(rest_client, get_config_request_shape_and_success) {
     ASSERT_TRUE(res.has_value());
     EXPECT_EQ(res->level, rc::registry_compatibility_level::backward);
     EXPECT_EQ(res->raw, "BACKWARD");
-    EXPECT_TRUE(res->unknown_fields.empty());
+    EXPECT_TRUE(res->unsupported.empty());
 }
 
 TEST(rest_client, get_config_open_enum_tolerates_unknown_value) {
@@ -1051,7 +1051,7 @@ TEST(rest_client, get_config_open_enum_tolerates_unknown_value) {
 
 TEST(rest_client, get_config_records_unmodeled_fields) {
     // A Confluent registry may return a rich object; the client models only
-    // compatibilityLevel and names the rest in unknown_fields.
+    // compatibilityLevel and surfaces the rest as unsupported features.
     rc::client client{
       make_http_client([](mock_client& m) {
           EXPECT_CALL(m, request_and_collect_response(_, _, _))
@@ -1068,8 +1068,9 @@ TEST(rest_client, get_config_records_unmodeled_fields) {
 
     ASSERT_TRUE(res.has_value());
     EXPECT_EQ(res->level, rc::registry_compatibility_level::full);
-    EXPECT_THAT(
-      res->unknown_fields, ElementsAre("normalize", "validateFields"));
+    ASSERT_EQ(res->unsupported.size(), size_t{2});
+    EXPECT_EQ(res->unsupported[0].json_pointer, "/normalize");
+    EXPECT_EQ(res->unsupported[1].json_pointer, "/validateFields");
 }
 
 TEST(rest_client, get_config_no_credentials_omits_auth_header) {
@@ -1208,7 +1209,7 @@ TEST(rest_client, get_subject_config_request_shape_and_success) {
     ASSERT_TRUE(res.has_value());
     EXPECT_EQ(res->level, rc::registry_compatibility_level::none);
     EXPECT_EQ(res->raw, "NONE");
-    EXPECT_TRUE(res->unknown_fields.empty());
+    EXPECT_TRUE(res->unsupported.empty());
 }
 
 TEST(rest_client, get_subject_config_encodes_qualified_subject) {
diff --git a/src/v/pandaproxy/schema_registry/rest_client/tests/parse_test.cc b/src/v/pandaproxy/schema_registry/rest_client/tests/parse_test.cc
index bc57eb9405d1f..7a967e1dae306 100644
--- a/src/v/pandaproxy/schema_registry/rest_client/tests/parse_test.cc
+++ b/src/v/pandaproxy/schema_registry/rest_client/tests/parse_test.cc
@@ -452,13 +452,13 @@ TEST(registry_mode_test, to_string_view_and_format) {
 
 TEST_CORO(parse_config_test, compatibility_level_only) {
     // Redpanda's server emits just compatibilityLevel; nothing is recorded as
-    // an unknown field.
+    // an unsupported feature.
     auto res = co_await parse_config(
       iobuf::from(R"({"compatibilityLevel": "BACKWARD"})"));
     ASSERT_TRUE_CORO(res.has_value());
     ASSERT_EQ_CORO(res->level, registry_compatibility_level::backward);
     ASSERT_EQ_CORO(res->raw, "BACKWARD");
-    ASSERT_TRUE_CORO(res->unknown_fields.empty());
+    ASSERT_TRUE_CORO(res->unsupported.empty());
 }
 
 TEST_CORO(parse_config_test, all_known_levels_map_to_enumerators) {
@@ -504,8 +504,9 @@ TEST_CORO(parse_config_test, empty_level_is_unknown_not_error) {
 
 TEST_CORO(parse_config_test, records_unmodeled_fields) {
     // A Confluent registry may return a rich object. Only compatibilityLevel is
-    // modeled; every other top-level field's name is recorded (in encounter
-    // order) and its value skipped, whatever its shape (scalar, object, null).
+    // modeled; every other non-null top-level field is recorded (in encounter
+    // order) as an unsupported feature with a JSON pointer and type. A
+    // null-valued field is treated as absent (defaultRuleSet below).
     auto res = co_await parse_config(
       iobuf::from(
         R"({"compatibilityLevel": "FULL", "normalize": true, )"
@@ -514,11 +515,13 @@ TEST_CORO(parse_config_test, records_unmodeled_fields) {
         R"("defaultRuleSet": null})"));
     ASSERT_TRUE_CORO(res.has_value());
     ASSERT_EQ_CORO(res->level, registry_compatibility_level::full);
-    ASSERT_EQ_CORO(res->unknown_fields.size(), size_t{4});
-    ASSERT_EQ_CORO(res->unknown_fields[0], "normalize");
-    ASSERT_EQ_CORO(res->unknown_fields[1], "validateFields");
-    ASSERT_EQ_CORO(res->unknown_fields[2], "defaultMetadata");
-    ASSERT_EQ_CORO(res->unknown_fields[3], "defaultRuleSet");
+    ASSERT_EQ_CORO(res->unsupported.size(), size_t{3});
+    ASSERT_EQ_CORO(res->unsupported[0].json_pointer, "/normalize");
+    ASSERT_EQ_CORO(res->unsupported[0].json_type, "boolean");
+    ASSERT_EQ_CORO(res->unsupported[1].json_pointer, "/validateFields");
+    ASSERT_EQ_CORO(res->unsupported[1].json_type, "boolean");
+    ASSERT_EQ_CORO(res->unsupported[2].json_pointer, "/defaultMetadata");
+    ASSERT_EQ_CORO(res->unsupported[2].json_type, "object");
 }
 
 TEST_CORO(parse_config_test, compatibility_level_need_not_come_first) {
@@ -526,19 +529,55 @@ TEST_CORO(parse_config_test, compatibility_level_need_not_come_first) {
       iobuf::from(R"({"normalize": true, "compatibilityLevel": "FORWARD"})"));
     ASSERT_TRUE_CORO(res.has_value());
     ASSERT_EQ_CORO(res->level, registry_compatibility_level::forward);
-    ASSERT_EQ_CORO(res->unknown_fields.size(), size_t{1});
-    ASSERT_EQ_CORO(res->unknown_fields[0], "normalize");
+    ASSERT_EQ_CORO(res->unsupported.size(), size_t{1});
+    ASSERT_EQ_CORO(res->unsupported[0].json_pointer, "/normalize");
 }
 
-TEST_CORO(parse_config_test, missing_compatibility_level_is_error) {
-    // compatibilityLevel is the one field a config response must carry.
-    for (std::string_view body : {R"({})", R"({"normalize": true})"}) {
+TEST_CORO(parse_config_test, missing_compatibility_level_is_absent) {
+    // A config without compatibilityLevel parses with the level absent (a
+    // Confluent subject config may carry only governance fields), and a null
+    // level is treated the same way. The unmodeled fields still surface.
+    for (std::string_view body :
+         {R"({})",
+          R"({"normalize": true})",
+          R"({"compatibilityLevel": null, "normalize": true})"}) {
         SCOPED_TRACE(body);
         auto res = co_await parse_config(iobuf::from(body));
-        ASSERT_FALSE_CORO(res.has_value());
+        ASSERT_TRUE_CORO(res.has_value());
+        ASSERT_FALSE_CORO(res->level.has_value());
     }
 }
 
+TEST_CORO(parse_config_test, duplicate_compatibility_level_is_last_wins) {
+    // Duplicate keys follow last-wins JSON semantics: a later null clears an
+    // earlier value, and a later string replaces an earlier null.
+    auto cleared = co_await parse_config(
+      iobuf::from(
+        R"({"compatibilityLevel": "FULL", "compatibilityLevel": null})"));
+    ASSERT_TRUE_CORO(cleared.has_value());
+    ASSERT_FALSE_CORO(cleared->level.has_value());
+
+    auto set = co_await parse_config(
+      iobuf::from(
+        R"({"compatibilityLevel": null, "compatibilityLevel": "FULL"})"));
+    ASSERT_TRUE_CORO(set.has_value());
+    ASSERT_EQ_CORO(set->level, registry_compatibility_level::full);
+}
+
+TEST_CORO(parse_config_test, governance_only_config_surfaces_unsupported) {
+    // A subject config where only governance fields were set (legal in
+    // Confluent Data Contracts): no compatibilityLevel, and the governance
+    // field must reach the unsupported-feature policy instead of being masked
+    // by a parse error.
+    auto res = co_await parse_config(
+      iobuf::from(R"({"compatibilityGroup": "app.major.version"})"));
+    ASSERT_TRUE_CORO(res.has_value());
+    ASSERT_FALSE_CORO(res->level.has_value());
+    ASSERT_EQ_CORO(res->unsupported.size(), size_t{1});
+    ASSERT_EQ_CORO(res->unsupported[0].json_pointer, "/compatibilityGroup");
+    ASSERT_EQ_CORO(res->unsupported[0].json_type, "string");
+}
+
 TEST_CORO(parse_config_test, non_object_is_error) {
     for (std::string_view body :
          {R"(["BACKWARD"])", R"("BACKWARD")", "42", "null", "true"}) {
@@ -549,9 +588,9 @@ TEST_CORO(parse_config_test, non_object_is_error) {
 }
 
 TEST_CORO(parse_config_test, non_string_compatibility_level_is_error) {
+    // Null is absent (covered above); any other non-string level is malformed.
     for (std::string_view body :
          {R"({"compatibilityLevel": 5})",
-          R"({"compatibilityLevel": null})",
           R"({"compatibilityLevel": ["BACKWARD"]})",
           R"({"compatibilityLevel": {}})",
           R"({"compatibilityLevel": true})"}) {
@@ -590,8 +629,8 @@ TEST_CORO(parse_config_test, fragmented_input) {
     ASSERT_EQ_CORO(
       res->level, registry_compatibility_level::backward_transitive);
     ASSERT_EQ_CORO(res->raw, "BACKWARD_TRANSITIVE");
-    ASSERT_EQ_CORO(res->unknown_fields.size(), size_t{1});
-    ASSERT_EQ_CORO(res->unknown_fields[0], "normalize");
+    ASSERT_EQ_CORO(res->unsupported.size(), size_t{1});
+    ASSERT_EQ_CORO(res->unsupported[0].json_pointer, "/normalize");
 }
 
 TEST_CORO(parse_config_test, malformed_or_truncated_is_error) {
diff --git a/tests/rptest/tests/cluster_linking_schema_registry_sync_test.py b/tests/rptest/tests/cluster_linking_schema_registry_sync_test.py
index 2160ec9cf4978..46578d4d85b6f 100644
--- a/tests/rptest/tests/cluster_linking_schema_registry_sync_test.py
+++ b/tests/rptest/tests/cluster_linking_schema_registry_sync_test.py
@@ -133,10 +133,18 @@ def _register(
         subject: str,
         schema: dict,
         references: list[dict] | None = None,
+        metadata: dict | None = None,
+        rule_set: dict | None = None,
     ) -> int:
         payload: dict[str, Any] = {"schema": json.dumps(schema)}
         if references:
             payload["references"] = references
+        # Confluent-only Data Contract fields Redpanda does not model. A source
+        # that carries them lets a test drive the unsupported-feature policy.
+        if metadata is not None:
+            payload["metadata"] = metadata
+        if rule_set is not None:
+            payload["ruleSet"] = rule_set
         resp = client.post_subjects_subject_versions(
             subject=subject, data=json.dumps(payload)
         )
@@ -212,6 +220,8 @@ def _create_sr_link(
         source_filter_subjects: list[str] | None = None,
         source_filter_contexts: list[str] | None = None,
         exact_context_map: dict[str, str] | None = None,
+        feature_policy: shadow_link_pb2.UnsupportedSchemaFeaturePolicy.ValueType
+        | None = None,
     ) -> str:
         # Create a shadow link that syncs only the Schema Registry, in API mode,
         # pointing at the source cluster's SR endpoint (or an explicit URL, e.g.
@@ -233,6 +243,8 @@ def _create_sr_link(
                 seconds=full_sync_interval_sec
             ),
         )
+        if feature_policy is not None:
+            api.unsupported_schema_feature_policy = feature_policy
         if source_filter_subjects is not None:
             api.source_filter.subjects.extend(source_filter_subjects)
         if source_filter_contexts is not None:
@@ -340,10 +352,12 @@ def synced() -> bool:
     # --- status-counter checks ---------------------------------------------
 
     # Counters expected to stay zero in the override-free DAG tests: those DAGs
-    # register no mode or compatibility overrides, so mode/config replication is
-    # a no-op, and unsupported-feature handling is unimplemented. A test that
-    # sets overrides (test_schema_registry_api_sync_compatibility) asserts the
-    # compatibility counter advances.
+    # register no mode or compatibility overrides and carry no unsupported
+    # fields, so mode/config replication and unsupported-feature handling are
+    # no-ops there. Tests that do exercise them
+    # (test_schema_registry_api_sync_compatibility, the
+    # test_schema_registry_api_sync_unsupported_* suite) assert their counters
+    # advance.
     EXPECTED_ZERO_COUNTERS = (
         "compatibility_configs_changed",
         "modes_changed",
@@ -415,7 +429,7 @@ def counters_ready() -> bool:
         assert lfs.HasField("start_time") and lfs.HasField("finish_time"), lfs
         assert lfs.finish_time.ToNanoseconds() >= lfs.start_time.ToNanoseconds(), lfs
 
-        # Mapped-but-unimplemented counters must be zero (flagged above).
+        # Counters this suite does not exercise must stay zero (flagged above).
         totals = sr.totals_since_task_start
         # The cumulative summary carries the task start time (stamped once on the
         # task's first run) and, being open-ended, has no finish time.
@@ -1122,6 +1136,135 @@ def returning_leader_fresh() -> bool:
             err_msg="reused instance did not reset state on regaining leadership",
         )
 
+    # --- unsupported-feature policy ----------------------------------------
+    # Confluent-only: only a Confluent source accepts the unsupported fields
+    # (schema metadata.tags, config compatibilityGroup) on registration, so a
+    # Redpanda source cannot seed them. The destination models neither, so the
+    # sync surfaces them and applies the configured policy.
+
+    def _schema_tags(self, i: int) -> dict:
+        # A metadata.tags block: Redpanda models only metadata.properties, so
+        # this surfaces "/metadata/tags" as an unsupported schema feature.
+        return {"tags": {f"{NS}.Leaf{i}": ["PII"]}}
+
+    def _set_source_config(
+        self, client: SchemaRegistryRedpandaClient, subject: str, body: dict
+    ) -> None:
+        resp = client.set_config_subject(subject, json.dumps(body))
+        assert resp.status_code == 200, f"set config {subject} failed: {resp.text}"
+
+    def _wait_totals(self, predicate: Any, err_msg: str, timeout_sec: int = 90):
+        # Waits until the cumulative counters satisfy `predicate`, then returns
+        # the status for logging and further assertions.
+        def ready() -> bool:
+            return predicate(self._admin_sr_status().totals_since_task_start)
+
+        wait_until(ready, timeout_sec=timeout_sec, backoff_sec=1, err_msg=err_msg)
+        return self._admin_sr_status()
+
+    def _test_schema_registry_api_sync_unsupported_schema_remove(self):
+        src = self._make_source_client()
+        dest = SchemaRegistryRedpandaClient(self.target_cluster_service)
+        # A clean subject and one carrying an unsupported metadata.tags block.
+        self._register(src, "clean-value", self._leaf_schema(1))
+        self._register(
+            src, "tagged-value", self._leaf_schema(2), metadata=self._schema_tags(2)
+        )
+        self._create_sr_link(
+            feature_policy=shadow_link_pb2.UNSUPPORTED_SCHEMA_FEATURE_POLICY_REMOVE
+        )
+        # Both import (the tagged one as its supported projection); the removed
+        # feature is counted and no error is raised.
+        self._wait_synced(src, dest, [("clean-value", 1), ("tagged-value", 1)])
+        sr = self._wait_totals(
+            lambda t: t.unsupported_features_removed >= 1 and t.errors == 0,
+            "REMOVE did not count an unsupported schema feature",
+        )
+        self._log_counters("admin API", sr)
+
+    def _test_schema_registry_api_sync_unsupported_schema_fail(self):
+        src = self._make_source_client()
+        dest = SchemaRegistryRedpandaClient(self.target_cluster_service)
+        self._register(src, "clean-value", self._leaf_schema(1))
+        self._register(
+            src, "tagged-value", self._leaf_schema(2), metadata=self._schema_tags(2)
+        )
+        self._create_sr_link(
+            feature_policy=shadow_link_pb2.UNSUPPORTED_SCHEMA_FEATURE_POLICY_FAIL
+        )
+        # The clean subject syncs; the tagged one is a per-item error, skipped.
+        self._wait_synced(src, dest, [("clean-value", 1)])
+        sr = self._wait_totals(
+            lambda t: t.errors >= 1 and t.unsupported_features_removed == 0,
+            "FAIL did not count the unsupported schema as an error",
+        )
+        self._log_counters("admin API", sr)
+        assert self._schema_view(dest, "tagged-value", 1) is None, (
+            "FAIL must not import a schema carrying unsupported features"
+        )
+
+    def _test_schema_registry_api_sync_unsupported_config_remove(self):
+        src = self._make_source_client()
+        dest = SchemaRegistryRedpandaClient(self.target_cluster_service)
+        self._register(src, "cfg-value", self._leaf_schema(1))
+        # Subject config with an unsupported governance field
+        # (compatibilityGroup) alongside the supported compatibilityLevel.
+        self._set_source_config(
+            src,
+            "cfg-value",
+            {"compatibility": "FULL", "compatibilityGroup": "app.major.version"},
+        )
+        # A governance-only subject config (no compatibility level set at all):
+        # the sync must treat it as "no override" plus policy input rather than
+        # fail the config read, so the whole run stays error-free.
+        self._register(src, "gov-value", self._leaf_schema(2))
+        self._set_source_config(
+            src, "gov-value", {"compatibilityGroup": "app.major.version"}
+        )
+        self._create_sr_link(
+            feature_policy=shadow_link_pb2.UNSUPPORTED_SCHEMA_FEATURE_POLICY_REMOVE
+        )
+        self._wait_synced(src, dest, [("cfg-value", 1), ("gov-value", 1)])
+        # Each full sync re-reads both configs and re-counts their dropped
+        # field (+2 per sync), even once the writes are no-ops; >= 3 therefore
+        # proves a second sync counted, pinning the per-sync semantics.
+        sr = self._wait_totals(
+            lambda t: t.unsupported_features_removed >= 3 and t.errors == 0,
+            "REMOVE did not keep counting unsupported config features",
+        )
+        self._log_counters("admin API", sr)
+        # The supported compatibilityLevel is still synced to the destination.
+        resp = dest.get_config_subject("cfg-value")
+        assert (
+            resp.status_code == 200 and resp.json()["compatibilityLevel"] == "FULL"
+        ), f"REMOVE must still sync the compat level: {resp.status_code} {resp.text}"
+
+    def _test_schema_registry_api_sync_unsupported_config_fail(self):
+        src = self._make_source_client()
+        dest = SchemaRegistryRedpandaClient(self.target_cluster_service)
+        self._register(src, "cfg-value", self._leaf_schema(1))
+        self._set_source_config(
+            src,
+            "cfg-value",
+            {"compatibility": "FULL", "compatibilityGroup": "app.major.version"},
+        )
+        self._create_sr_link(
+            feature_policy=shadow_link_pb2.UNSUPPORTED_SCHEMA_FEATURE_POLICY_FAIL
+        )
+        # The clean schema still imports; the config carrying the unsupported
+        # field is a per-item error and its write is skipped.
+        self._wait_synced(src, dest, [("cfg-value", 1)])
+        sr = self._wait_totals(
+            lambda t: t.errors >= 1 and t.unsupported_features_removed == 0,
+            "FAIL did not count the unsupported config as an error",
+        )
+        self._log_counters("admin API", sr)
+        # The subject config write is skipped: no FULL override lands.
+        resp = dest.get_config_subject("cfg-value")
+        assert (
+            resp.status_code != 200 or resp.json().get("compatibilityLevel") != "FULL"
+        ), f"FAIL must not sync the unsupported config: {resp.status_code} {resp.text}"
+
 
 class SchemaRegistrySyncE2ETest(ShadowLinkTestBase, SchemaRegistrySyncMixin):
     """Redpanda source: a secondary Redpanda cluster provides the source Schema
@@ -1287,3 +1430,19 @@ def test_schema_registry_api_sync_hard_delete(self):
     @cluster(num_nodes=5)
     def test_schema_registry_api_sync_context_remap(self):
         self._test_schema_registry_api_sync_context_remap()
+
+    @cluster(num_nodes=5)
+    def test_schema_registry_api_sync_unsupported_schema_remove(self):
+        self._test_schema_registry_api_sync_unsupported_schema_remove()
+
+    @cluster(num_nodes=5)
+    def test_schema_registry_api_sync_unsupported_schema_fail(self):
+        self._test_schema_registry_api_sync_unsupported_schema_fail()
+
+    @cluster(num_nodes=5)
+    def test_schema_registry_api_sync_unsupported_config_remove(self):
+        self._test_schema_registry_api_sync_unsupported_config_remove()
+
+    @cluster(num_nodes=5)
+    def test_schema_registry_api_sync_unsupported_config_fail(self):
+        self._test_schema_registry_api_sync_unsupported_config_fail()