schema_registry: Add error compatibility_not_found

If a request for config fails, the error 40408 should be returned
regardless of whether the subject exists.

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

@@ -421,14 +421,17 @@ class store {
     ///\brief Get the compatibility level for a subject, or fallback to global.
     result<compatibility_level>
     get_compatibility(const subject& sub, default_to_global fallback) const {
-        auto sub_it = BOOST_OUTCOME_TRYX(
-          get_subject_iter(sub, include_deleted::no));
+        auto sub_it_res = get_subject_iter(sub, include_deleted::no);
+        if (sub_it_res.has_error()) {
+            return compatibility_not_found(sub);
+        }
+        auto sub_it = std::move(sub_it_res).assume_value();
         if (fallback) {
             return sub_it->second.compatibility.value_or(_compatibility);
         } 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

@@ -752,7 +752,9 @@ def test_config(self):
         self.logger.debug("Get invalid subject config")
         result_raw = self._get_config_subject(subject="invalid_subject")
         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 'invalid_subject' does not have subject-level compatibility configured"
 
         schema_1_data = json.dumps({"schema": schema1_def})
 
@@ -765,7 +767,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",