[kafka_consumer] URL-encode subject names in Schema Registry API calls (DataDog#22924)

* [kafka_consumer] URL-encode subject names in Schema Registry API calls

Subjects containing slashes (e.g. Protobuf references like
`google/protobuf/timestamp.proto`) were causing 404 errors because
the slashes were interpreted as URL path separators instead of being
encoded as `%2F`.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

* [kafka_consumer] Add changelog entry for URL-encoding fix

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: piochelepiotr

kafka_consumer/changelog.d/22924.fixed

@@ -0,0 +1 @@
+URL-encode subject names in Schema Registry API calls to fix 404 errors for Protobuf reference subjects containing slashes.

kafka_consumer/datadog_checks/kafka_consumer/cluster_metadata.py

@@ -10,6 +10,7 @@
 import time
 from concurrent.futures import ThreadPoolExecutor, as_completed
 from typing import TypedDict
+from urllib.parse import quote
 
 from confluent_kafka.admin import ConfigResource, ResourceType
 
@@ -163,13 +164,15 @@ def _get_schema_registry_subjects(self):
     def _get_schema_registry_versions(self, subject: str) -> list[int]:
         """Fetch the list of version numbers for a subject (lightweight call)."""
         base_url = self.config._collect_schema_registry
-        response = self.http.get(f"{base_url}/subjects/{subject}/versions")
+        encoded_subject = quote(subject, safe='')
+        response = self.http.get(f"{base_url}/subjects/{encoded_subject}/versions")
         response.raise_for_status()
         return response.json()
 
     def _get_schema_registry_latest_version(self, subject):
         base_url = self.config._collect_schema_registry
-        response = self.http.get(f"{base_url}/subjects/{subject}/versions/latest")
+        encoded_subject = quote(subject, safe='')
+        response = self.http.get(f"{base_url}/subjects/{encoded_subject}/versions/latest")
         response.raise_for_status()
         return response.json()
 

kafka_consumer/tests/test_cluster_metadata.py

@@ -1324,3 +1324,30 @@ def test_cluster_metadata_with_cluster_id_override(check, dd_run_check, aggregat
         if 'kafka_cluster_id' in payload:
             assert payload['kafka_cluster_id'] == 'my-override-id'
             assert payload['original_kafka_cluster_id'] == 'auto-detected-id'
+
+
+def test_schema_registry_url_encodes_subject_names(check):
+    """Subjects with slashes (e.g. Protobuf references) must be URL-encoded in API calls."""
+    instance = {
+        'kafka_connect_str': 'localhost:9092',
+        'schema_registry_url': 'http://localhost:8081',
+    }
+    kafka_check = check(instance)
+    collector = kafka_check.metadata_collector
+
+    mock_response = mock.MagicMock()
+    mock_response.json.return_value = [1]
+    mock_response.raise_for_status.return_value = None
+    collector.http = mock.MagicMock()
+    collector.http.get.return_value = mock_response
+
+    subject = 'google/protobuf/timestamp.proto'
+
+    collector._get_schema_registry_versions(subject)
+    collector.http.get.assert_called_with('http://localhost:8081/subjects/google%2Fprotobuf%2Ftimestamp.proto/versions')
+
+    collector.http.get.reset_mock()
+    collector._get_schema_registry_latest_version(subject)
+    collector.http.get.assert_called_with(
+        'http://localhost:8081/subjects/google%2Fprotobuf%2Ftimestamp.proto/versions/latest'
+    )