schema_registry: enforce strict validation for missing required attributes

The `is_object_required_superset` function in the JSON schema validator contained a bug where `std::views::filter` was incorrectly dropping missing required fields during compatibility checks. This allowed schemas with missing required attributes to falsely pass validation.
This commit removes the faulty filter to strictly enforce superset validation. If a reader schema demands a required attribute that the writer schema lacks (and does not provide a default for), the validator now correctly identifies the incompatibility.
Additionally, updated the `test_json_compat_messages` test case to expect the newly exposed `required_attribute_added` error, which was previously masked by this bug.

Author: TheJohnMatti

src/v/pandaproxy/schema_registry/json.cc

@@ -1670,41 +1670,39 @@ json_compatibility_result is_object_required_superset(
   const json::Value& newer,
   const std::filesystem::path& p) {
     json_compatibility_result res;
-    // to pass the check, a required property from newer has to be present in
-    // older, or if new it needs to have a default value.
-    // note that:
-    // 1. we check only required properties that are in both newer["properties"]
-    // and older["properties"]
-    // 2. there is no explicit check that older has an open content model
-    //    there might be a property name outside of (1) that could be rejected
-    //    by update, if update["additionalProperties"] is false
+    // older is a superset of newer iff every required property in older is
+    // also required in newer, with one exception: if older provides a
+    // "default" for the property, the consumer can fill it in when a writer
+    // (newer) omits it, so it doesn't have to be in newer.required.
+    //
+    // Note: this does not check for new required properties added on the
+    // newer side when older is closed (additionalProperties: false). That
+    // case is covered separately by
+    // required_property_added_to_unopen_content_model (TODO).
 
     auto older_req = get_array_or_empty(older, "required");
     auto newer_req = get_array_or_empty(newer, "required");
     auto older_props = get_object_or_empty(older, "properties");
-    auto newer_props = get_object_or_empty(newer, "properties");
-
-    // TODO O(n^2) lookup that can be a set_intersection.
-    auto older_req_in_both_properties
-      = older_req | std::views::filter([&](const json::Value& o) {
-            return newer_props.HasMember(o) && older_props.HasMember(o);
-        });
 
     // for each element:
     // in older.required? | in newer.required? | result
     //       yes          |        yes         |  yes
     //       yes          |         no         |  if it has "default" in older
     //       no           |        yes         |  yes
-    std::ranges::for_each(
-      older_req_in_both_properties, [&](const json::Value& o) {
-          if (
-            std::ranges::find(newer_req, o) == newer_req.End()
-            && !older_props.FindMember(o)->value.HasMember("default")) {
-              res.emplace<json_incompatibility>(
-                p / "required" / as_string_view(o),
-                json_incompatibility_type::required_attribute_added);
-          }
-      });
+
+    // TODO O(n^2) lookup that can be optimized using a flat_hash_set of
+    // newer_req.
+    std::ranges::for_each(older_req, [&](const json::Value& o) {
+        auto it = older_props.FindMember(o);
+        bool has_default = it != older_props.MemberEnd()
+                           && it->value.HasMember("default");
+        if (
+          std::ranges::find(newer_req, o) == newer_req.End() && !has_default) {
+            res.emplace<json_incompatibility>(
+              p / "required" / as_string_view(o),
+              json_incompatibility_type::required_attribute_added);
+        }
+    });
     return res;
 }
 

src/v/pandaproxy/schema_registry/test/test_json_schema.cc

@@ -812,6 +812,21 @@ static const auto compatibility_test_cases = std::to_array<compatibility_test_ca
     // Note: this is reported as combined_type_changed by the reference impl
     .compat_result = {{"#/", incompat_t::combined_type_subschemas_changed}},
   },
+  // #30398: adding a required field whose name is absent from the writer's
+  // properties must be flagged. Previously masked by a faulty filter.
+  {
+    .reader_schema = R"({
+      "type": "object",
+      "properties": {"name": {"type": "string"}, "id": {"type": "integer"}},
+      "required": ["name", "id"]
+    })",
+    .writer_schema = R"({
+      "type": "object",
+      "properties": {"name": {"type": "string"}},
+      "required": ["name"]
+    })",
+    .compat_result = {{"#/required/id", incompat_t::required_attribute_added}},
+  },
   // object checks: removing a required property not ok if didn't have a default value
   {
     .reader_schema = R"(
@@ -2298,6 +2313,7 @@ const absl::flat_hash_set<incompatibility> forward_expected{
   // for this schema update.
   // {"#/properties/cccc",
   //  incompat_t::required_property_added_to_unopen_content_model},
+  {"#/required/cccc", incompat_t::required_attribute_added},
   {"#/dependencies/a", incompat_t::dependency_array_extended},
   {"#/dependencies/d", incompat_t::dependency_array_added},
   {"#/dependencies/b", incompat_t::type_narrowed},