Only write schema cache once (#1057)

* Only write to the writer schema cache once.

* Add test for the _get_encoder_func caching.

* Patch against self.ms, change iterators to list for reuse across test cases

Co-authored-by: Dennis Taylor <[email protected]>
Co-authored-by: Braden Staudacher <[email protected]>

Author: 3 people

src/confluent_kafka/avro/serializer/message_serializer.py

@@ -113,7 +113,8 @@ def encode_record_with_schema(self, topic, schema, record, is_key=False):
             raise serialize_err(message)
 
         # cache writer
-        self.id_to_writers[schema_id] = self._get_encoder_func(schema)
+        if schema_id not in self.id_to_writers:
+            self.id_to_writers[schema_id] = self._get_encoder_func(schema)
 
         return self.encode_record_with_schema_id(schema_id, record, is_key=is_key)
 

tests/avro/data_gen.py

@@ -54,7 +54,7 @@ def create_basic_item(i):
     }
 
 
-BASIC_ITEMS = map(create_basic_item, range(1, 20))
+BASIC_ITEMS = list(map(create_basic_item, range(1, 20)))
 
 ADVANCED_SCHEMA = load_schema_file(os.path.join(avsc_dir, 'adv_schema.avsc'))
 
@@ -68,7 +68,7 @@ def create_adv_item(i):
     return basic
 
 
-ADVANCED_ITEMS = map(create_adv_item, range(1, 20))
+ADVANCED_ITEMS = list(map(create_adv_item, range(1, 20)))
 
 
 def _write_items(base_name, schema_str, items):

tests/avro/test_message_serializer.py

@@ -23,6 +23,7 @@
 import struct
 
 import unittest
+from unittest.mock import patch
 
 from tests.avro import data_gen
 from confluent_kafka.avro.serializer.message_serializer import MessageSerializer
@@ -75,8 +76,19 @@ def test_encode_record_with_schema(self):
             message = self.ms.encode_record_with_schema(topic, basic, record)
             self.assertMessageIsSame(message, record, schema_id)
 
+    def test_encode_record_with_schema_sets_writers_cache_once(self):
+        topic = 'test'
+        basic = avro.loads(data_gen.BASIC_SCHEMA)
+        subject = 'test-value'
+        self.client.register(subject, basic)
+        records = data_gen.BASIC_ITEMS
+        with patch.object(self.ms, "_get_encoder_func") as encoder_func_mock:
+            for record in records:
+                self.ms.encode_record_with_schema(topic, basic, record)
+        encoder_func_mock.assert_called_once_with(basic)
+
     def test_decode_none(self):
-        """"null/None messages should decode to None"""
+        """null/None messages should decode to None"""
 
         self.assertIsNone(self.ms.decode_message(None))