diff --git a/src/v/cluster_link/schema_registry_sync/http_source_reader.cc b/src/v/cluster_link/schema_registry_sync/http_source_reader.cc index 631e6be04df5d..ad320dc4fb637 100644 --- a/src/v/cluster_link/schema_registry_sync/http_source_reader.cc +++ b/src/v/cluster_link/schema_registry_sync/http_source_reader.cc @@ -285,10 +285,11 @@ http_source_reader::read_subject_version( if (!res.has_value()) { co_return std::unexpected(to_source_error(std::move(res.error()))); } - // 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`. + // Carry the read through unchanged: the schema, any unsupported fields the + // source served but Redpanda cannot store, and whether the source reported + // the `deleted` flag. The reconciler applies the configured + // unsupported-feature policy to `source_schema_read::unsupported` and + // prefers the reported `deleted` over its listing-derived fallback. co_return std::move(res.value()); } diff --git a/src/v/cluster_link/schema_registry_sync/mirroring_task.cc b/src/v/cluster_link/schema_registry_sync/mirroring_task.cc index a26ae5195639e..329ae1eb2f22a 100644 --- a/src/v/cluster_link/schema_registry_sync/mirroring_task.cc +++ b/src/v/cluster_link/schema_registry_sync/mirroring_task.cc @@ -602,10 +602,9 @@ ss::future mirroring_task::full_source_sync( node); if (!dest_deleted) { work.upserts.push_back(node); - // Authoritative deleted-state from the listing partition: the - // reconciler imports these soft-deleted regardless of the - // per-version source body, which a standard source (e.g. Confluent) - // does not populate with a `deleted` flag by default. + // Listing-derived soft-delete, recorded as the fallback signal: the + // reconciler prefers the version body's own `deleted` flag and + // consults this set only when the source omits it from the body. work.soft_deleted.insert(node); } } diff --git a/src/v/cluster_link/schema_registry_sync/reconciler.cc b/src/v/cluster_link/schema_registry_sync/reconciler.cc index c8d93b8a52929..bcc58cede9846 100644 --- a/src/v/cluster_link/schema_registry_sync/reconciler.cc +++ b/src/v/cluster_link/schema_registry_sync/reconciler.cc @@ -45,10 +45,10 @@ constexpr size_t reconcile_reserve_bytes = 100_KiB; // on them. Unqualified references resolve against the referring schema's own // context. chunked_vector -resolve_refs(const ppsr::stored_schema& stored) { - const auto& parent_ctx = stored.schema.sub().ctx; +resolve_refs(const ppsr::subject_schema& schema) { + const auto& parent_ctx = schema.sub().ctx; chunked_vector out; - for (const auto& ref : stored.schema.def().refs()) { + for (const auto& ref : schema.def().refs()) { out.push_back( ppsr::subject_version{ref.sub.resolve(parent_ctx), ref.version}); } @@ -57,8 +57,8 @@ resolve_refs(const ppsr::stored_schema& stored) { // Byte-size proxy for a schema body: the canonical definition's length. This // is what the byte-semaphore budgets, and what tests control via the fake. -size_t body_size(const ppsr::stored_schema& s) { - return s.schema.def().raw()().size_bytes(); +size_t body_size(const ppsr::subject_schema& s) { + return s.def().raw()().size_bytes(); } // Rewrites a schema's contexts source->destination for import: the subject's @@ -439,16 +439,21 @@ ss::future reconciler::import_body( if (fail_if_contains_unsupported(n, read.unsupported)) { co_return false; } - // Deleted-state is authoritative from the caller's active-vs-deleted - // listing partition, not the per-version source body: a standard source - // (e.g. Confluent) omits the `deleted` flag from that body by default, so - // the fetched flag cannot be trusted to propagate a soft-delete. - read.schema.deleted = _soft_deleted.contains(n) ? ppsr::is_deleted::yes - : ppsr::is_deleted::no; data(n).state = node_state::importing; + // into_stored() consumes `read`, so lift out the unsupported-feature list + // first: it is accounted for only after a successful import below. + const auto unsupported = std::move(read.unsupported); + // Materialize the stored schema, preferring the source's own reported + // deleted flag: it is fresher (read on a later call than the listing) and + // avoids the listing set-difference, which can race. Fall back to the + // caller's active-vs-deleted listing partition only when the source omitted + // the flag, as a source that does not report it cannot otherwise convey a + // soft-delete. + auto stored = std::move(read).into_stored( + _soft_deleted.contains(n) ? ppsr::is_deleted::yes : ppsr::is_deleted::no); // 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(read.schema)); + auto remapped = remap_for_import(*_mapper, std::move(stored)); if (!remapped.has_value()) { vlog( cllog.warn, @@ -484,8 +489,8 @@ ss::future reconciler::import_body( ++_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); + // removed. + count_if_contains_unsupported_removed(n, unsupported); wake(n); co_return true; } diff --git a/src/v/cluster_link/schema_registry_sync/tests/http_source_reader_test.cc b/src/v/cluster_link/schema_registry_sync/tests/http_source_reader_test.cc index 4a5a1859d0d68..94ce105d84563 100644 --- a/src/v/cluster_link/schema_registry_sync/tests/http_source_reader_test.cc +++ b/src/v/cluster_link/schema_registry_sync/tests/http_source_reader_test.cc @@ -173,15 +173,9 @@ TEST(http_source_reader, read_subject_version_returns_schema) { EXPECT_THAT( res, Optional(AllOf( + Field("id", &pps::source_schema_read::id, pps::schema_id{100001}), 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}))), + "version", &pps::source_schema_read::version, pps::schema_version{3}), Field( "unsupported", &pps::source_schema_read::unsupported, IsEmpty())))); } diff --git a/src/v/cluster_link/schema_registry_sync/tests/reconciler_test.cc b/src/v/cluster_link/schema_registry_sync/tests/reconciler_test.cc index b7732ce0df18f..1434208e88882 100644 --- a/src/v/cluster_link/schema_registry_sync/tests/reconciler_test.cc +++ b/src/v/cluster_link/schema_registry_sync/tests/reconciler_test.cc @@ -21,6 +21,8 @@ #include #include +#include + #include #include #include @@ -364,20 +366,29 @@ TEST(reconciler, seed_replicated_upsert_is_still_imported) { EXPECT_GE(h.source.reads(a, 1), 1); } -// A soft-deleted source version imports preserving its deleted state: the -// caller declares it in work_set.soft_deleted and import_schema stores it -// soft-deleted. Underpins the task syncing soft-deleted source versions -// preserving their deleted state. -TEST(reconciler, imports_soft_deleted_as_deleted) { +// Soft-delete propagation onto an active destination version via the +// listing-derived fallback: with a source that omits the `deleted` flag from +// the body, re-importing the version with the caller's declared deleted-state +// transitions an existing active version to soft-deleted. The node is in the +// seed (a destination version is in `all`), so this also guards that a seeded +// upsert is still imported. Underpins the task propagating a source soft-delete +// onto a live destination version. +TEST(reconciler, propagates_soft_delete_over_active_destination) { reconcile_harness h; auto a = ppsr::context_subject::unqualified("a"); + // Destination already holds a:v1 active; the source has soft-deleted it but + // does not echo the flag, so only the listing-derived set conveys it. + h.destination.import_schema(make_schema(a, 1, R"({"v":1})")).get(); + h.source.reports_deleted_flag = false; h.source.add(a, 1, ppsr::is_deleted::yes); + chunked_hash_set seed; + seed.insert(key(a, 1)); srs::work_set work; work.upserts.push_back(key(a, 1)); work.soft_deleted.insert(key(a, 1)); - auto stats = h.run(std::move(work)).get(); + auto stats = h.run(std::move(work), std::move(seed)).get(); ASSERT_TRUE(stats.has_value()); EXPECT_EQ(stats->versions_changed, 1); EXPECT_EQ(stats->errors, 0); @@ -388,26 +399,24 @@ TEST(reconciler, imports_soft_deleted_as_deleted) { EXPECT_EQ(all[ia].deleted, ppsr::is_deleted::yes); } -// Soft-delete propagation onto an active destination version: re-importing the -// version with the caller's declared deleted-state transitions an existing -// active version to soft-deleted. The node is in the seed (a destination -// version is in `all`), so this also guards that a seeded upsert is still -// imported. Underpins the task propagating a source soft-delete onto a live -// destination version. -TEST(reconciler, propagates_soft_delete_over_active_destination) { +// When the source omits the `deleted` flag from the per-version body, the +// caller's listing-derived soft-delete set is the fallback: +// read_subject_version reports deleted == nullopt, so declaring the node in +// work_set.soft_deleted still lands it soft-deleted on the destination. +// Guards the fallback path for a source that does not echo the flag. +TEST(reconciler, listing_soft_delete_used_when_source_omits_flag) { reconcile_harness h; auto a = ppsr::context_subject::unqualified("a"); - // Destination already holds a:v1 active; the source has soft-deleted it. - h.destination.import_schema(make_schema(a, 1, R"({"v":1})")).get(); - h.source.add(a, 1, ppsr::is_deleted::yes); + // Source does not echo a deleted flag, so the body cannot convey the + // soft-delete; only the listing-derived set can. + h.source.reports_deleted_flag = false; + h.source.add(a, 1, ppsr::is_deleted::no); - chunked_hash_set seed; - seed.insert(key(a, 1)); srs::work_set work; work.upserts.push_back(key(a, 1)); work.soft_deleted.insert(key(a, 1)); - auto stats = h.run(std::move(work), std::move(seed)).get(); + auto stats = h.run(std::move(work)).get(); ASSERT_TRUE(stats.has_value()); EXPECT_EQ(stats->versions_changed, 1); EXPECT_EQ(stats->errors, 0); @@ -418,32 +427,61 @@ TEST(reconciler, propagates_soft_delete_over_active_destination) { EXPECT_EQ(all[ia].deleted, ppsr::is_deleted::yes); } -// The caller's declared soft-delete wins over the per-version source body: a -// standard (e.g. Confluent) source omits the `deleted` flag from that body by -// default, so read_subject_version reports the version active, yet declaring it -// in work_set.soft_deleted still lands it soft-deleted on the destination. -// Guards the cross-vendor soft-delete propagation the HTTP source reader -// relies on. -TEST(reconciler, caller_soft_delete_wins_over_active_source_body) { +// The source's reported deleted flag wins over the listing-derived set when the +// two disagree: a version the source reports soft-deleted lands deleted even +// though it is absent from work_set.soft_deleted. The body is authoritative +// when present. +TEST(reconciler, reported_deleted_wins_over_listing_absent) { + reconcile_harness h; + auto a = ppsr::context_subject::unqualified("a"); + // Source reports the version soft-deleted; the listing set does NOT list + // it. + h.source.add(a, 1, ppsr::is_deleted::yes); + + srs::work_set work; + work.upserts.push_back(key(a, 1)); + // work.soft_deleted intentionally left empty. + + EXPECT_THAT( + h.run(std::move(work)).get(), + testing::Optional( + testing::AllOf( + testing::Field( + "versions_changed", &srs::reconcile_stats::versions_changed, 1), + testing::Field("errors", &srs::reconcile_stats::errors, 0)))); + + const auto& all = h.destination.get_all(); + auto ia = index_of(all, "a"); + ASSERT_GE(ia, 0); + EXPECT_EQ(all[ia].deleted, ppsr::is_deleted::yes); +} + +// The mirror case, and the point of preferring the body: a fresh body reporting +// the version active wins over a stale listing set that still lists it as +// soft-deleted (the listing was read earlier and may have raced). The version +// lands active despite being in work_set.soft_deleted. +TEST(reconciler, reported_active_wins_over_stale_listing_soft_delete) { reconcile_harness h; auto a = ppsr::context_subject::unqualified("a"); - // Source body reports the version ACTIVE, like a standard SR that does not - // echo a deleted flag. + // Source reports the version active; the stale listing set still lists it. h.source.add(a, 1, ppsr::is_deleted::no); srs::work_set work; work.upserts.push_back(key(a, 1)); work.soft_deleted.insert(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_THAT( + h.run(std::move(work)).get(), + testing::Optional( + testing::AllOf( + testing::Field( + "versions_changed", &srs::reconcile_stats::versions_changed, 1), + testing::Field("errors", &srs::reconcile_stats::errors, 0)))); const auto& all = h.destination.get_all(); auto ia = index_of(all, "a"); ASSERT_GE(ia, 0); - EXPECT_EQ(all[ia].deleted, ppsr::is_deleted::yes); + EXPECT_EQ(all[ia].deleted, ppsr::is_deleted::no); } TEST(reconciler, cyclic_source_does_not_hang) { diff --git a/src/v/cluster_link/schema_registry_sync/tests/sr_sync_test_fixtures.h b/src/v/cluster_link/schema_registry_sync/tests/sr_sync_test_fixtures.h index 047df6d3971d9..ff251f6d91dce 100644 --- a/src/v/cluster_link/schema_registry_sync/tests/sr_sync_test_fixtures.h +++ b/src/v/cluster_link/schema_registry_sync/tests/sr_sync_test_fixtures.h @@ -107,6 +107,12 @@ key(const ppsr::context_subject& sub, int32_t version) { struct fake_source_state { chunked_vector contexts{ppsr::default_context}; chunked_vector schemas; + // Whether the source echoes the `deleted` flag in per-version bodies. True + // (Redpanda, and Confluent queried with the extended-fields header) makes + // read_subject_version report source_schema_read::deleted; false models a + // source that omits it entirely, so deleted is nullopt and the caller must + // fall back to the listing-derived soft-delete set. + bool reports_deleted_flag = true; std::optional list_contexts_error; std::optional list_subjects_error; // Forces list_subject_versions to fail for specific subjects, letting a @@ -280,8 +286,18 @@ class fake_source_reader final : public srs::source_reader { it != _state->unsupported_features.end()) { unsupported = it->second.copy(); } + // A source that reports the flag echoes its state; one that + // omits it yields nullopt, and the reconciler's fallback (via + // into_stored) supplies the deleted state instead. + auto shared = s.share(); co_return ppsr::source_schema_read{ - .schema = s.share(), .unsupported = std::move(unsupported)}; + .schema = std::move(shared.schema), + .version = shared.version, + .id = shared.id, + .deleted = _state->reports_deleted_flag + ? std::optional{s.deleted} + : std::nullopt, + .unsupported = std::move(unsupported)}; } } co_return std::unexpected( diff --git a/src/v/pandaproxy/schema_registry/rest_client/parse.cc b/src/v/pandaproxy/schema_registry/rest_client/parse.cc index a62cbf4edceaf..45a44fc6b4134 100644 --- a/src/v/pandaproxy/schema_registry/rest_client/parse.cc +++ b/src/v/pandaproxy/schema_registry/rest_client/parse.cc @@ -682,7 +682,7 @@ parse_subject_version(iobuf body, qualified_subjects_enabled qualified) { std::optional schema; schema_type type{schema_type::avro}; schema_definition::references refs; - is_deleted deleted{false}; + std::optional deleted; std::optional metadata; chunked_vector unsupported; @@ -698,20 +698,22 @@ 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 - // `unsupported` above for the caller to act on. + // `unsupported` above for the caller to act on. `deleted` stays + // nullopt when the body omitted the flag, so a caller can tell + // an absent flag from an explicit false; into_stored() resolves + // it to a concrete value with the caller's fallback. co_return source_schema_read{ - .schema = stored_schema{ - .schema = subject_schema{ - subject.value_or(invalid_subject), - schema_definition{ - schema_definition::raw_string{ - std::move(schema).value_or(iobuf{})}, - type, - std::move(refs), - std::move(metadata)}}, - .version = version.value_or(invalid_schema_version), - .id = id.value_or(invalid_schema_id), - .deleted = deleted}, + .schema = subject_schema{ + subject.value_or(invalid_subject), + schema_definition{ + schema_definition::raw_string{ + std::move(schema).value_or(iobuf{})}, + type, + std::move(refs), + std::move(metadata)}}, + .version = version.value_or(invalid_schema_version), + .id = id.value_or(invalid_schema_id), + .deleted = deleted, .unsupported = std::move(unsupported)}; } if (p.token() != token::key) { diff --git a/src/v/pandaproxy/schema_registry/rest_client/tests/client_integration_test.cc b/src/v/pandaproxy/schema_registry/rest_client/tests/client_integration_test.cc index 88f07363d2eb2..c840e60866a1f 100644 --- a/src/v/pandaproxy/schema_registry/rest_client/tests/client_integration_test.cc +++ b/src/v/pandaproxy/schema_registry/rest_client/tests/client_integration_test.cc @@ -295,7 +295,7 @@ FIXTURE_TEST(sr_rest_client_integration, pandaproxy_test_fixture) { BOOST_REQUIRE(res.has_value()); // Redpanda's SR emits only fields we model, so nothing is dropped. BOOST_REQUIRE(res->unsupported.empty()); - const auto& s = res->schema; + const auto& s = res.value(); BOOST_REQUIRE_EQUAL(s.schema.sub(), multi); BOOST_REQUIRE_EQUAL(s.version, pps::schema_version{2}); BOOST_REQUIRE_GE(s.id(), 1); @@ -312,7 +312,7 @@ FIXTURE_TEST(sr_rest_client_integration, pandaproxy_test_fixture) { // metadata.properties is modeled, so it parses back in full and nothing // is reported as unsupported. BOOST_REQUIRE(res->unsupported.empty()); - const auto& def = res->schema.schema.def(); + const auto& def = res->schema.def(); BOOST_REQUIRE(def.meta().has_value()); BOOST_REQUIRE(def.meta()->properties.has_value()); const auto& props = def.meta()->properties.value(); @@ -327,7 +327,7 @@ FIXTURE_TEST(sr_rest_client_integration, pandaproxy_test_fixture) { = sut.get_schema_by_version(ctx_sub, pps::schema_version{1}, rtc) .get(); BOOST_REQUIRE(res.has_value()); - const auto& s = res->schema; + const auto& s = res.value(); BOOST_REQUIRE_EQUAL(s.schema.sub(), ctx_sub); BOOST_REQUIRE_EQUAL(s.version, pps::schema_version{1}); } @@ -378,7 +378,7 @@ FIXTURE_TEST(sr_rest_client_integration, pandaproxy_test_fixture) { = sut.get_schema_by_version(multi, pps::schema_version{1}, rtc).get(); BOOST_REQUIRE(v1.has_value()); - auto res = sut.get_schema_id_subject_versions(v1->schema.id, rtc).get(); + auto res = sut.get_schema_id_subject_versions(v1->id, rtc).get(); BOOST_REQUIRE(res.has_value()); // Order is not guaranteed; check membership. BOOST_REQUIRE(sv_contains(res.value(), multi, pps::schema_version{1})); @@ -407,8 +407,7 @@ FIXTURE_TEST(sr_rest_client_integration, pandaproxy_test_fixture) { BOOST_REQUIRE(cs.has_value()); auto res - = sut.get_schema_id_subject_versions(cs->schema.id, rtc, ctx_sub) - .get(); + = sut.get_schema_id_subject_versions(cs->id, rtc, ctx_sub).get(); BOOST_REQUIRE(res.has_value()); BOOST_REQUIRE( sv_contains(res.value(), ctx_sub, pps::schema_version{1})); @@ -460,10 +459,10 @@ FIXTURE_TEST(sr_rest_client_integration, pandaproxy_test_fixture) { pps::include_deleted::yes) .get(); BOOST_REQUIRE(found.has_value()); - BOOST_REQUIRE_EQUAL(found->schema.version, pps::schema_version{1}); + BOOST_REQUIRE_EQUAL(found->version, pps::schema_version{1}); // Only the per-version response carries an explicit deleted flag; - // confirm it round-trips into stored_schema. - BOOST_REQUIRE(found->schema.deleted == pps::is_deleted::yes); + // confirm it round-trips as the source-reported flag. + BOOST_REQUIRE(found->deleted == pps::is_deleted::yes); } info("list_subjects hides a fully-deleted subject without deleted"); diff --git a/src/v/pandaproxy/schema_registry/rest_client/tests/client_test.cc b/src/v/pandaproxy/schema_registry/rest_client/tests/client_test.cc index 944238961f0ea..b2050846fc6f3 100644 --- a/src/v/pandaproxy/schema_registry/rest_client/tests/client_test.cc +++ b/src/v/pandaproxy/schema_registry/rest_client/tests/client_test.cc @@ -1586,7 +1586,7 @@ TEST(rest_client, get_schema_by_version_success) { ASSERT_TRUE(res.has_value()); EXPECT_TRUE(res->unsupported.empty()); - const auto& s = res->schema; + const auto& s = res.value(); EXPECT_EQ(s.schema.sub(), subject); EXPECT_EQ(s.version, pps::schema_version{3}); EXPECT_EQ(s.id, pps::schema_id{100001}); diff --git a/src/v/pandaproxy/schema_registry/rest_client/tests/parse_test.cc b/src/v/pandaproxy/schema_registry/rest_client/tests/parse_test.cc index bc57eb9405d1f..40fdd024022ae 100644 --- a/src/v/pandaproxy/schema_registry/rest_client/tests/parse_test.cc +++ b/src/v/pandaproxy/schema_registry/rest_client/tests/parse_test.cc @@ -41,8 +41,9 @@ iobuf fragmented_iobuf(std::string_view s, size_t chunk_size) { return buf; } -// Linearize the raw schema text out of a parsed stored_schema for comparison. -ss::sstring schema_text(const stored_schema& s) { +// Linearize the raw schema text out of a parsed source_schema_read for +// comparison. +ss::sstring schema_text(const source_schema_read& s) { return s.schema.def().raw()().linearize_to_string(); } @@ -941,13 +942,15 @@ TEST_CORO(parse_subject_version_test, minimal_avro_defaults) { qualified_subjects_enabled::yes); ASSERT_TRUE_CORO(res.has_value()); ASSERT_TRUE_CORO(res.value().unsupported.empty()); - const auto& s = res.value().schema; + const auto& s = res.value(); ASSERT_EQ_CORO( s.schema.sub(), (context_subject{default_context, subject{"User"}})); ASSERT_EQ_CORO(s.version, schema_version{1}); ASSERT_EQ_CORO(s.id, schema_id{100001}); ASSERT_EQ_CORO(s.schema.type(), schema_type::avro); // default - ASSERT_EQ_CORO(s.deleted, is_deleted::no); // default + // Absent `deleted`: nullopt, so the caller can tell it apart from an + // explicit false. + ASSERT_FALSE_CORO(s.deleted.has_value()); ASSERT_TRUE_CORO(s.schema.def().refs().empty()); ASSERT_EQ_CORO(schema_text(s), R"({"type":"string"})"); } @@ -959,7 +962,7 @@ TEST_CORO(parse_subject_version_test, schema_types) { R"("schema":"{\"type\":\"object\"}"})"), qualified_subjects_enabled::yes); ASSERT_TRUE_CORO(json.has_value()); - ASSERT_EQ_CORO(json.value().schema.schema.type(), schema_type::json); + ASSERT_EQ_CORO(json.value().schema.type(), schema_type::json); // PROTOBUF: the .proto text (quotes and newlines) is preserved verbatim. auto proto = co_await parse_subject_version( @@ -968,7 +971,7 @@ TEST_CORO(parse_subject_version_test, schema_types) { R"("schema":"syntax = \"proto3\";\nmessage M {}\n"})"), qualified_subjects_enabled::yes); ASSERT_TRUE_CORO(proto.has_value()); - const auto& p = proto.value().schema; + const auto& p = proto.value(); ASSERT_EQ_CORO(p.schema.type(), schema_type::protobuf); ASSERT_EQ_CORO(schema_text(p), "syntax = \"proto3\";\nmessage M {}\n"); } @@ -983,7 +986,7 @@ TEST_CORO(parse_subject_version_test, full_object_maps_all_fields) { qualified_subjects_enabled::yes); ASSERT_TRUE_CORO(res.has_value()); ASSERT_TRUE_CORO(res.value().unsupported.empty()); - const auto& s = res.value().schema; + const auto& s = res.value(); ASSERT_EQ_CORO( s.schema.sub(), (context_subject{context{".ctx"}, subject{"MyRecord"}})); ASSERT_EQ_CORO(s.version, schema_version{2}); @@ -1000,6 +1003,32 @@ TEST_CORO(parse_subject_version_test, full_object_maps_all_fields) { ASSERT_EQ_CORO(refs[0].version, schema_version{1}); } +TEST_CORO(parse_subject_version_test, reports_deleted_presence) { + constexpr std::string_view base + = R"("subject":"User","version":1,"id":1,"schema":"\"string\"")"; + + // Present and true: `deleted` mirrors the flag. + auto present_true = co_await parse_subject_version( + iobuf::from(fmt::format("{{{},\"deleted\":true}}", base)), + qualified_subjects_enabled::yes); + ASSERT_TRUE_CORO(present_true.has_value()); + ASSERT_EQ_CORO(present_true.value().deleted, is_deleted::yes); + + // Present and false: distinct from absent -- an explicit `no`, not nullopt. + auto present_false = co_await parse_subject_version( + iobuf::from(fmt::format("{{{},\"deleted\":false}}", base)), + qualified_subjects_enabled::yes); + ASSERT_TRUE_CORO(present_false.has_value()); + ASSERT_EQ_CORO(present_false.value().deleted, is_deleted::no); + + // Absent: nullopt, so a caller can tell it apart from an explicit false. + auto absent = co_await parse_subject_version( + iobuf::from(fmt::format("{{{}}}", base)), + qualified_subjects_enabled::yes); + ASSERT_TRUE_CORO(absent.has_value()); + ASSERT_FALSE_CORO(absent.value().deleted.has_value()); +} + TEST_CORO(parse_subject_version_test, reference_subject_honors_policy) { constexpr std::string_view body = R"({"subject":"r","version":1,"id":2,"schema":"x",)" @@ -1008,7 +1037,7 @@ TEST_CORO(parse_subject_version_test, reference_subject_honors_policy) { auto on = co_await parse_subject_version( iobuf::from(body), qualified_subjects_enabled::yes); ASSERT_TRUE_CORO(on.has_value()); - const auto& on_ref = on.value().schema.schema.def().refs()[0]; + const auto& on_ref = on.value().schema.def().refs()[0]; ASSERT_EQ_CORO(on_ref.sub.qualified, is_qualified::yes); ASSERT_EQ_CORO( on_ref.sub.sub, (context_subject{context{".ctx"}, subject{"Sub"}})); @@ -1016,7 +1045,7 @@ TEST_CORO(parse_subject_version_test, reference_subject_honors_policy) { auto off = co_await parse_subject_version( iobuf::from(body), qualified_subjects_enabled::no); ASSERT_TRUE_CORO(off.has_value()); - const auto& off_ref = off.value().schema.schema.def().refs()[0]; + const auto& off_ref = off.value().schema.def().refs()[0]; ASSERT_EQ_CORO(off_ref.sub.qualified, is_qualified::no); ASSERT_EQ_CORO( off_ref.sub.sub, @@ -1040,7 +1069,7 @@ TEST_CORO(parse_subject_version_test, surfaces_unsupported_fields) { R"("schemaTags":[{"tags":["PII"]}],"futureField":[1,2,3]})"), qualified_subjects_enabled::yes); ASSERT_TRUE_CORO(res.has_value()); - const auto& s = res.value().schema; + const auto& s = res.value(); ASSERT_EQ_CORO( s.schema.sub(), (context_subject{default_context, subject{"User"}})); ASSERT_EQ_CORO(s.version, schema_version{1}); @@ -1090,7 +1119,7 @@ TEST_CORO(parse_subject_version_test, metadata_properties_coercion) { qualified_subjects_enabled::yes); ASSERT_TRUE_CORO(res.has_value()); ASSERT_TRUE_CORO(res.value().unsupported.empty()); - const auto& meta = res.value().schema.schema.def().meta(); + const auto& meta = res.value().schema.def().meta(); ASSERT_TRUE_CORO(meta.has_value()); ASSERT_TRUE_CORO(meta->properties.has_value()); const auto& props = *meta->properties; @@ -1111,7 +1140,7 @@ TEST_CORO(parse_subject_version_test, metadata_present_without_properties) { R"({"tags":{"f":["PII"]},"sensitive":["ssn"]}})"), qualified_subjects_enabled::yes); ASSERT_TRUE_CORO(res.has_value()); - const auto& meta = res.value().schema.schema.def().meta(); + const auto& meta = res.value().schema.def().meta(); ASSERT_TRUE_CORO(meta.has_value()); ASSERT_FALSE_CORO(meta->properties.has_value()); const auto& unsupported = res.value().unsupported; @@ -1129,14 +1158,14 @@ TEST_CORO(parse_subject_version_test, metadata_empty_and_null) { qualified_subjects_enabled::yes); ASSERT_TRUE_CORO(empty_obj.has_value()); ASSERT_TRUE_CORO(empty_obj.value().unsupported.empty()); - ASSERT_TRUE_CORO(empty_obj.value().schema.schema.def().meta().has_value()); + ASSERT_TRUE_CORO(empty_obj.value().schema.def().meta().has_value()); auto null_meta = co_await parse_subject_version( iobuf::from(R"({"subject":"r","version":1,"id":2,"metadata":null})"), qualified_subjects_enabled::yes); ASSERT_TRUE_CORO(null_meta.has_value()); ASSERT_TRUE_CORO(null_meta.value().unsupported.empty()); - ASSERT_FALSE_CORO(null_meta.value().schema.schema.def().meta().has_value()); + ASSERT_FALSE_CORO(null_meta.value().schema.def().meta().has_value()); // metadata.properties: null leaves properties absent (metadata present). auto null_props = co_await parse_subject_version( @@ -1144,7 +1173,7 @@ TEST_CORO(parse_subject_version_test, metadata_empty_and_null) { R"({"subject":"r","version":1,"metadata":{"properties":null}})"), qualified_subjects_enabled::yes); ASSERT_TRUE_CORO(null_props.has_value()); - const auto& meta = null_props.value().schema.schema.def().meta(); + const auto& meta = null_props.value().schema.def().meta(); ASSERT_TRUE_CORO(meta.has_value()); ASSERT_FALSE_CORO(meta->properties.has_value()); } @@ -1154,18 +1183,18 @@ TEST_CORO(parse_subject_version_test, absent_fields_use_sentinels_not_error) { auto empty = co_await parse_subject_version( iobuf::from("{}"), qualified_subjects_enabled::yes); ASSERT_TRUE_CORO(empty.has_value()); - const auto& e = empty.value().schema; + const auto& e = empty.value(); ASSERT_EQ_CORO(e.version, invalid_schema_version); ASSERT_EQ_CORO(e.id, invalid_schema_id); ASSERT_EQ_CORO(e.schema.sub(), invalid_subject); ASSERT_EQ_CORO(e.schema.type(), schema_type::avro); - ASSERT_EQ_CORO(e.deleted, is_deleted::no); + ASSERT_FALSE_CORO(e.deleted.has_value()); auto no_schema = co_await parse_subject_version( iobuf::from(R"({"subject":"r","version":1,"id":2})"), qualified_subjects_enabled::yes); ASSERT_TRUE_CORO(no_schema.has_value()); - ASSERT_EQ_CORO(no_schema.value().schema.version, schema_version{1}); + ASSERT_EQ_CORO(no_schema.value().version, schema_version{1}); } TEST_CORO(parse_subject_version_test, id_zero_is_valid) { @@ -1175,7 +1204,7 @@ TEST_CORO(parse_subject_version_test, id_zero_is_valid) { iobuf::from(R"({"subject":"r","version":1,"id":0,"schema":"x"})"), qualified_subjects_enabled::yes); ASSERT_TRUE_CORO(res.has_value()); - ASSERT_EQ_CORO(res.value().schema.id, schema_id{0}); + ASSERT_EQ_CORO(res.value().id, schema_id{0}); } TEST_CORO(parse_subject_version_test, rejects_unrepresentable) { @@ -1223,7 +1252,7 @@ TEST_CORO(parse_subject_version_test, fragmented_input) { auto res = co_await parse_subject_version( fragmented_iobuf(body, 1), qualified_subjects_enabled::yes); ASSERT_TRUE_CORO(res.has_value()); - const auto& s = res.value().schema; + const auto& s = res.value(); ASSERT_EQ_CORO( s.schema.sub(), (context_subject{context{".ctx"}, subject{"User"}})); ASSERT_EQ_CORO(s.version, schema_version{12}); @@ -1297,7 +1326,7 @@ TEST_CORO(parse_subject_version_test, reference_unknown_keys_skipped) { qualified_subjects_enabled::yes); ASSERT_TRUE_CORO(res.has_value()); ASSERT_TRUE_CORO(res->unsupported.empty()); - const auto& refs = res->schema.schema.def().refs(); + const auto& refs = res->schema.def().refs(); ASSERT_EQ_CORO(refs.size(), size_t{1}); ASSERT_EQ_CORO(refs[0].name, "n"); ASSERT_EQ_CORO(refs[0].version, schema_version{2}); diff --git a/src/v/pandaproxy/schema_registry/types.h b/src/v/pandaproxy/schema_registry/types.h index 3fd9e7fad9895..8ad2fae49565e 100644 --- a/src/v/pandaproxy/schema_registry/types.h +++ b/src/v/pandaproxy/schema_registry/types.h @@ -24,6 +24,7 @@ #include #include +#include #include namespace avro { @@ -772,8 +773,24 @@ struct unsupported_feature { /// schema projected into Redpanda's supported model, plus any unsupported /// fields that were seen but not stored. struct source_schema_read { - stored_schema schema; + subject_schema schema; + schema_version version{invalid_schema_version}; + schema_id id{invalid_schema_id}; + /// The soft-delete state the source explicitly reported for this version, + /// or nullopt when the source omitted `deleted` from the body. + std::optional deleted; chunked_vector unsupported; + + /// Materializes the read into a stored_schema, resolving \ref deleted to + /// the source-reported flag when present, or \p fallback when the source + /// omitted it (\ref deleted is nullopt). Consumes the read. + stored_schema into_stored(is_deleted fallback = is_deleted::no) && { + return { + .schema = std::move(schema), + .version = version, + .id = id, + .deleted = deleted.value_or(fallback)}; + } }; ///\brief A mapping of version and schema id for a subject.