kafka_actions: fall back to string when schema registry magic byte is absent (DataDog#23951)

* kafka_actions: fall back to string when schema registry magic byte is absent

When uses_schema_registry=True but the first byte of a message is not 0x00,
the deserializer now decodes the message as a plain UTF-8 string instead of
raising an error. This handles producers that mix framed and unframed messages
on the same topic.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

* kafka_actions: add changelog entry for PR DataDog#23951

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

* kafka_actions: simplify SR fallback — debug log, no raise, cleaner condition

- Replace warning with debug log
- Single condition: len >= 5 && magic byte == 0x00 → SR path, else → string fallback
- No ValueError raised for any non-SR message

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

* kafka_actions: remove msg_hex from debug log

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

* kafka_actions: fix lint errors (E501, I001)

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

* kafka_actions: fix remaining I001 import sort errors using root pyproject.toml

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

---------

Co-authored-by: Claude Sonnet 4.6 <noreply@anthropic.com>

Author: piochelepiotr

kafka_actions/changelog.d/23951.fixed

@@ -0,0 +1 @@
+Fall back to string deserialization when schema registry magic byte is absent.

kafka_actions/datadog_checks/kafka_actions/message_deserializer.py

@@ -243,20 +243,21 @@ def _deserialize_bytes_maybe_schema_registry(
     ) -> tuple[str | None, int | None]:
         """Deserialize message, handling Schema Registry format if present."""
         if uses_schema_registry:
-            if len(message) < 5 or message[0] != SCHEMA_REGISTRY_MAGIC_BYTE:
-                msg_hex = message[:5].hex() if len(message) >= 5 else message.hex()
-                raise ValueError(
-                    f"Expected schema registry format (magic byte 0x00 + 4-byte schema ID), "
-                    f"but message is too short or has wrong magic byte: {msg_hex}"
-                )
-            schema_id = int.from_bytes(message[1:5], 'big')
-            message = message[5:]  # Skip the magic byte and schema ID bytes
+            if len(message) >= 5 and message[0] == SCHEMA_REGISTRY_MAGIC_BYTE:
+                schema_id = int.from_bytes(message[1:5], 'big')
+                message = message[5:]  # Skip the magic byte and schema ID bytes
 
-            actual_format = message_format
-            if self.schema_registry is not None:
-                schema, actual_format = self._fetch_and_build_schema(schema_id, message_format)
+                actual_format = message_format
+                if self.schema_registry is not None:
+                    schema, actual_format = self._fetch_and_build_schema(schema_id, message_format)
 
-            return self._deserialize_bytes(message, actual_format, schema, uses_schema_registry=True), schema_id
+                return self._deserialize_bytes(message, actual_format, schema, uses_schema_registry=True), schema_id
+            else:
+                self.log.debug(
+                    "Expected schema registry format (magic byte 0x00 + 4-byte schema ID), "
+                    "but message is too short or has wrong magic byte, falling back to string",
+                )
+                return self._deserialize_bytes(message, 'string', None, uses_schema_registry=False), None
         else:
             # Fallback behavior: try without schema registry format first, then with it
             try:

kafka_actions/tests/test_message_deserializer.py

@@ -270,9 +270,11 @@ def test_avro_explicit_schema_registry_configuration(self):
         assert result[1] is None, "Should have no schema ID"
         assert 'The Go Programming Language' in result[0]
 
-        # Test 2: uses_schema_registry=True with plain Avro message - should fail (missing magic byte)
+        # Test 2: uses_schema_registry=True with plain Avro message - falls back to string (non-UTF-8 bytes → error)
         result = deserializer.deserialize_message(avro_message_no_sr, 'avro', avro_schema, True)
-        assert result[0].startswith("<deserialization error:"), "Should fail when uses_schema_registry=True"
+        assert result[0].startswith("<deserialization error:"), (
+            "Non-UTF-8 avro bytes produce error after string fallback"
+        )
         assert result[1] is None
 
         # Test 3: uses_schema_registry=True with Schema Registry format - should succeed
@@ -281,17 +283,18 @@ def test_avro_explicit_schema_registry_configuration(self):
         assert result[1] == 350, "Should extract schema ID 350"
         assert 'The Go Programming Language' in result[0]
 
-        # Test 4: Wrong magic byte - should fail
+        # Test 4: Wrong magic byte - falls back to string (non-UTF-8 bytes → error)
         wrong_magic_byte = (
             b'\x01\x00\x00\x01\x5e\xd0\xf5\xe4\xd6\xa3\xb9\x046The Go Programming Language\x18Alan Donovan'
         )
         result = deserializer.deserialize_message(wrong_magic_byte, 'avro', avro_schema, True)
-        assert result[0].startswith("<deserialization error:"), "Should fail with wrong magic byte"
+        assert result[0].startswith("<deserialization error:"), "Non-UTF-8 bytes produce error after string fallback"
 
-        # Test 5: Message too short (less than 5 bytes) - should fail
+        # Test 5: Magic byte 0x00 present but message too short (< 5 bytes) - falls back to string
         too_short = b'\x00\x00\x01'
         result = deserializer.deserialize_message(too_short, 'avro', avro_schema, True)
-        assert result[0].startswith("<deserialization error:"), "Should fail when message too short"
+        assert result[0] is not None, "Too-short message falls back to string, not an error"
+        assert result[1] is None
 
         # Test 6: Test through DeserializedMessage wrapper
         kafka_msg = MockKafkaMessage(key=key_bytes, value=avro_message_no_sr)
@@ -406,10 +409,10 @@ def test_protobuf_explicit_schema_registry_configuration(self):
         assert result[1] is None, "Should have no schema ID"
         assert 'The Go Programming Language' in result[0]
 
-        # Test 2: uses_schema_registry=True with plain Protobuf message - should fail
+        # Test 2: uses_schema_registry=True with plain Protobuf message - falls back to string (non-UTF-8 → error)
         result = deserializer.deserialize_message(protobuf_message_no_sr, 'protobuf', protobuf_schema, True)
         assert result[0].startswith("<deserialization error:"), (
-            "Protobuf should fail when uses_schema_registry=True but no SR format"
+            "Non-UTF-8 protobuf bytes produce error after string fallback"
         )
         assert result[1] is None
 
@@ -419,15 +422,16 @@ def test_protobuf_explicit_schema_registry_configuration(self):
         assert result[1] == 350, "Should extract schema ID 350"
         assert 'The Go Programming Language' in result[0]
 
-        # Test 4: Wrong magic byte - should fail
+        # Test 4: Wrong magic byte - falls back to string (non-UTF-8 bytes → error)
         wrong_magic_byte = b'\x01\x00\x00\x01\x5e' + protobuf_message_no_sr
         result = deserializer.deserialize_message(wrong_magic_byte, 'protobuf', protobuf_schema, True)
-        assert result[0].startswith("<deserialization error:"), "Should fail with wrong magic byte"
+        assert result[0].startswith("<deserialization error:"), "Non-UTF-8 bytes produce error after string fallback"
 
-        # Test 5: Message too short (less than 5 bytes) - should fail
+        # Test 5: Magic byte 0x00 present but message too short (< 5 bytes) - falls back to string
         too_short = b'\x00\x00\x01'
         result = deserializer.deserialize_message(too_short, 'protobuf', protobuf_schema, True)
-        assert result[0].startswith("<deserialization error:"), "Should fail when message too short"
+        assert result[0] is not None, "Too-short message falls back to string, not an error"
+        assert result[1] is None
 
         # Test 6: Test through DeserializedMessage wrapper
         kafka_msg = MockKafkaMessage(key=key_bytes, value=protobuf_message_no_sr)
@@ -655,6 +659,31 @@ def test_value_skip_bytes_via_deserialized_message(self):
         assert msg.value == {'price': 99.95}
         assert msg.key == 'order-42'
 
+    def test_schema_registry_wrong_magic_byte_falls_back_to_string(self):
+        """When uses_schema_registry=True but first byte is not 0x00, fall back to string decoding."""
+        log = MagicMock()
+        deserializer = MessageDeserializer(log)
+
+        # A plain UTF-8 string message — no schema registry framing.
+        plain_text = b'hello-world'
+        result, schema_id = deserializer.deserialize_message(plain_text, 'avro', uses_schema_registry=True)
+
+        assert result == json.dumps('hello-world')
+        assert schema_id is None
+        log.debug.assert_called()
+
+    def test_schema_registry_wrong_magic_byte_non_utf8_returns_error(self):
+        """When fallback-to-string is triggered but bytes are not valid UTF-8, return error string."""
+        log = MagicMock()
+        deserializer = MessageDeserializer(log)
+
+        # Bytes that are not valid UTF-8 and don't start with 0x00.
+        invalid_utf8 = b'\xff\xfe\xfd'
+        result, schema_id = deserializer.deserialize_message(invalid_utf8, 'json', uses_schema_registry=True)
+
+        assert result.startswith("<deserialization error:")
+        assert schema_id is None
+
     def test_skip_bytes_runs_before_schema_registry_detection(self):
         """skip_bytes is applied before the Confluent SR magic-byte check.