From 776090cddfa8f283e71cac2ef5f8b6e130319cb7 Mon Sep 17 00:00:00 2001
From: Ben Pope <ben@redpanda.com>
Date: Fri, 23 Aug 2024 21:28:09 +0100
Subject: [PATCH] schema_registry: Tolerate null metadata and ruleSet

In the absence of supporting these features, it's friendly to tolerate
clients that explicitly set them to null, such as the latest DotNet
client.

Fixes #23038

Signed-off-by: Ben Pope <ben@redpanda.com>
(cherry picked from commit 55bd576e345459a4468c357962bdb2541e600f31)
---
 .../requests/post_subject_versions.h          | 36 +++++++++++++++++++
 tests/rptest/tests/schema_registry_test.py    | 28 +++++++++++++++
 2 files changed, 64 insertions(+)

diff --git a/src/v/pandaproxy/schema_registry/requests/post_subject_versions.h b/src/v/pandaproxy/schema_registry/requests/post_subject_versions.h
index 455cdfaf38143..5af67ea2ec4e9 100644
--- a/src/v/pandaproxy/schema_registry/requests/post_subject_versions.h
+++ b/src/v/pandaproxy/schema_registry/requests/post_subject_versions.h
@@ -36,6 +36,8 @@ class post_subject_versions_request_handler
         schema,
         id,
         version,
+        metadata,
+        ruleset,
         schema_type,
         references,
         reference,
@@ -74,6 +76,8 @@ class post_subject_versions_request_handler
                                      .match("schema", state::schema)
                                      .match("id", state::id)
                                      .match("version", state::version)
+                                     .match("metadata", state::metadata)
+                                     .match("ruleSet", state::ruleset)
                                      .match("schemaType", state::schema_type)
                                      .match("references", state::references)
                                      .default_match(std::nullopt)};
@@ -97,6 +101,8 @@ class post_subject_versions_request_handler
         case state::schema:
         case state::id:
         case state::version:
+        case state::metadata:
+        case state::ruleset:
         case state::schema_type:
         case state::references:
         case state::reference_name:
@@ -107,6 +113,28 @@ class post_subject_versions_request_handler
         return false;
     }
 
+    bool Null() {
+        switch (_state) {
+        case state::metadata:
+        case state::ruleset:
+            _state = state::record;
+            return true;
+        case state::empty:
+        case state::record:
+        case state::schema:
+        case state::id:
+        case state::version:
+        case state::schema_type:
+        case state::references:
+        case state::reference:
+        case state::reference_name:
+        case state::reference_subject:
+        case state::reference_version:
+            break;
+        }
+        return false;
+    }
+
     bool Uint(int i) {
         switch (_state) {
         case state::id: {
@@ -127,6 +155,8 @@ class post_subject_versions_request_handler
         case state::empty:
         case state::record:
         case state::schema:
+        case state::metadata:
+        case state::ruleset:
         case state::schema_type:
         case state::references:
         case state::reference:
@@ -170,6 +200,8 @@ class post_subject_versions_request_handler
         case state::record:
         case state::id:
         case state::version:
+        case state::metadata:
+        case state::ruleset:
         case state::references:
         case state::reference:
         case state::reference_version:
@@ -193,6 +225,8 @@ class post_subject_versions_request_handler
         case state::schema:
         case state::id:
         case state::version:
+        case state::metadata:
+        case state::ruleset:
         case state::schema_type:
         case state::reference:
         case state::reference_name:
@@ -222,6 +256,8 @@ class post_subject_versions_request_handler
         case state::schema:
         case state::id:
         case state::version:
+        case state::metadata:
+        case state::ruleset:
         case state::schema_type:
         case state::references:
         case state::reference_name:
diff --git a/tests/rptest/tests/schema_registry_test.py b/tests/rptest/tests/schema_registry_test.py
index f6e446752f0ea..1ad0ff8f81972 100644
--- a/tests/rptest/tests/schema_registry_test.py
+++ b/tests/rptest/tests/schema_registry_test.py
@@ -995,6 +995,34 @@ def test_post_subjects_subject_versions_version_many(self):
             assert result_raw.status_code == requests.codes.ok
             assert result_raw.json()["id"] == 1
 
+    @cluster(num_nodes=3)
+    def test_post_subjects_subject_versions_metadata_ruleset(self):
+        """
+        Verify posting a schema with metatada and ruleSet
+        These are not supported, but if they're null, we let it pass.
+        """
+
+        topic = create_topic_names(1)[0]
+
+        self.logger.debug("Dump the schema with null metadata and ruleSet")
+        schema_1_data = json.dumps({
+            "schema": schema1_def,
+            "metadata": None,
+            "ruleSet": None
+        })
+
+        self.logger.debug("Posting schema as a subject key")
+        result_raw = self._post_subjects_subject_versions(
+            subject=f"{topic}-key", data=schema_1_data)
+        self.logger.debug(result_raw)
+        assert result_raw.status_code == requests.codes.ok
+
+        self.logger.debug("Retrieving schema")
+        result_raw = self._post_subjects_subject(subject=f"{topic}-key",
+                                                 data=schema_1_data)
+        self.logger.debug(result_raw)
+        assert result_raw.status_code == requests.codes.ok
+
     @cluster(num_nodes=3)
     def test_post_subjects_subject(self):
         """