schema_registry/rest_client: structured unsupported-feature diagnostics

Replace the names-only parsed_schema::unknown_fields with a structured
source_schema_read carrying unsupported_feature entries (JSON pointers),
so a schema migration task can apply its unsupported-feature policy to
GET /subjects/{subject}/versions/{version} responses.

Server-assigned fields (guid, ts) are dropped silently and null-valued
unmodeled fields are treated as absent, so a source that returns them
does not spuriously surface a feature. The config path is unchanged.

The unsupported_feature struct carries only the JSON pointer for now;
its JSON type and a shallow summary can be added later without changing
the source_reader signature.

Author: bartoszpiekny-redpanda

src/v/pandaproxy/schema_registry/rest_client/client.cc

@@ -35,6 +35,14 @@ namespace {
 
 constexpr std::string_view accept_json = "application/json";
 
+// Confluent returns extended schema fields (guid, ts, deleted) on a schema
+// response only when this request header is present. The migration client opts
+// in so the tolerant parser sees the full response and the ignorable-field list
+// explicitly drops the server-assigned ones (guid, ts) rather than relying on
+// the server to omit them.
+constexpr std::string_view accept_unknown_properties_header
+  = "Confluent-Accept-Unknown-Properties";
+
 // Schema Registry error_codes for the not-found conditions.
 constexpr int32_t error_code_subject_not_found = 40401;
 constexpr int32_t error_code_version_not_found = 40402;
@@ -564,7 +572,7 @@ client::list_subject_versions(
     co_return std::move(parsed.value());
 }
 
-ss::future<std::expected<parsed_schema, domain_error>>
+ss::future<std::expected<source_schema_read, domain_error>>
 client::get_schema_by_version(
   const context_subject& subject,
   schema_version version,
@@ -580,7 +588,8 @@ client::get_schema_by_version(
           .path(
             fmt::format(
               "/subjects/{}/versions/{}", encode_subject(subject), version()))
-          .header("accept", accept_json);
+          .header("accept", accept_json)
+          .header(accept_unknown_properties_header, "true");
     add_deleted_param(request, deleted);
     maybe_add_basic_auth(request);
 

src/v/pandaproxy/schema_registry/rest_client/client.h

@@ -191,11 +191,11 @@ class client {
     /// be fetched. A missing subject yields subject_not_found; a missing
     /// version (of an existing subject) yields version_not_found.
     ///
-    /// The result wraps the schema together with the names of any response
-    /// fields the client did not model (parsed_schema::unknown_fields); a
+    /// The result wraps the schema together with any unsupported response
+    /// fields the client could not store (source_schema_read::unsupported); a
     /// caller that needs fidelity can inspect them and apply its own strictness
     /// policy.
-    ss::future<std::expected<parsed_schema, domain_error>>
+    ss::future<std::expected<source_schema_read, domain_error>>
     get_schema_by_version(
       const context_subject& subject,
       schema_version version,

src/v/pandaproxy/schema_registry/rest_client/config.h

@@ -105,9 +105,12 @@ registry_compatibility_level_from_wire(std::string_view sv) {
 /// 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 (mirroring parsed_schema::unknown_fields). Redpanda's own server
-/// emits only `compatibilityLevel`, so unknown_fields is empty against it; a
-/// third-party Confluent-compatible registry may populate it.
+/// 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.
 struct config_info {
     registry_compatibility_level level;
     ss::sstring raw;

src/v/pandaproxy/schema_registry/rest_client/parse.cc

@@ -15,9 +15,12 @@
 
 #include <seastar/core/coroutine.hh>
 
+#include <algorithm>
+#include <array>
 #include <cstdint>
 #include <limits>
 #include <optional>
+#include <string_view>
 #include <utility>
 
 namespace pandaproxy::schema_registry::rest_client {
@@ -43,6 +46,59 @@ std::optional<int32_t> checked_nonnegative_i32(int64_t v) {
     return static_cast<int32_t>(v);
 }
 
+// Server-assigned response fields Redpanda does not model but which carry no
+// user content. The client opts into them via the
+// `Confluent-Accept-Unknown-Properties` header (see client.cc), so they arrive
+// on schema responses; this list drops them rather than surfacing them to the
+// unsupported-feature policy, so a source that returns them does not spuriously
+// trip it.
+//
+// A constexpr array scanned with ranges::contains is a deliberate choice at
+// this size: it keeps the list trivially extensible (just add a literal) rather
+// than a switch/case, and for N=2 a linear scan beats a hash set (no static
+// init, stays constexpr). Promote to a flat_hash_set only if this list grows
+// large or gains a bulk-lookup site (cf. cluster_link's
+// disallowed_topic_properties, materialized into a set in its validator).
+constexpr auto ignorable_fields = std::to_array<std::string_view>(
+  {"guid", "ts"});
+
+bool is_ignorable_field(std::string_view key) {
+    return std::ranges::contains(ignorable_fields, key);
+}
+
+// The field's JSON type name, for unsupported-feature diagnostics; takes the
+// parser's current value token.
+const char* json_type_name(serde::json::token t) {
+    using token = serde::json::token;
+    switch (t) {
+    case token::start_object:
+        return "object";
+    case token::start_array:
+        return "array";
+    case token::value_string:
+        return "string";
+    case token::value_int:
+    case token::value_double:
+        return "number";
+    case token::value_true:
+    case token::value_false:
+        return "boolean";
+    case token::value_null:
+        return "null";
+    // Non-value tokens never reach here (this is called only on the parser's
+    // current value token). They are enumerated rather than folded into a
+    // default so the switch stays exhaustive: -Wswitch (via -Werror) then flags
+    // a newly-added token at compile time instead of silently returning
+    // "unknown".
+    case token::error:
+    case token::key:
+    case token::end_object:
+    case token::end_array:
+    case token::eof:
+        return "unknown";
+    }
+}
+
 } // namespace
 
 ss::future<std::expected<chunked_vector<context_subject>, parse_error>>
@@ -517,21 +573,19 @@ parse_references(serde::json::parser& p, qualified_subjects_enabled qualified) {
       parse_error{.reason = "truncated or malformed references array"});
 }
 
-// The result of parsing a metadata object: the modeled portion plus the names
-// of any sub-keys we don't model.
+// The result of parsing a metadata object: the modeled portion plus any
+// unsupported sub-fields, each already reported as a `/metadata/<key>` pointer.
 struct parsed_metadata {
     schema_metadata metadata;
-    // Unmodeled keys found directly inside `metadata` (e.g. `tags`,
-    // `sensitive`), unqualified; the caller qualifies and propagates them.
-    chunked_vector<ss::sstring> unknown_fields;
+    chunked_vector<unsupported_feature> unsupported;
 };
 
 // Parse a metadata object of the form {"properties": {<str>: <str>}, ...}.
 // Entered with the current token at the object start; leaves the parser at the
 // end_object token. Only `properties` is modeled; its values are stored as
 // strings, with numbers and booleans coerced to strings to match the write
-// path. Any other key (e.g. `tags`, `sensitive`) is returned, unqualified, in
-// parsed_metadata::unknown_fields for the caller to record.
+// path. Any other non-null key (e.g. `tags`, `sensitive`) is reported in
+// parsed_metadata::unsupported as a `/metadata/<key>` pointer.
 ss::future<std::expected<parsed_metadata, parse_error>>
 parse_metadata(serde::json::parser& p) {
     using token = serde::json::token;
@@ -546,7 +600,14 @@ parse_metadata(serde::json::parser& p) {
               parse_error{.reason = "truncated JSON in schema metadata"});
         }
         if (key != "properties") {
-            result.unknown_fields.push_back(std::move(key));
+            // A null value means the sub-field is absent; anything else is an
+            // unsupported feature, surfaced as a `/metadata/<key>` pointer.
+            if (p.token() != token::value_null) {
+                result.unsupported.push_back(
+                  unsupported_feature{
+                    .json_pointer = ssx::sformat("/metadata/{}", key),
+                    .json_type = json_type_name(p.token())});
+            }
             co_await p.skip_value();
             continue;
         }
@@ -602,7 +663,7 @@ parse_metadata(serde::json::parser& p) {
 
 } // namespace
 
-ss::future<std::expected<parsed_schema, parse_error>>
+ss::future<std::expected<source_schema_read, parse_error>>
 parse_subject_version(iobuf body, qualified_subjects_enabled qualified) {
     using token = serde::json::token;
     // Firewall exceptions from the parser: malformed input is reported via the
@@ -623,7 +684,7 @@ parse_subject_version(iobuf body, qualified_subjects_enabled qualified) {
         schema_definition::references refs;
         is_deleted deleted{false};
         std::optional<schema_metadata> metadata;
-        chunked_vector<ss::sstring> unknown_fields;
+        chunked_vector<unsupported_feature> unsupported;
 
         while (co_await p.next()) {
             if (p.token() == token::end_object) {
@@ -637,8 +698,8 @@ parse_subject_version(iobuf body, qualified_subjects_enabled qualified) {
                 }
                 // Absent fields fall back to defaults/sentinels; completeness
                 // is a higher-layer concern. Unmodeled fields were recorded in
-                // unknown_fields above for the caller to act on.
-                co_return parsed_schema{
+                // `unsupported` above for the caller to act on.
+                co_return source_schema_read{
                   .schema = stored_schema{
                     .schema = subject_schema{
                       subject.value_or(invalid_subject),
@@ -651,7 +712,7 @@ parse_subject_version(iobuf body, qualified_subjects_enabled qualified) {
                     .version = version.value_or(invalid_schema_version),
                     .id = id.value_or(invalid_schema_id),
                     .deleted = deleted},
-                  .unknown_fields = std::move(unknown_fields)};
+                  .unsupported = std::move(unsupported)};
             }
             if (p.token() != token::key) {
                 co_return std::unexpected(
@@ -730,30 +791,35 @@ parse_subject_version(iobuf body, qualified_subjects_enabled qualified) {
                 refs = std::move(*r);
             } else if (key == "metadata") {
                 // Partially modeled: parse_metadata captures `properties` and
-                // returns any other sub-key (e.g. `tags`), which we qualify
-                // with a `metadata.` prefix into unknown_fields. A null
-                // metadata is treated as absent; any other non-object is
-                // unrepresentable.
+                // returns any other sub-key (e.g. `tags`) as a
+                // `/metadata/<key>` unsupported feature. A null metadata is
+                // treated as absent; any other non-object is unrepresentable.
                 if (p.token() == token::start_object) {
                     auto m = co_await parse_metadata(p);
                     if (!m) {
                         co_return std::unexpected(std::move(m.error()));
                     }
                     metadata = std::move(m->metadata);
-                    for (const auto& sub : m->unknown_fields) {
-                        unknown_fields.push_back(
-                          ssx::sformat("metadata.{}", sub));
+                    for (auto& f : m->unsupported) {
+                        unsupported.push_back(std::move(f));
                     }
                 } else if (p.token() != token::value_null) {
                     co_return std::unexpected(
                       parse_error{.reason = "metadata must be an object"});
                 }
             } else {
-                // Unknown / not-yet-modeled field (guid, ts, ruleSet,
-                // schemaTags, ...): skip its value, but record the top-level
-                // key so the caller can decide whether dropping it is
-                // acceptable.
-                unknown_fields.push_back(std::move(key));
+                // Unmodeled field. Server-assigned fields that carry no user
+                // content (`guid`, `ts`) are dropped silently; a null value
+                // means the field is absent; anything else is an unsupported
+                // feature, surfaced as a `/<key>` pointer for the migration
+                // policy to act on.
+                if (
+                  !is_ignorable_field(key) && 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();
             }
         }

src/v/pandaproxy/schema_registry/rest_client/parse.h

@@ -130,49 +130,36 @@ ss::future<std::expected<chunked_vector<subject_version>, parse_error>>
 parse_schema_id_subject_versions(
   iobuf body, qualified_subjects_enabled qualified);
 
-/// The outcome of parsing a get-schema-by-version response: the schema, plus
-/// the names of any top-level response fields the parser did not model.
-///
-/// parse_subject_version is deliberately lenient — it never rejects a response
-/// merely for carrying fields it doesn't model; it skips them and records their
-/// names here. This lets a caller that needs fidelity (e.g. schema migration)
-/// apply its own policy — reject, warn, or ignore — while a caller that doesn't
-/// care simply disregards the list. Recorded names are top-level keys, with one
-/// exception: `metadata` is only partially modeled (just `metadata.properties`
-/// is captured), so an unmodeled key directly under it is reported with a
-/// `metadata.` prefix (e.g. `metadata.tags`). An unmodeled key nested inside
-/// any other modeled field (e.g. within a reference) is skipped without being
-/// reported. It is therefore a best-effort signal that content was dropped, not
-/// a proof of a lossless round-trip.
-struct parsed_schema {
-    stored_schema schema;
-    chunked_vector<ss::sstring> unknown_fields;
-};
-
 /// Parse the body of a `GET /subjects/{subject}/versions/{version}` response
-/// into a parsed_schema (the schema plus the names of any unmodeled top-level
-/// fields).
+/// into a source_schema_read: the schema projected into Redpanda's supported
+/// model, plus a sidecar list of unsupported fields it could not store.
+///
+/// This is a faithful, lenient deserialization (the lowest layer): it never
+/// rejects a response merely for carrying fields it doesn't model. Server-
+/// assigned fields that carry no user content (`guid`, `ts`) are dropped
+/// silently. Any other unmodeled field is surfaced in
+/// source_schema_read::unsupported as a JSON pointer (e.g. `/ruleSet`), for a
+/// caller that needs fidelity (e.g. schema migration) to apply its policy —
+/// fail, remove, or ignore. `metadata` is partially modeled:
+/// `metadata.properties` is captured into the schema's metadata (values are
+/// stringified, matching the write path), while any other key under `metadata`
+/// (e.g. `metadata.tags`) is reported as `/metadata/<key>`. A field whose value
+/// is null is treated as absent (not reported). An unmodeled key nested inside
+/// another modeled field (e.g. within a reference) is skipped without being
+/// reported — so `unsupported` is a best-effort signal that content was
+/// dropped, not a proof of a lossless round-trip.
 ///
-/// This is a faithful, lenient deserialization (the lowest layer): unknown or
-/// not-yet-modeled fields (`guid`, `ts`, `ruleSet`, `schemaTags`, ...) are
-/// skipped — their names are collected in parsed_schema::unknown_fields for the
-/// caller to act on. `metadata` is partially modeled: `metadata.properties` is
-/// captured into the schema's metadata (values are stringified, matching the
-/// write path), while any other key under `metadata` (e.g. `metadata.tags`) is
-/// reported in unknown_fields under a `metadata.` prefix. Absent fields take
-/// their default/sentinel (absent `schemaType` -> AVRO, `deleted` -> false,
-/// `references` -> empty, `metadata` -> absent, and absent
-/// `subject`/`version`/`id`/`schema` -> the invalid sentinels). It does NOT
-/// enforce completeness or reject for unmodeled fields — whether an incomplete
-/// or lossy response is acceptable (a strict mode) is a higher-layer concern.
-/// It rejects only inputs it cannot represent: a non-object body, malformed
-/// JSON, a present modeled field with a wrong-typed or out-of-range value, or
-/// an unknown `schemaType`.
+/// Absent fields take their default/sentinel (absent `schemaType` -> AVRO,
+/// `deleted` -> false, `references` -> empty, `metadata` -> absent, and absent
+/// `subject`/`version`/`id`/`schema` -> the invalid sentinels). It rejects only
+/// inputs it cannot represent: a non-object body, malformed JSON, a present
+/// modeled field with a wrong-typed or out-of-range value, or an unknown
+/// `schemaType`.
 ///
 /// \p qualified is the caller-supplied policy for interpreting
 /// context-qualified subject strings (the response `subject` and each
 /// reference's `subject`). The function does not throw.
-ss::future<std::expected<parsed_schema, parse_error>>
+ss::future<std::expected<source_schema_read, parse_error>>
 parse_subject_version(iobuf body, qualified_subjects_enabled qualified);
 
 /// The structured error body Schema Registry returns on failures:

src/v/pandaproxy/schema_registry/rest_client/tests/client_integration_test.cc

@@ -294,7 +294,7 @@ FIXTURE_TEST(sr_rest_client_integration, pandaproxy_test_fixture) {
           = sut.get_schema_by_version(multi, pps::schema_version{2}, rtc).get();
         BOOST_REQUIRE(res.has_value());
         // Redpanda's SR emits only fields we model, so nothing is dropped.
-        BOOST_REQUIRE(res->unknown_fields.empty());
+        BOOST_REQUIRE(res->unsupported.empty());
         const auto& s = res->schema;
         BOOST_REQUIRE_EQUAL(s.schema.sub(), multi);
         BOOST_REQUIRE_EQUAL(s.version, pps::schema_version{2});
@@ -310,8 +310,8 @@ FIXTURE_TEST(sr_rest_client_integration, pandaproxy_test_fixture) {
               .get();
         BOOST_REQUIRE(res.has_value());
         // metadata.properties is modeled, so it parses back in full and nothing
-        // is reported as unknown.
-        BOOST_REQUIRE(res->unknown_fields.empty());
+        // is reported as unsupported.
+        BOOST_REQUIRE(res->unsupported.empty());
         const auto& def = res->schema.schema.def();
         BOOST_REQUIRE(def.meta().has_value());
         BOOST_REQUIRE(def.meta()->properties.has_value());