Fix Avro bytes serialization

Author: rayokota

src/confluent_kafka/schema_registry/_async/avro.py

@@ -348,9 +348,16 @@ def field_transformer(rule_ctx, field_transform, msg): return (  # noqa: E731
             parsed_schema = self._parsed_schema
 
         with _ContextStringIO() as fo:
-            # write the record to the rest of the buffer
-            schemaless_writer(fo, parsed_schema, value)
-            buffer = fo.getvalue()
+            # Check if it's a simple bytes type
+            is_bytes = (parsed_schema == "bytes" or
+                        (isinstance(parsed_schema, dict) and parsed_schema.get("type") == "bytes"))
+            if is_bytes:
+                # For simple bytes type, write value directly
+                buffer = value if isinstance(value, bytes) else value.encode()
+            else:
+                # write the record to the rest of the buffer
+                schemaless_writer(fo, parsed_schema, value)
+                buffer = fo.getvalue()
 
             if latest_schema is not None:
                 buffer = self._execute_rules_with_phase(
@@ -585,17 +592,29 @@ async def __deserialize(
             reader_schema_raw = writer_schema_raw
             reader_schema = writer_schema
 
+        # Check if it's a simple bytes type
+        is_bytes = (writer_schema == "bytes" or
+                    (isinstance(writer_schema, dict) and writer_schema.get("type") == "bytes"))
+
         if migrations:
-            obj_dict = schemaless_reader(payload,
-                                         writer_schema,
-                                         None,
-                                         self._return_record_name)
+            if is_bytes:
+                # For simple bytes type, read payload directly
+                obj_dict = payload.read()
+            else:
+                obj_dict = schemaless_reader(payload,
+                                             writer_schema,
+                                             None,
+                                             self._return_record_name)
             obj_dict = self._execute_migrations(ctx, subject, migrations, obj_dict)
         else:
-            obj_dict = schemaless_reader(payload,
-                                         writer_schema,
-                                         reader_schema,
-                                         self._return_record_name)
+            if is_bytes:
+                # For simple bytes type, read payload directly
+                obj_dict = payload.read()
+            else:
+                obj_dict = schemaless_reader(payload,
+                                             writer_schema,
+                                             reader_schema,
+                                             self._return_record_name)
 
         def field_transformer(rule_ctx, field_transform, message): return (  # noqa: E731
             transform(rule_ctx, reader_schema, message, field_transform))

src/confluent_kafka/schema_registry/_sync/avro.py

@@ -348,9 +348,16 @@ def field_transformer(rule_ctx, field_transform, msg): return (  # noqa: E731
             parsed_schema = self._parsed_schema
 
         with _ContextStringIO() as fo:
-            # write the record to the rest of the buffer
-            schemaless_writer(fo, parsed_schema, value)
-            buffer = fo.getvalue()
+            # Check if it's a simple bytes type
+            is_bytes = (parsed_schema == "bytes" or
+                        (isinstance(parsed_schema, dict) and parsed_schema.get("type") == "bytes"))
+            if is_bytes:
+                # For simple bytes type, write value directly
+                buffer = value if isinstance(value, bytes) else value.encode()
+            else:
+                # write the record to the rest of the buffer
+                schemaless_writer(fo, parsed_schema, value)
+                buffer = fo.getvalue()
 
             if latest_schema is not None:
                 buffer = self._execute_rules_with_phase(
@@ -585,17 +592,29 @@ def __deserialize(
             reader_schema_raw = writer_schema_raw
             reader_schema = writer_schema
 
+        # Check if it's a simple bytes type
+        is_bytes = (writer_schema == "bytes" or
+                    (isinstance(writer_schema, dict) and writer_schema.get("type") == "bytes"))
+
         if migrations:
-            obj_dict = schemaless_reader(payload,
-                                         writer_schema,
-                                         None,
-                                         self._return_record_name)
+            if is_bytes:
+                # For simple bytes type, read payload directly
+                obj_dict = payload.read()
+            else:
+                obj_dict = schemaless_reader(payload,
+                                             writer_schema,
+                                             None,
+                                             self._return_record_name)
             obj_dict = self._execute_migrations(ctx, subject, migrations, obj_dict)
         else:
-            obj_dict = schemaless_reader(payload,
-                                         writer_schema,
-                                         reader_schema,
-                                         self._return_record_name)
+            if is_bytes:
+                # For simple bytes type, read payload directly
+                obj_dict = payload.read()
+            else:
+                obj_dict = schemaless_reader(payload,
+                                             writer_schema,
+                                             reader_schema,
+                                             self._return_record_name)
 
         def field_transformer(rule_ctx, field_transform, message): return (  # noqa: E731
             transform(rule_ctx, reader_schema, message, field_transform))

tests/schema_registry/_async/test_avro_serdes.py

@@ -198,6 +198,22 @@ async def test_avro_serialize_use_schema_id():
     assert obj == obj2
 
 
+async def test_avro_serialize_bytes():
+    conf = {'url': _BASE_URL}
+    client = AsyncSchemaRegistryClient.new_client(conf)
+    ser_conf = {'auto.register.schemas': True}
+    obj = b'\x02\x03\x04'
+    schema = 'bytes'
+    ser = await AsyncAvroSerializer(client, schema_str=json.dumps(schema), conf=ser_conf)
+    ser_ctx = SerializationContext(_TOPIC, MessageField.VALUE)
+    obj_bytes = await ser(obj, ser_ctx)
+    assert b'\x00\x00\x00\x00\x01\x02\x03\x04' == obj_bytes
+
+    deser = await AsyncAvroDeserializer(client)
+    obj2 = await deser(obj_bytes, ser_ctx)
+    assert obj == obj2
+
+
 async def test_avro_serialize_nested():
     conf = {'url': _BASE_URL}
     client = AsyncSchemaRegistryClient.new_client(conf)

tests/schema_registry/_sync/test_avro_serdes.py

@@ -198,6 +198,22 @@ def test_avro_serialize_use_schema_id():
     assert obj == obj2
 
 
+def test_avro_serialize_bytes():
+    conf = {'url': _BASE_URL}
+    client = SchemaRegistryClient.new_client(conf)
+    ser_conf = {'auto.register.schemas': True}
+    obj = b'\x02\x03\x04'
+    schema = 'bytes'
+    ser = AvroSerializer(client, schema_str=json.dumps(schema), conf=ser_conf)
+    ser_ctx = SerializationContext(_TOPIC, MessageField.VALUE)
+    obj_bytes = ser(obj, ser_ctx)
+    assert b'\x00\x00\x00\x00\x01\x02\x03\x04' == obj_bytes
+
+    deser = AvroDeserializer(client)
+    obj2 = deser(obj_bytes, ser_ctx)
+    assert obj == obj2
+
+
 def test_avro_serialize_nested():
     conf = {'url': _BASE_URL}
     client = SchemaRegistryClient.new_client(conf)