schema_registry: Add error compatibility_not_found

Signed-off-by: Ben Pope <[email protected]>

src/v/pandaproxy/error.cc

@@ -127,6 +127,8 @@ struct reply_error_category final : std::error_category {
             return "subject_version_soft_deleted";
         case reply_error_code::subject_version_not_deleted:
             return "subject_version_not_deleted";
+        case reply_error_code::compatibility_not_found:
+            return "compatibility_not_found";
         case reply_error_code::serialization_error:
             return "serialization_error";
         case reply_error_code::consumer_already_exists:

src/v/pandaproxy/error.h

@@ -78,6 +78,7 @@ enum class reply_error_code : uint16_t {
     subject_not_deleted = 40405,
     subject_version_soft_deleted = 40406,
     subject_version_not_deleted = 40407,
+    compatibility_not_found = 40408,
     serialization_error = 40801,
     consumer_already_exists = 40902,
     schema_empty = 42201,

src/v/pandaproxy/schema_registry/error.cc

@@ -50,6 +50,9 @@ struct error_category final : std::error_category {
                    "permanently";
         case error_code::subject_version_not_deleted:
             return "Version not deleted before being permanently deleted";
+        case error_code::compatibility_not_found:
+            return "Subject does not have subject-level compatibility "
+                   "configured";
         case error_code::subject_version_operaton_not_permitted:
             return "Overwrite new schema is not permitted.";
         case error_code::subject_version_has_references:
@@ -88,6 +91,8 @@ struct error_category final : std::error_category {
             return reply_error_code::subject_version_soft_deleted; // 40406
         case error_code::subject_version_not_deleted:
             return reply_error_code::subject_version_not_deleted; // 40407
+        case error_code::compatibility_not_found:
+            return reply_error_code::compatibility_not_found; // 40408
         case error_code::subject_schema_invalid:
             return reply_error_code::internal_server_error; // 500
         case error_code::write_collision:

src/v/pandaproxy/schema_registry/error.h

@@ -28,6 +28,7 @@ enum class error_code {
     subject_not_deleted,
     subject_version_soft_deleted,
     subject_version_not_deleted,
+    compatibility_not_found,
     subject_version_operaton_not_permitted,
     subject_version_has_references,
     subject_version_schema_id_already_exists,

src/v/pandaproxy/schema_registry/errors.h

@@ -150,4 +150,12 @@ inline error_info has_references(const subject& sub, schema_version ver) {
         ver())};
 }
 
+inline error_info compatibility_not_found(const subject& sub) {
+    return error_info{
+      error_code::compatibility_not_found,
+      fmt::format(
+        "Subject '{}' does not have subject-level compatibility configured",
+        sub())};
+}
+
 } // namespace pandaproxy::schema_registry

src/v/pandaproxy/schema_registry/store.h

@@ -428,7 +428,7 @@ class store {
         } else if (sub_it->second.compatibility) {
             return sub_it->second.compatibility.value();
         }
-        return not_found(sub);
+        return compatibility_not_found(sub);
     }
 
     ///\brief Set the global compatibility level.

tests/rptest/tests/schema_registry_test.py

@@ -765,7 +765,9 @@ def test_config(self):
         self.logger.debug("Get subject config - should fail")
         result_raw = self._get_config_subject(subject=f"{topic}-key")
         assert result_raw.status_code == requests.codes.not_found
-        assert result_raw.json()["error_code"] == 40401
+        assert result_raw.json()["error_code"] == 40408
+        assert result_raw.json(
+        )["message"] == f"Subject '{topic}-key' does not have subject-level compatibility configured"
 
         self.logger.debug("Get subject config - fallback to global")
         result_raw = self._get_config_subject(subject=f"{topic}-key",