feat: use topic record name strategy (#172)

Update serializers to allow multiple event types per topic. See https://docs.openedx.org/projects/event-bus-kafka/en/latest/decisions/0011_multiple_event_types_per_topic.html for the full explanation of why.

CHANGELOG.rst

@@ -14,6 +14,12 @@ Change Log
 Unreleased
 **********
 
+[5.1.0] - 2023-05-17
+********************
+Changed
+=======
+* Reconfigured serializers to use topic_record_name_strategy, allowing multiple event types per topic
+
 [5.0.0] - 2023-05-17
 ********************
 Changed

edx_event_bus_kafka/__init__.py

@@ -9,4 +9,4 @@
 from edx_event_bus_kafka.internal.consumer import KafkaEventConsumer
 from edx_event_bus_kafka.internal.producer import KafkaEventProducer, create_producer
 
-__version__ = '5.0.0'
+__version__ = '5.1.0'

edx_event_bus_kafka/internal/producer.py

@@ -31,6 +31,7 @@
 try:
     import confluent_kafka
     from confluent_kafka import Producer
+    from confluent_kafka.schema_registry import topic_record_subject_name_strategy
     from confluent_kafka.schema_registry.avro import AvroSerializer
     from confluent_kafka.serialization import MessageField, SerializationContext
 except ImportError:  # pragma: no cover
@@ -165,11 +166,14 @@ def inner_to_dict(event_data, ctx=None):  # pylint: disable=unused-argument
         schema_str=extract_key_schema(signal_serializer, event_key_field),
         schema_registry_client=client,
         to_dict=inner_to_dict,
+        conf={'subject.name.strategy': topic_record_subject_name_strategy}
     )
+
     value_serializer = AvroSerializer(
         schema_str=signal_serializer.schema_string(),
         schema_registry_client=client,
         to_dict=inner_to_dict,
+        conf={'subject.name.strategy': topic_record_subject_name_strategy}
     )
 
     return key_serializer, value_serializer

edx_event_bus_kafka/internal/tests/test_producer.py

@@ -26,6 +26,7 @@
 
 # See https://github.com/openedx/event-bus-kafka/blob/main/docs/decisions/0005-optional-import-of-confluent-kafka.rst
 try:
+    from confluent_kafka.schema_registry import topic_record_subject_name_strategy
     from confluent_kafka.schema_registry.avro import AvroSerializer
 except ImportError:  # pragma: no cover
     pass
@@ -79,6 +80,10 @@ def test_serializers_configured(self):
             # We can't actually call them because they want to talk to the schema server.
             assert isinstance(key_ser, AvroSerializer)
             assert isinstance(value_ser, AvroSerializer)
+            assert key_ser._subject_name_func ==\
+                   topic_record_subject_name_strategy  # pylint: disable=protected-access,comparison-with-callable
+            assert value_ser._subject_name_func ==\
+                   topic_record_subject_name_strategy  # pylint: disable=protected-access,comparison-with-callable
 
     def test_serializers_unconfigured(self):
         with pytest.raises(Exception, match="missing library or settings"):