cluster_link: unsupported schema-feature policy for SR shadow sync#31047
Conversation
There was a problem hiding this comment.
Pull request overview
Implements end-to-end handling of Schema Registry “unsupported schema features” during API-mode shadow syncing, enabling migrations from Confluent-compatible registries when responses include fields Redpanda cannot store (e.g. rule sets, schema tags, metadata.tags). The SR REST client now surfaces structured diagnostics (JSON pointer + value type), and the cluster-link reconciler applies the configured unsupported_schema_feature_policy (FAIL default / REMOVE) to either skip offending schemas (per-item failure) or import the supported projection and count removals.
Changes:
- Add
source_schema_read { schema, unsupported[] }with structuredunsupported_featurediagnostics, and update SR response parsing/client API to populate it (including opt-in viaConfluent-Accept-Unknown-Propertiesand silent dropping of ignorable server-assigned fields). - Plumb
source_schema_readthrough the schema-registry sync source reader and apply the unsupported-feature policy in the reconciler (count removals underREMOVE, per-item skip+error underFAIL). - Extend unit/integration tests for parsing, client header behavior, reconciler policy semantics, and mirroring-task status/counters.
Reviewed changes
Copilot reviewed 19 out of 19 changed files in this pull request and generated 1 comment.
Show a summary per file
| File | Description |
|---|---|
| src/v/pandaproxy/schema_registry/types.h | Introduces unsupported_feature and source_schema_read types for structured unsupported-field diagnostics. |
| src/v/pandaproxy/schema_registry/rest_client/tests/parse_test.cc | Updates/extends parser tests to assert JSON-pointer/type diagnostics and ignorable/null behavior. |
| src/v/pandaproxy/schema_registry/rest_client/tests/client_test.cc | Updates client tests for unsupported and adds header opt-in + extended-field handling coverage. |
| src/v/pandaproxy/schema_registry/rest_client/tests/client_integration_test.cc | Updates integration expectations from unknown_fields to unsupported. |
| src/v/pandaproxy/schema_registry/rest_client/parse.h | Changes parse API to return source_schema_read and documents new lenient/diagnostic behavior. |
| src/v/pandaproxy/schema_registry/rest_client/parse.cc | Implements detection/surfacing of unsupported fields (JSON pointers + type) and drops ignorable fields. |
| src/v/pandaproxy/schema_registry/rest_client/config.h | Updates documentation reference for “unknown fields” vs new unsupported diagnostics. |
| src/v/pandaproxy/schema_registry/rest_client/client.h | Updates client API contract to return source_schema_read. |
| src/v/pandaproxy/schema_registry/rest_client/client.cc | Always opts into Confluent extended schema fields via Confluent-Accept-Unknown-Properties. |
| src/v/cluster_link/schema_registry_sync/unavailable_source_reader.h | Updates source-reader interface to return source_schema_read. |
| src/v/cluster_link/schema_registry_sync/unavailable_source_reader.cc | Matches unavailable-reader implementation to new return type. |
| src/v/cluster_link/schema_registry_sync/tests/sr_sync_test_fixtures.h | Extends fake source to attach unsupported diagnostics and passes feature policy into reconciler. |
| src/v/cluster_link/schema_registry_sync/tests/reconciler_test.cc | Adds coverage for REMOVE (strip+count) and FAIL (per-item error+skip, continue syncing). |
| src/v/cluster_link/schema_registry_sync/tests/mirroring_task_test.cc | Verifies task-level status/counters reflect unsupported-feature policy behavior. |
| src/v/cluster_link/schema_registry_sync/source_reader.h | Documents/updates read_subject_version to return schema + unsupported diagnostics. |
| src/v/cluster_link/schema_registry_sync/reconciler.h | Adds policy plumbing, stats counter for removals, and FAIL-policy behavior description. |
| src/v/cluster_link/schema_registry_sync/reconciler.cc | Enforces FAIL/REMOVE policy at import/discovery time; logs and counts removals after successful import. |
| src/v/cluster_link/schema_registry_sync/mirroring_task.h | Threads unsupported-feature policy into full sync execution. |
| src/v/cluster_link/schema_registry_sync/mirroring_task.cc | Accumulates unsupported_features_removed into status reports and snapshots per-link policy safely across awaits. |
bartoszpiekny-redpanda
force-pushed
the
core-16775-unsupported-feature-diagnostics
branch
2 times, most recently
from
f19f3b4 to
5c5e50b
Compare
|
Tested it locally. Example of the logs: Remove: Failure: |
bartoszpiekny-redpanda
force-pushed
the
core-16775-unsupported-feature-diagnostics
branch
from
5c5e50b to
2321ae6
Compare
bartoszpiekny-redpanda
force-pushed
the
core-16775-unsupported-feature-diagnostics
branch
from
2321ae6 to
9677b75
Compare
bartoszpiekny-redpanda
assigned
nguyen-andrew and
bartoszpiekny-redpanda and unassigned
nguyen-andrew
bartoszpiekny-redpanda
requested review from
dotnwat,
mnajda-redpanda and
nguyen-andrew
bartoszpiekny-redpanda
force-pushed
the
core-16775-unsupported-feature-diagnostics
branch
from
9677b75 to
e9f2508
Compare
bartoszpiekny-redpanda
force-pushed
the
core-16775-unsupported-feature-diagnostics
branch
from
e9f2508 to
78789b8
Compare
|
force-push: rebase on dev (and fix conflicts) |
| ASSERT_TRUE(res.has_value()); | ||
| EXPECT_EQ(res->id, pps::schema_id{100001}); | ||
| EXPECT_EQ(res->version, pps::schema_version{3}); | ||
| EXPECT_EQ(res->schema.id, pps::schema_id{100001}); | ||
| EXPECT_EQ(res->schema.version, pps::schema_version{3}); | ||
| EXPECT_TRUE(res->unsupported.empty()); |
There was a problem hiding this comment.
nit:
I'd use a matcher for all this:
EXPECT_THAT(
res,
testing::Optional(
testing::AllOf(
Field(&pps::stored_schema::id, pps::schema_id{100001}),
Field(&pps::stored_schema::version, pps::schema_version{3}),
Field(&pps::stored_schema::unsupported, testing::IsEmpty())
)
)
);| chunked_vector<ppsr::unsupported_feature> unsupported; | ||
| unsupported.push_back( | ||
| ppsr::unsupported_feature{.json_pointer = "/ruleSet"}); | ||
| unsupported.push_back( | ||
| ppsr::unsupported_feature{.json_pointer = "/metadata/tags"}); |
There was a problem hiding this comment.
nit:
you could create this in-place
| chunked_vector<ppsr::unsupported_feature> unsupported; | |
| unsupported.push_back( | |
| ppsr::unsupported_feature{.json_pointer = "/ruleSet"}); | |
| unsupported.push_back( | |
| ppsr::unsupported_feature{.json_pointer = "/metadata/tags"}); | |
| chunked_vector<ppsr::unsupported_feature> unsupported{ | |
| ppsr::unsupported_feature{.json_pointer = "/ruleSet"}, | |
| ppsr::unsupported_feature{.json_pointer = "/metadata/tags"} | |
| }; |
bartoszpiekny-redpanda
force-pushed
the
core-16775-unsupported-feature-diagnostics
branch
from
78789b8 to
7d26f6d
Compare
|
force-push: Added missing "@fmt" to the |
CI test resultstest results on build#86883
test results on build#86915
test results on build#86924
|
dotnwat
left a comment
dotnwat
left a comment
There was a problem hiding this comment.
This LGTM. Nice clean-up of the REST client interface. Just one question about reconciler. Should probably get Gellert to sign-off too.
| ss::future<bool> reconciler::import_body( | ||
| const ppsr::subject_version& n, ppsr::source_schema_read read) { | ||
| const bool removing = _feature_policy |
There was a problem hiding this comment.
why doesn't import_body have enforce the policy (it appears to only report)? It seems like a centralized place for enforcement, but currently we have the fail_if_unsupported scattered around.
There was a problem hiding this comment.
I thought it might be nice to stop before enqueuing refs if we already know we have unsupported types so that's why I've added additional check before moving forward. Let me see if I can make it more clear and refactor a bit. If you think it's not a good decision (not enqueuing refs) please let me know - should be fairly easy to make it in only one place.
There was a problem hiding this comment.
Hope it's more clear now: https://github.com/redpanda-data/redpanda/compare/79b32cf8ee..a7b4e9aa87 added also test for this scenario: https://github.com/redpanda-data/redpanda/compare/a7b4e9aa87..e8d02aa30c
bartoszpiekny-redpanda
force-pushed
the
core-16775-unsupported-feature-diagnostics
branch
from
7d26f6d to
79b32cf
Compare
|
Force-pushed a test-only cleanup on top (no product code changed):
(Note: the link uses two-dot |
bartoszpiekny-redpanda
force-pushed
the
core-16775-unsupported-feature-diagnostics
branch
from
79b32cf to
a7b4e9a
Compare
|
Force-pushed a refactor of how the unsupported-feature policy is enforced (plus one test-format fix):
This addresses the earlier review question about centralizing enforcement: FAIL and REMOVE now both live in
|
bartoszpiekny-redpanda
force-pushed
the
core-16775-unsupported-feature-diagnostics
branch
from
a7b4e9a to
e8d02aa
Compare
|
Force-pushed:
|
bartoszpiekny-redpanda
force-pushed
the
core-16775-unsupported-feature-diagnostics
branch
from
e8d02aa to
a6619b6
Compare
|
Force-pushed: |
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.
Null-valued unmodeled fields are treated as absent, and server-assigned
fields (guid, ts) are on an ignorable list, 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.
Send the Confluent-Accept-Unknown-Properties header on the schema-fetch request so a Confluent-compatible source returns the extended fields (guid, ts, deleted). The tolerant parser then sees the full response: the ignorable-field list drops the server-assigned guid/ts and deleted is modeled, so only genuinely unsupported fields surface as diagnostics, rather than relying on the server to omit them.
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.
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.
bartoszpiekny-redpanda
force-pushed
the
core-16775-unsupported-feature-diagnostics
branch
from
a6619b6 to
ed0c885
Compare
|
force-push: split commits |
| // 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)) { |
There was a problem hiding this comment.
having this authoritative gate is nice, in addition to avoiding the queing as you pointed out.
nit: before this change removed had been expressed in terms of the remove policy enum. Now with fail_if_contains_unsupported this gate is expressed in terms of the fail policy enum. That works because there are only two policies. We're probably unlikely to add another policy, but just pointing out that maybe there a footgun depending on what kind of new policy is added in the future. Nothing to do here really.
There was a problem hiding this comment.
Yeah I was also thinking a lot about that🙂 but didn't want to perform switch case in both places (fail function and remove function). I assumed when we'll be adding another policy we'll have to implement it anyway and add another new behavior.
pgellert
left a comment
pgellert
left a comment
There was a problem hiding this comment.
Config-borne governance features (compatibilityGroup, default/override metadata & rule sets on GET /config) are out of scope
I think we'll want this in and so far we only have this partially (the parser already surfaces the unknown fields, but we don't apply the policy yet). We can do this in a follow up PR though if easier.
Implements the
unsupported_schema_feature_policyconfig (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 astructured
source_schema_read { schema, unsupported[] }— eachunsupported_featurecarrying a JSON pointer (/ruleSet,/metadata/tags) andJSON 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-Propertiesheader, and a small allow-list of server-assigned fields (
guid,ts) isdropped silently.
REMOVE imports only the supported projection, increments
unsupported_features_removed(surfaced inrpk shadow statusand totals), andlogs 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-syncabort semantics.
Note: this covers the schema-fetch path only — the policy acts on per-schema
metadata/ruleSet(source_schema_read::unsupported). Config-bornegovernance features (
compatibilityGroup, default/override metadata & rule setson
GET /config) are out of scope: config sync replicates onlycompatibilityLeveland records the rest as droppedunknown_fields. Noproto/model/converter/rpk changes: the config contract, the counter, and
rpk shadow statusrendering already exist.Tests: reconciler and mirroring-task unit/integration tests cover REMOVE (strip
the rest, task stays active), plus diagnostics parsing (surfaced fields, ignored
guid/ts,Confluent-Accept-Unknown-Propertiesopt-in).Note: e2e tests will be postponed and implemented once: https://redpandadata.atlassian.net/browse/CORE-16776 is delivered. Besides unit tests PR has been tested manually.
Backports Required
Release Notes