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/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();
@@ -289,6 +294,9 @@ reconciler::discover(const ppsr::subject_version& n, ss::abort_source& as) {
         co_return source_result<void>{};
     }
 
+    if (fail_if_unsupported(n, fetched.value())) {
+        co_return source_result<void>{};
+    }
     adjust_units(_mem, units, body_size(fetched.value().schema));
 
     auto refs = resolve_refs(fetched.value().schema);
@@ -369,14 +377,44 @@ reconciler::do_import(const ppsr::subject_version& n, ss::abort_source& as) {
         co_return source_result<void>{};
     }
 
+    if (fail_if_unsupported(n, fetched.value())) {
+        co_return source_result<void>{};
+    }
     adjust_units(_mem, units, body_size(fetched.value().schema));
 
     co_await import_body(n, std::move(fetched.value()));
     co_return source_result<void>{};
 }
 
+bool reconciler::fail_if_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;
+}
+
 ss::future<bool> reconciler::import_body(
   const ppsr::subject_version& n, ppsr::source_schema_read read) {
+    const bool removing = _feature_policy
+                            == model::schema_registry_sync_config::
+                              unsupported_feature_policy::remove
+                          && !read.unsupported.empty();
     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 +452,19 @@ ss::future<bool> reconciler::import_body(
     }
     data(n).state = node_state::done;
     ++_stats->versions_changed;
+    // Count and log the 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.
+    if (removing) {
+        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();
+    }
     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.
     ///
@@ -168,6 +175,15 @@ class reconciler {
     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.
+    bool fail_if_unsupported(
+      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 +205,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,69 @@ 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);
+    chunked_vector<ppsr::unsupported_feature> unsupported;
+    unsupported.push_back(
+      ppsr::unsupported_feature{.json_pointer = "/ruleSet"});
+    _source_state.set_unsupported(a, 1, std::move(unsupported));
+
+    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);
+    chunked_vector<ppsr::unsupported_feature> unsupported;
+    unsupported.push_back(
+      ppsr::unsupported_feature{.json_pointer = "/ruleSet"});
+    _source_state.set_unsupported(a, 1, std::move(unsupported));
+
+    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{