Merge pull request #31047 from bartoszpiekny-redpanda/core-16775-unsupported-feature-diagnostics

cluster_link: unsupported schema-feature policy for SR shadow sync

Implements the unsupported_schema_feature_policy config (FAIL default /
REMOVE) for API-mode Schema Registry shadowing, so a migration from a
Confluent-compatible source handles schema features Redpanda cannot store (rule
sets, schema tags, metadata.tags). The tolerant SR client now returns a
structured source_schema_read { schema, unsupported[] } — each
unsupported_feature carrying a JSON pointer (/ruleSet, /metadata/tags) and
JSON type — which the source reader carries through to the reconciler, which
applies the configured policy at the single per-node import point. The client
opts into extended source fields via the Confluent-Accept-Unknown-Properties
header, and a small allow-list of server-assigned fields (guid, ts) is
dropped silently.

REMOVE imports only the supported projection, increments
unsupported_features_removed (surfaced in rpk shadow status and totals), and
logs the stripped fields to the broker log. FAIL (the default) 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;
the task stays active, and only global errors (source unavailable) fail-fast.

Note: FAIL deliberately deviates from the RFC, which describes it as aborting /
faulting the whole sync. Per-item handling collects all per-item errors rather
than stopping on the first offender and keeps replicating whatever the
destination can store — more useful for a migration. The offending
subject/version and which fields tripped the policy are written to the broker
log (logger shadow_link). Happy to revisit if reviewers prefer whole-sync
abort semantics.

Note: this covers the schema-fetch path only — the policy acts on per-schema
metadata/ruleSet (source_schema_read::unsupported). Config-borne
governance features (compatibilityGroup, default/override metadata & rule sets
on GET /config) are out of scope: config sync replicates only
compatibilityLevel and records the rest as dropped unknown_fields. No
proto/model/converter/rpk changes: the config contract, the counter, and rpk shadow status rendering already exist.

Tests: reconciler and mirroring-task unit/integration tests cover REMOVE (strip

count; no count on a failed import) and FAIL (count + skip the offender, sync
the rest, task stays active), plus diagnostics parsing (surfaced fields, ignored
guid/ts, Confluent-Accept-Unknown-Properties opt-in).

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/http_source_reader.cc

@@ -266,7 +266,7 @@ http_source_reader::list_subject_versions(
     co_return std::move(res.value());
 }
 
-ss::future<source_result<ppsr::stored_schema>>
+ss::future<source_result<ppsr::source_schema_read>>
 http_source_reader::read_subject_version(
   ppsr::context_subject sub,
   ppsr::schema_version version,
@@ -285,9 +285,11 @@ http_source_reader::read_subject_version(
     if (!res.has_value()) {
         co_return std::unexpected(to_source_error(std::move(res.error())));
     }
-    // Unmodeled response fields (parsed_schema::unknown_fields) are ignored;
-    // honoring the unsupported-feature policy is future work.
-    co_return std::move(res.value().schema);
+    // Carry the read through unchanged: the schema plus any unsupported fields
+    // the source served but Redpanda cannot store. The reconciler applies the
+    // configured unsupported-feature policy to
+    // `source_schema_read::unsupported`.
+    co_return std::move(res.value());
 }
 
 ss::future<source_result<std::optional<ppsr::mode>>>

src/v/cluster_link/schema_registry_sync/http_source_reader.h

@@ -77,7 +77,7 @@ class http_source_reader final : public source_reader {
     list_subject_versions(
       ppsr::context_subject, ppsr::include_deleted, ss::abort_source&) override;
 
-    ss::future<source_result<ppsr::stored_schema>> read_subject_version(
+    ss::future<source_result<ppsr::source_schema_read>> read_subject_version(
       ppsr::context_subject, ppsr::schema_version, ss::abort_source&) override;
 
     ss::future<source_result<std::optional<ppsr::mode>>>

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,9 +294,9 @@ reconciler::discover(const ppsr::subject_version& n, ss::abort_source& as) {
         co_return source_result<void>{};
     }
 
-    adjust_units(_mem, units, body_size(fetched.value()));
+    adjust_units(_mem, units, body_size(fetched.value().schema));
 
-    auto refs = resolve_refs(fetched.value());
+    auto refs = resolve_refs(fetched.value().schema);
     chunked_hash_set<ppsr::subject_version> missing;
     for (auto& ref : refs) {
         if (!_in_scope(ref.sub)) {
@@ -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().unsupported)) {
+        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.
@@ -369,18 +380,68 @@ reconciler::do_import(const ppsr::subject_version& n, ss::abort_source& as) {
         co_return source_result<void>{};
     }
 
-    adjust_units(_mem, units, body_size(fetched.value()));
+    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_contains_unsupported(
+  const ppsr::subject_version& n,
+  const chunked_vector<ppsr::unsupported_feature>& unsupported) {
+    if (
+      _feature_policy
+        != model::schema_registry_sync_config::unsupported_feature_policy::fail
+      || 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,
+      unsupported.size(),
+      fmt::join(unsupported, ", "));
+    fail(n);
+    return true;
+}
+
+void reconciler::count_if_contains_unsupported_removed(
+  const ppsr::subject_version& n,
+  const chunked_vector<ppsr::unsupported_feature>& unsupported) {
+    if (
+      _feature_policy
+        != model::schema_registry_sync_config::unsupported_feature_policy::
+          remove
+      || unsupported.empty()) {
+        return;
+    }
+    vlog(
+      cllog.info,
+      "Removed {} unsupported schema feature(s) from {}/{}: {}",
+      unsupported.size(),
+      n.sub,
+      n.version,
+      fmt::join(unsupported, ", "));
+    _stats->unsupported_features_removed += unsupported.size();
+}
+
 ss::future<bool> reconciler::import_body(
-  const ppsr::subject_version& n, ppsr::stored_schema schema) {
+  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.unsupported)) {
+        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.
-    auto remapped = remap_for_import(*_mapper, std::move(schema));
+    auto remapped = remap_for_import(*_mapper, std::move(read.schema));
     if (!remapped.has_value()) {
         vlog(
           cllog.warn,
@@ -414,6 +475,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. Only read.schema was moved above; read.unsupported is intact.
+    count_if_contains_unsupported_removed(n, read.unsupported);
     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.
     ///
@@ -160,10 +167,35 @@ class reconciler {
     ss::future<source_result<void>>
     do_import(const ppsr::subject_version& n, ss::abort_source& as);
 
-    /// Imports a fetched body, classifying conflicts as per-item failures.
-    /// Returns true on a successful import.
+    /// 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 authoritative per-node
+    /// import point (FAIL rejection and REMOVE accounting both happen here).
     ss::future<bool>
-    import_body(const ppsr::subject_version& n, ppsr::stored_schema schema);
+    import_body(const ppsr::subject_version& n, ppsr::source_schema_read read);
+
+    /// Under the FAIL policy, treats a 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 the list is empty.
+    ///
+    /// 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 chunked_vector<ppsr::unsupported_feature>& unsupported);
+
+    /// Under the REMOVE policy, accounts for the `unsupported` fields the
+    /// parser already dropped from a schema: logs them and adds them to
+    /// `unsupported_features_removed`. A no-op under FAIL or when the list is
+    /// empty. 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 chunked_vector<ppsr::unsupported_feature>& unsupported);
 
     /// Marks a node replicated and releases dependents whose in-degree hits 0.
     void wake(const ppsr::subject_version& n);
@@ -186,6 +218,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/source_reader.h

@@ -75,7 +75,11 @@ class source_reader {
 
     /// Reads a specific subject version's schema. The reconcile engine's
     /// schema-body fetch path: called for every node it discovers and imports.
-    virtual ss::future<source_result<ppsr::stored_schema>> read_subject_version(
+    /// Returns the schema projected into Redpanda's supported model plus any
+    /// unsupported fields the source carried but Redpanda cannot store, for the
+    /// caller to apply its unsupported-feature policy.
+    virtual ss::future<source_result<ppsr::source_schema_read>>
+    read_subject_version(
       ppsr::context_subject, ppsr::schema_version, ss::abort_source&) = 0;
 
     /// Reads the source's own (non-inherited) mode override for a subject or

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

@@ -170,9 +170,20 @@ TEST(http_source_reader, read_subject_version_returns_schema) {
                  .get();
     reader.stop().get();
 
-    ASSERT_TRUE(res.has_value());
-    EXPECT_EQ(res->id, pps::schema_id{100001});
-    EXPECT_EQ(res->version, pps::schema_version{3});
+    EXPECT_THAT(
+      res,
+      Optional(AllOf(
+        Field(
+          "schema",
+          &pps::source_schema_read::schema,
+          AllOf(
+            Field("id", &pps::stored_schema::id, pps::schema_id{100001}),
+            Field(
+              "version",
+              &pps::stored_schema::version,
+              pps::schema_version{3}))),
+        Field(
+          "unsupported", &pps::source_schema_read::unsupported, IsEmpty()))));
 }
 
 // Discovery and body fetch must request deleted entries so the reconcile can