Merge pull request #676 from aiven/eliax1996/add-subjects-only-with-header-specified

Author: HelenMel

karapace/schema_registry_apis.py

@@ -472,6 +472,7 @@ async def schemas_get(
                 status=HTTPStatus.NOT_FOUND,
             )
 
+        include_subjects = request.query.get("includeSubjects", "false").lower() == "true"
         fetch_max_id = request.query.get("fetchMaxId", "false").lower() == "true"
         schema = self.schema_registry.schemas_get(parsed_schema_id, fetch_max_id=fetch_max_id)
 
@@ -505,9 +506,11 @@ def _has_subject_with_id() -> bool:
                 status=HTTPStatus.NOT_FOUND,
             )
 
-        subjects = self.schema_registry.database.subjects_for_schema(parsed_schema_id)
+        response_body = {"schema": schema.schema_str}
+
+        if include_subjects:
+            response_body["subjects"] = self.schema_registry.database.subjects_for_schema(parsed_schema_id)
 
-        response_body = {"schema": schema.schema_str, "subjects": subjects}
         if schema.schema_type is not SchemaType.AVRO:
             response_body["schemaType"] = schema.schema_type
         if schema.references:

karapace/serialization.py

@@ -115,7 +115,7 @@ async def get_latest_schema(self, subject: str) -> Tuple[SchemaId, ParsedTypedSc
             raise SchemaRetrievalError(f"Failed to parse schema string from response: {json_result}") from e
 
     async def get_schema_for_id(self, schema_id: SchemaId) -> Tuple[TypedSchema, List[Subject]]:
-        result = await self.client.get(f"schemas/ids/{schema_id}")
+        result = await self.client.get(f"schemas/ids/{schema_id}", params={"includeSubjects": "True"})
         if not result.ok:
             raise SchemaRetrievalError(result.json()["message"])
         json_result = result.json()

tests/integration/test_schema.py

@@ -962,7 +962,7 @@ async def test_union_comparing_to_other_types(registry_async_client: Client) ->
         res = await registry_async_client.post(f"subjects/{subject}/versions", json={"schema": json.dumps(plain_schema)})
         assert res.status_code == status_code
 
-        res = await registry_async_client.get(f"/schemas/ids/{initial_schema_id}")
+        res = await registry_async_client.get(f"/schemas/ids/{initial_schema_id}", params={"includeSubjects": "True"})
         assert subject in res.json()["subjects"]
 
     expected_results = [("BACKWARD", 200), ("FORWARD", 409), ("FULL", 409)]
@@ -2352,9 +2352,16 @@ async def test_malformed_kafka_message(
         sleep=1,
     )
     res_data = res.json()
-    expected_payload = {"schema": json_encode({"foo": "bar"}, compact=True), "subjects": ["foo"]}
+    expected_payload = {"schema": json_encode({"foo": "bar"}, compact=True)}
     assert res_data == expected_payload, res_data
 
+    with_subjects_reply = await registry_async_client.get(path, params={"includeSubjects": "True"})
+    assert with_subjects_reply.ok, "a subsequent request once the server is up should be valid"
+    json_reply = with_subjects_reply.json()
+    assert "subjects" in json_reply, "subjects should be present if specified"
+    expected_payload["subjects"] = ["foo"]
+    assert expected_payload == json_reply, "the reply should be equal as the previous one with additional subjects field"
+
 
 async def test_inner_type_compat_failure(registry_async_client: Client) -> None:
     subject = create_subject_name_factory("test_inner_type_compat_failure")()