cluster_link: carry schema-read diagnostics through the source reader

read_subject_version now returns source_schema_read (the schema plus the
unsupported fields the source carried but Redpanda cannot store) instead
of a bare stored_schema, so the reconciler can later apply the
unsupported-feature policy at a single per-node import point.

This commit only threads the diagnostics through the interface: the
stub, the test fake, and the reconciler are updated to the new type, but
the reconciler does not yet act on the unsupported set, so behavior is
unchanged. A test asserts a read carrying diagnostics still imports.

Author: bartoszpiekny-redpanda

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

@@ -289,9 +289,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)) {
@@ -369,18 +369,18 @@ 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>{};
 }
 
 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) {
     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,

src/v/cluster_link/schema_registry_sync/reconciler.h

@@ -160,10 +160,13 @@ 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 single per-node import
+    /// point.
     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);
 
     /// Marks a node replicated and releases dependents whose in-degree hits 0.
     void wake(const ppsr::subject_version& n);

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

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

@@ -57,6 +57,28 @@ TEST(reconciler, replicates_chain_in_reference_order) {
     EXPECT_LT(ia, ib);
 }
 
+// A read carrying unsupported-feature diagnostics still imports normally: the
+// source_reader interface carries `unsupported` through, but the reconciler
+// does not yet act on it (policy enforcement is a later commit).
+TEST(reconciler, imports_schema_with_unsupported_diagnostics) {
+    reconcile_harness h;
+    auto a = ppsr::context_subject::unqualified("a");
+    h.source.add(a, 1);
+    chunked_vector<ppsr::unsupported_feature> unsupported;
+    unsupported.push_back(
+      ppsr::unsupported_feature{.json_pointer = "/ruleSet"});
+    h.source.set_unsupported(a, 1, std::move(unsupported));
+
+    srs::work_set work;
+    work.upserts.push_back(key(a, 1));
+
+    auto stats = h.run(std::move(work)).get();
+    ASSERT_TRUE(stats.has_value());
+    EXPECT_EQ(stats->versions_changed, 1);
+    EXPECT_EQ(stats->errors, 0);
+    EXPECT_GE(index_of(h.destination.get_all(), "a"), 0);
+}
+
 // Tolerance test: a schema that lists the same reference twice is handled and
 // the referrer still imports referent-first. Holds with or without the dedup.
 TEST(reconciler, duplicate_reference_still_imports) {

src/v/cluster_link/schema_registry_sync/tests/sr_sync_test_fixtures.h

@@ -141,6 +141,21 @@ struct fake_source_state {
           std::move(err));
     }
 
+    // Unsupported features to attach to a specific (subject, version)'s read,
+    // letting a test drive the reconciler's unsupported-feature policy.
+    chunked_hash_map<
+      ppsr::subject_version,
+      chunked_vector<ppsr::unsupported_feature>>
+      unsupported_features;
+
+    void set_unsupported(
+      const ppsr::context_subject& sub,
+      int32_t version,
+      chunked_vector<ppsr::unsupported_feature> features) {
+        unsupported_features[ppsr::subject_version{
+          sub, ppsr::schema_version{version}}] = std::move(features);
+    }
+
     void add(
       const ppsr::context_subject& sub,
       int32_t version,
@@ -244,7 +259,8 @@ class fake_source_reader final : public srs::source_reader {
         co_return versions;
     }
 
-    ss::future<srs::source_result<ppsr::stored_schema>> read_subject_version(
+    ss::future<srs::source_result<ppsr::source_schema_read>>
+    read_subject_version(
       ppsr::context_subject sub,
       ppsr::schema_version version,
       ss::abort_source&) override {
@@ -257,7 +273,15 @@ class fake_source_reader final : public srs::source_reader {
         }
         for (const auto& s : _state->schemas) {
             if (s.schema.sub() == sub && s.version == version) {
-                co_return s.share();
+                chunked_vector<ppsr::unsupported_feature> unsupported;
+                if (
+                  auto it = _state->unsupported_features.find(
+                    ppsr::subject_version{sub, version});
+                  it != _state->unsupported_features.end()) {
+                    unsupported = it->second.copy();
+                }
+                co_return ppsr::source_schema_read{
+                  .schema = s.share(), .unsupported = std::move(unsupported)};
             }
         }
         co_return std::unexpected(

src/v/cluster_link/schema_registry_sync/unavailable_source_reader.cc

@@ -39,7 +39,7 @@ unavailable_source_reader::list_subject_versions(
     co_return std::unexpected(unavailable());
 }
 
-ss::future<source_result<ppsr::stored_schema>>
+ss::future<source_result<ppsr::source_schema_read>>
 unavailable_source_reader::read_subject_version(
   ppsr::context_subject, ppsr::schema_version, ss::abort_source&) {
     co_return std::unexpected(unavailable());

src/v/cluster_link/schema_registry_sync/unavailable_source_reader.h

@@ -43,7 +43,7 @@ class unavailable_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>>>