cluster_link: apply the unsupported-feature policy on schema import

The reconciler now reads the link's configured policy and acts on the
unsupported fields a source schema carries. REMOVE writes only the
supported projection, counts the removed fields, and logs them. FAIL
treats an unsupported schema as a per-item failure: it logs the offending
fields, counts an error, and skips that schema while the rest of the
subjects still replicate. If unset, the policy defaults to FAIL.

FAIL deliberately does not abort the whole sync -- only global errors
like source unavailability fail-fast. This collects all per-item errors
and keeps replicating whatever the destination can store, which is more
useful for a migration than stopping on the first offender. The
unsupported detail (subject/version and which fields) is written to the
broker log; the error count flows through reconcile_stats into the sync
summary.

REMOVE is counted at the single per-node import point (once, even on the
2-fetch path), and its count reaches unsupported_features_removed in the
sync summary (current sync and totals), already surfaced by rpk shadow
status. The source reader stays policy-agnostic.

Author: bartoszpiekny-redpanda

src/v/cluster_link/schema_registry_sync/BUILD

@@ -76,6 +76,7 @@ redpanda_cc_library(
         "//src/v/ssx:future_util",
         "//src/v/ssx:sformat",
         "@abseil-cpp//absl/container:flat_hash_set",
+        "@fmt",
     ],
     visibility = ["//visibility:public"],
     deps = [

src/v/cluster_link/schema_registry_sync/mirroring_task.cc

@@ -215,9 +215,13 @@ model::task_status_report mirroring_task::get_status_report() const {
             status.current_sync->summary.subject_versions_changed
               += _reconcile_stats.versions_changed;
             status.current_sync->summary.errors += _reconcile_stats.errors;
+            status.current_sync->summary.unsupported_features_removed
+              += _reconcile_stats.unsupported_features_removed;
             status.totals_since_task_start.subject_versions_changed
               += _reconcile_stats.versions_changed;
             status.totals_since_task_start.errors += _reconcile_stats.errors;
+            status.totals_since_task_start.unsupported_features_removed
+              += _reconcile_stats.unsupported_features_removed;
         }
         report.detail = model::task_detail{
           .schema_registry_sync_status = std::move(status)};
@@ -499,6 +503,7 @@ ss::future<> mirroring_task::list_one_subject(
 ss::future<task::state_transition> mirroring_task::full_source_sync(
   ss::abort_source& as,
   const chunked_hash_set<ppsr::context>& contexts,
+  model::schema_registry_sync_config::unsupported_feature_policy feature_policy,
   const ss::noncopyable_function<bool(const ppsr::context_subject&)>&
     in_scope) {
     // Cluster-global, so safe to read mid-sync (unlike the per-link config a
@@ -624,7 +629,8 @@ ss::future<task::state_transition> mirroring_task::full_source_sync(
       _destination,
       [&in_scope](const ppsr::context_subject& cs) { return in_scope(cs); },
       _mapper,
-      limits};
+      limits,
+      feature_policy};
 
     // The reconciler increments _reconcile_stats live (reflected mid-sync by
     // get_status_report); the fold below moves them into persistent state.
@@ -655,9 +661,13 @@ ss::future<task::state_transition> mirroring_task::full_source_sync(
     _status.current_sync->summary.subject_versions_changed
       += stats.versions_changed;
     _status.current_sync->summary.errors += stats.errors;
+    _status.current_sync->summary.unsupported_features_removed
+      += stats.unsupported_features_removed;
     _status.totals_since_task_start.subject_versions_changed
       += stats.versions_changed;
     _status.totals_since_task_start.errors += stats.errors;
+    _status.totals_since_task_start.unsupported_features_removed
+      += stats.unsupported_features_removed;
 
     const auto purged = co_await purge_destination_only_versions(
       std::move(to_purge), as);
@@ -826,9 +836,13 @@ mirroring_task::run_impl(ss::abort_source& as) {
     // phase shares one mapping without re-reading _config (which a concurrent
     // update_config could swap mid-run).
     _mapper = context_mapper::make(_config);
+    // Snapshot the per-link policy while `api` is known non-null. A concurrent
+    // update_config can swap `_config` across the co_awaits below (and inside
+    // full_source_sync), so it must not re-read api_mode() after suspending.
+    const auto feature_policy = api->feature_policy;
 
     co_await refresh_destination_inventory(in_scope, as);
-    co_return co_await full_source_sync(as, contexts, in_scope);
+    co_return co_await full_source_sync(as, contexts, feature_policy, in_scope);
 }
 
 task::state_transition

src/v/cluster_link/schema_registry_sync/mirroring_task.h

@@ -123,6 +123,8 @@ class mirroring_task : public task {
     ss::future<state_transition> full_source_sync(
       ss::abort_source&,
       const chunked_hash_set<ppsr::context>& contexts,
+      model::schema_registry_sync_config::unsupported_feature_policy
+        feature_policy,
       const ss::noncopyable_function<bool(const ppsr::context_subject&)>&
         in_scope);
 

src/v/cluster_link/schema_registry_sync/reconciler.cc

@@ -22,6 +22,8 @@
 #include <seastar/coroutine/as_future.hh>
 #include <seastar/util/defer.hh>
 
+#include <fmt/ranges.h>
+
 #include <algorithm>
 #include <array>
 #include <chrono>
@@ -110,12 +112,14 @@ reconciler::reconciler(
   schema::registry* destination,
   ss::noncopyable_function<bool(const ppsr::context_subject&)> in_scope,
   const context_mapper& mapper,
-  limits lim)
+  limits lim,
+  model::schema_registry_sync_config::unsupported_feature_policy feature_policy)
   : _source(source)
   , _destination(destination)
   , _in_scope(std::move(in_scope))
   , _mapper(&mapper)
   , _limits(lim)
+  , _feature_policy(feature_policy)
   , _mem(std::max<size_t>(1, lim.memory_bytes), "schema_registry_sync/memory") {
     // Floor the budget so the clamp `min(body_size, memory_bytes)` and the
     // semaphore agree; a degenerate 0 admits one over-budget body at a time.
@@ -138,6 +142,7 @@ ss::future<source_result<void>> reconciler::reconcile(
     _stats = &stats;
     _stats->versions_changed = 0;
     _stats->errors = 0;
+    _stats->unsupported_features_removed = 0;
     _outstanding = 0;
     _done = false;
     _fault.reset();
@@ -316,6 +321,12 @@ reconciler::discover(const ppsr::subject_version& n, ss::abort_source& as) {
         co_return source_result<void>{};
     }
 
+    // Reject under FAIL before queuing references, so a schema we are going to
+    // reject does not drag its dependency subtree into the destination.
+    if (fail_if_contains_unsupported(n, fetched.value())) {
+        co_return source_result<void>{};
+    }
+
     // The body is released here (fetched goes out of scope): the node will be
     // re-fetched once its references complete, leading to it being fetched
     // twice in this case; and at most twice on average overall.
@@ -375,8 +386,56 @@ reconciler::do_import(const ppsr::subject_version& n, ss::abort_source& as) {
     co_return source_result<void>{};
 }
 
+bool reconciler::fail_if_contains_unsupported(
+  const ppsr::subject_version& n, const ppsr::source_schema_read& read) {
+    if (
+      _feature_policy
+        != model::schema_registry_sync_config::unsupported_feature_policy::fail
+      || read.unsupported.empty()) {
+        return false;
+    }
+    // The FAIL policy refuses to import a schema the destination cannot store
+    // faithfully, but this is a per-item failure, not a whole-sync abort: count
+    // it, log the offending fields, and let the rest of the subjects replicate.
+    vlog(
+      cllog.warn,
+      "{}/{} carries {} unsupported schema feature(s) the destination cannot "
+      "store; failing it under the FAIL policy: {}",
+      n.sub,
+      n.version,
+      read.unsupported.size(),
+      fmt::join(read.unsupported, ", "));
+    fail(n);
+    return true;
+}
+
+void reconciler::count_if_contains_unsupported_removed(
+  const ppsr::subject_version& n, const ppsr::source_schema_read& read) {
+    if (
+      _feature_policy
+        != model::schema_registry_sync_config::unsupported_feature_policy::
+          remove
+      || read.unsupported.empty()) {
+        return;
+    }
+    vlog(
+      cllog.info,
+      "Removed {} unsupported schema feature(s) from {}/{}: {}",
+      read.unsupported.size(),
+      n.sub,
+      n.version,
+      fmt::join(read.unsupported, ", "));
+    _stats->unsupported_features_removed += read.unsupported.size();
+}
+
 ss::future<bool> reconciler::import_body(
   const ppsr::subject_version& n, ppsr::source_schema_read read) {
+    // Authoritative policy gate, next to the REMOVE handling below: every
+    // import path funnels through here (discover also rejects early, before
+    // queuing refs).
+    if (fail_if_contains_unsupported(n, read)) {
+        co_return false;
+    }
     data(n).state = node_state::importing;
     // The graph key `n` stays in the source namespace; only the schema written
     // to the destination is remapped.
@@ -414,6 +473,10 @@ ss::future<bool> reconciler::import_body(
     }
     data(n).state = node_state::done;
     ++_stats->versions_changed;
+    // Account for REMOVE-stripped features only after the projection actually
+    // lands on the destination, so a failed import does not report features as
+    // removed. read.schema is moved-from above, but read.unsupported is intact.
+    count_if_contains_unsupported_removed(n, read);
     wake(n);
     co_return true;
 }

src/v/cluster_link/schema_registry_sync/reconciler.h

@@ -11,6 +11,7 @@
 
 #pragma once
 
+#include "cluster_link/model/types.h"
 #include "cluster_link/schema_registry_sync/scope.h"
 #include "cluster_link/schema_registry_sync/source_reader.h"
 #include "container/chunked_hash_map.h"
@@ -37,9 +38,13 @@ struct reconcile_stats {
     /// Source versions imported into the destination this run.
     uint64_t versions_changed{0};
     /// Per-item failures (unresolvable references, import conflicts, unparsable
-    /// schemas). These do not abort the run; the reconciler counts them and
-    /// continues.
+    /// schemas, and -- under the FAIL policy -- source schemas carrying
+    /// unsupported fields). These do not abort the run; the reconciler counts
+    /// them and continues so the rest of the subjects still replicate.
     uint64_t errors{0};
+    /// Unsupported fields removed from imported schemas this run under the
+    /// REMOVE policy.
+    uint64_t unsupported_features_removed{0};
 };
 
 /// The set of changes a reconcile applies: each upsert is a (context, subject,
@@ -97,7 +102,9 @@ class reconciler {
       schema::registry* destination,
       ss::noncopyable_function<bool(const ppsr::context_subject&)> in_scope,
       const context_mapper& mapper,
-      limits lim);
+      limits lim,
+      model::schema_registry_sync_config::unsupported_feature_policy
+        feature_policy);
 
     /// Imports `work_set.upserts` referent-first.
     ///
@@ -163,11 +170,31 @@ class reconciler {
     /// Imports a fetched read, classifying conflicts as per-item failures.
     /// Returns true on a successful import. Carries the whole
     /// source_schema_read (not just the schema) so the unsupported-feature
-    /// policy can act on `read.unsupported` at the single per-node import
-    /// point.
+    /// policy can act on `read.unsupported` at the authoritative per-node
+    /// import point (FAIL rejection and REMOVE accounting both happen here).
     ss::future<bool>
     import_body(const ppsr::subject_version& n, ppsr::source_schema_read read);
 
+    /// Under the FAIL policy, treats a source schema carrying unsupported
+    /// fields as a per-item failure: logs the offending fields, marks the node
+    /// (and its dependents) failed via fail(n), and returns true so the caller
+    /// skips the import. The rest of the sync proceeds -- fail-fast is reserved
+    /// for global errors like source unavailability. Returns false (a no-op)
+    /// under REMOVE or when `read` carries nothing unsupported.
+    ///
+    /// Called by `import_body` (authoritative gate) and by `discover` before
+    /// queuing a node's references.
+    bool fail_if_contains_unsupported(
+      const ppsr::subject_version& n, const ppsr::source_schema_read& read);
+
+    /// Under the REMOVE policy, accounts for the unsupported fields the parser
+    /// already dropped from `read`: logs them and adds them to
+    /// `unsupported_features_removed`. A no-op under FAIL or when nothing was
+    /// dropped. Call only after the import succeeds, so a failed import does
+    /// not report features as removed.
+    void count_if_contains_unsupported_removed(
+      const ppsr::subject_version& n, const ppsr::source_schema_read& read);
+
     /// Marks a node replicated and releases dependents whose in-degree hits 0.
     void wake(const ppsr::subject_version& n);
 
@@ -189,6 +216,8 @@ class reconciler {
     ss::noncopyable_function<bool(const ppsr::context_subject&)> _in_scope;
     const context_mapper* _mapper;
     limits _limits;
+    model::schema_registry_sync_config::unsupported_feature_policy
+      _feature_policy;
 
     chunked_hash_set<ppsr::subject_version> _replicated;
     chunked_hash_map<ppsr::subject_version, node_data> _nodes;

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

@@ -251,6 +251,63 @@ TEST_F(mirroring_task_test, full_sync_imports_and_reports) {
     EXPECT_LT(index_of(all, "a"), index_of(all, "b"));
 }
 
+TEST_F(mirroring_task_test, remove_policy_strips_and_counts_unsupported) {
+    auto a = ppsr::context_subject::unqualified("a");
+    _source_state.add(a, 1);
+    _source_state.set_unsupported(a, 1, {{.json_pointer = "/ruleSet"}});
+
+    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 supported projection is imported and the removed feature is counted.
+    EXPECT_EQ(status->last_full_sync->subject_versions_changed, 1);
+    EXPECT_EQ(status->last_full_sync->unsupported_features_removed, 1);
+    EXPECT_EQ(status->totals_since_task_start.unsupported_features_removed, 1);
+    EXPECT_GE(index_of(_registry.get_all(), "a"), 0);
+}
+
+TEST_F(mirroring_task_test, fail_policy_counts_unsupported_and_syncs_rest) {
+    // Under FAIL an unsupported feature is a counted per-item error, not a
+    // whole-sync abort: the offending subject is skipped while the rest sync,
+    // and the task stays active. Fail-fast is reserved for global errors like
+    // source unavailability.
+    auto a = ppsr::context_subject::unqualified("a"); // unsupported -> skipped
+    auto b = ppsr::context_subject::unqualified("b"); // clean -> imported
+    _source_state.add(a, 1);
+    _source_state.add(b, 1);
+    _source_state.set_unsupported(a, 1, {{.json_pointer = "/ruleSet"}});
+
+    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();
+
+    // The full sync completes (the task does not fault): the unsupported
+    // subject is counted as an error and skipped, and the clean subject is
+    // imported.
+    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->subject_versions_changed, 1);
+    EXPECT_EQ(status->last_full_sync->unsupported_features_removed, 0);
+    EXPECT_EQ(index_of(_registry.get_all(), "a"), -1);
+    EXPECT_GE(index_of(_registry.get_all(), "b"), 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{