Merge pull request #12956 from BenPope/schema_registry_referenced_by_404

schema_registry: /referencedby should emit 404 if sub-ver doesn't exist

src/v/pandaproxy/schema_registry/sharded_store.cc

@@ -369,10 +369,18 @@ ss::future<bool> sharded_store::is_referenced(subject sub, schema_version ver) {
 ss::future<std::vector<schema_id>> sharded_store::referenced_by(
   subject sub, std::optional<schema_version> opt_ver) {
     schema_version ver;
+    // Ensure the subject exists
+    auto versions = co_await get_versions(sub, include_deleted::no);
     if (opt_ver.has_value()) {
         ver = *opt_ver;
+        auto version_not_found = std::none_of(
+          versions.begin(), versions.end(), [ver](auto const& v) {
+              return ver == v;
+          });
+        if (version_not_found) {
+            throw as_exception(not_found(sub, ver));
+        }
     } else {
-        auto versions = co_await get_versions(sub, include_deleted::no);
         vassert(
           !versions.empty(), "get_versions should not return empty versions");
         ver = versions.back();

tests/rptest/tests/schema_registry_test.py

@@ -1244,6 +1244,20 @@ def test_protobuf(self):
         assert result_raw.status_code == requests.codes.ok
         assert result_raw.json() == [2]
 
+        # invalid subject should error with 40401
+        result_raw = self._get_subjects_subject_versions_version_referenced_by(
+            "invalid_subject", 1)
+        self.logger.info(result_raw)
+        assert result_raw.status_code == requests.codes.not_found
+        assert result_raw.json()["error_code"] == 40401
+
+        # invalid version should error with 40402
+        result_raw = self._get_subjects_subject_versions_version_referenced_by(
+            "simple", 2)
+        self.logger.info(result_raw)
+        assert result_raw.status_code == requests.codes.not_found
+        assert result_raw.json()["error_code"] == 40402
+
     @cluster(num_nodes=4)
     @parametrize(protocol=SchemaType.AVRO, client_type=SerdeClientType.Python)
     @parametrize(protocol=SchemaType.AVRO, client_type=SerdeClientType.Java)