PR Feedback, round 1

* Remove plugin registration and lookup
* Remove interface in favor of callback

Author: delphyne

confluent_kafka/avro/__init__.py

@@ -2,8 +2,6 @@
     Avro schema registry module: Deals with encoding and decoding of messages with avro schemas
 
 """
-import pkg_resources
-
 from confluent_kafka import Producer, Consumer
 from confluent_kafka.avro.error import ClientError
 from confluent_kafka.avro.load import load, loads  # noqa
@@ -12,6 +10,7 @@
                                              KeySerializerError,
                                              ValueSerializerError)
 from confluent_kafka.avro.serializer.message_serializer import MessageSerializer
+from confluent_kafka.avro.serializer.name_strategies import topic_name_strategy
 
 
 class AvroProducer(Producer):
@@ -28,7 +27,8 @@ class AvroProducer(Producer):
     """
 
     def __init__(self, config, default_key_schema=None,
-                 default_value_schema=None, schema_registry=None):
+                 default_value_schema=None, schema_registry=None,
+                 value_subject_name_strategy=topic_name_strategy, key_subject_name_strategy=topic_name_strategy):
         schema_registry_url = config.pop("schema.registry.url", None)
         if schema_registry is None:
             if schema_registry_url is None:
@@ -38,27 +38,13 @@ def __init__(self, config, default_key_schema=None,
             raise ValueError("Cannot pass schema_registry along with schema.registry.url config")
 
         self._serializer = MessageSerializer(schema_registry,
-                                             self._load_name_strategy(config, "value.subject.name.strategy"),
-                                             self._load_name_strategy(config, "key.subject.name.strategy"))
+                                             key_subject_name_strategy=key_subject_name_strategy,
+                                             value_subject_name_strategy=value_subject_name_strategy)
 
         super(AvroProducer, self).__init__(config)
         self._key_schema = default_key_schema
         self._value_schema = default_value_schema
 
-    @staticmethod
-    def _load_name_strategy(config, key):
-        strategy_name = config.pop(key, "TopicName")
-        subject_name_strategy_plugins = {
-            entry_point.name: entry_point
-            for entry_point
-            in pkg_resources.iter_entry_points('subject.name.strategy')
-        }
-        subject_name_strategy = subject_name_strategy_plugins.get(strategy_name, None)
-        if subject_name_strategy is None:
-            raise ValueError("The subject name strategy plugin named {} could not be found.".format(
-                strategy_name))
-        return subject_name_strategy.load()()
-
     def produce(self, **kwargs):
         """
             Asynchronously sends message to Kafka by encoding with specified or default avro schema.

confluent_kafka/avro/serializer/message_serializer.py

@@ -32,7 +32,7 @@
 from confluent_kafka.avro.serializer import (SerializerError,
                                              KeySerializerError,
                                              ValueSerializerError)
-from confluent_kafka.avro.serializer.name_strategies import TopicNameStrategy
+from confluent_kafka.avro.serializer.name_strategies import topic_name_strategy
 
 log = logging.getLogger(__name__)
 
@@ -70,7 +70,7 @@ class MessageSerializer(object):
     """
 
     def __init__(self, registry_client,
-                 key_subject_name_strategy=TopicNameStrategy(), value_subject_name_strategy=TopicNameStrategy()):
+                 key_subject_name_strategy=topic_name_strategy, value_subject_name_strategy=topic_name_strategy):
         self.registry_client = registry_client
         self.key_subject_name_strategy = key_subject_name_strategy
         self.value_subject_name_strategy = value_subject_name_strategy
@@ -96,7 +96,7 @@ def encode_record_with_schema(self, topic, schema, record, is_key=False):
         serialize_err = KeySerializerError if is_key else ValueSerializerError
 
         name_strategy = self.key_subject_name_strategy if is_key else self.value_subject_name_strategy
-        subject = name_strategy.get_subject_name(topic, is_key, schema)
+        subject = name_strategy(topic, is_key, schema)
         # register it
         schema_id = self.registry_client.register(subject, schema)
         if not schema_id:

confluent_kafka/avro/serializer/name_strategies.py

@@ -20,52 +20,32 @@
 the strategies provided via the Java client (https://github.com/confluentinc/schema-registry).
 
 The built in strategies include:
-``TopicName``
+``topic_name_strategy``
   The topic name is used along with either "key" or "value".
 
-``RecordName``
+``record_name_strategy``
   The fully-qualified name of the avro schema is used.
 
-``TopicRecordName``
+``topic_record_name_strategy``
   The topic name is used along with the fully-qualified name of the avro schema.
 
-Additional strategies may be provided at run time by registering a class with the ``subject.name.strategy`` entry
-point.
+Additional strategies may be provided by passing a callable accepting topic, is_key, and schema as arguments.
 
 """
 
 
-class SubjectNameStrategy(object):
-    """
-    A {@link SubjectNameStrategy} is used by the Avro serializer to determine
-    the subject name under which the event record schemas should be registered
-    in the schema registry. The default is {@link TopicNameStrategy}.
-    """
-    def get_subject_name(self, topic, is_key, schema):
-        """
-        For a given topic and message, returns the subject name under which the
-        schema should be registered in the schema registry.
-        :param topic: The Kafka topic name to which the message is being published.
-        :param is_key: True when encoding a message key, false for a message value.
-        :param schema: The value to be published in the message.
-        :return: The subject name under which the schema should be registered.
-        """
-        raise NotImplemented("SubjectNameStrategy implementations must implement get_subject_name.")
-
-
-class TopicNameStrategy(SubjectNameStrategy):
+def topic_name_strategy(topic, is_key, schema):
     """
     Default {@link SubjectNameStrategy}: for any messages published to
     `topic`, the schema of the message key is registered under
     the subject name `topic`-key, and the message value is registered
     under the subject name `topic`-value.
     """
-    def get_subject_name(self, topic, is_key, schema):
-        suffix = "-key" if is_key else "-value"
-        return topic + suffix
+    suffix = "-key" if is_key else "-value"
+    return topic + suffix
 
 
-class RecordNameStrategy(SubjectNameStrategy):
+def record_name_strategy(topic, is_key, schema):
     """
     For any Avro record type that is published to Kafka, registers the schema
     in the registry under the fully-qualified record name (regardless of the
@@ -74,11 +54,10 @@ class RecordNameStrategy(SubjectNameStrategy):
     Instead, checks compatibility of any occurrences of the same record name
     across `all` topics.
     """
-    def get_subject_name(self, topic, is_key, schema):
-        return schema.fullname
+    return schema.fullname
 
 
-class TopicRecordNameStrategy(RecordNameStrategy):
+def topic_record_name_strategy(topic, is_key, schema):
     """
     For any Avro record type that is published to Kafka topic `topic`,
     registers the schema in the registry under the subject name
@@ -89,5 +68,4 @@ class TopicRecordNameStrategy(RecordNameStrategy):
     incompatible versions of the same record name, since the compatibility
     check is scoped to a particular record name within a particular topic.
     """
-    def get_subject_name(self, topic, is_key, schema):
-        return topic + "-" + RecordNameStrategy.get_subject_name(self, topic, is_key, schema)
+    return topic + "-" + record_name_strategy(topic, is_key, schema)

setup.py

@@ -23,8 +23,12 @@
                             'confluent_kafka/src/Admin.c'])
 
 
+install_requirements_path = os.environ.get('INSTALL_REQUIREMENTS_PATH_OVERRIDE',
+                                           os.path.dirname(__file__))
+
+
 def get_install_requirements(path):
-    content = open(os.path.join(os.path.dirname(__file__), path)).read()
+    content = open(os.path.join(install_requirements_path, path)).read()
     return [
         req
         for req in content.split("\n")
@@ -45,11 +49,4 @@ def get_install_requirements(path):
       extras_require={
           'avro': ['fastavro', 'requests', avro],
           'dev': get_install_requirements("test-requirements.txt")
-      },
-      entry_points={
-          'subject.name.strategy': [
-              'TopicName = confluent_kafka.avro.serializer.name_strategies:TopicNameStrategy [avro]',
-              'RecordName = confluent_kafka.avro.serializer.name_strategies:RecordNameStrategy [avro]',
-              'TopicRecordName = confluent_kafka.avro.serializer.name_strategies:TopicRecordNameStrategy [avro]'
-          ]
       })

tests/avro/test_subject_name_strategies.py

@@ -26,77 +26,69 @@
 
 class TestMessageSerializer(unittest.TestCase):
     def test_topic_name(self):
-        strategy = strategies.TopicNameStrategy()
+        strategy = strategies.topic_name_strategy
         topic = "some_legal_topic_name"
         expected = topic + "-key"
-        actual = strategy.get_subject_name(topic, True, None)
+        actual = strategy(topic, True, None)
         assert expected == actual
         expected = topic + "-value"
-        actual = strategy.get_subject_name(topic, False, None)
+        actual = strategy(topic, False, None)
         assert expected == actual
 
     def test_record_name(self):
-        strategy = strategies.RecordNameStrategy()
+        strategy = strategies.record_name_strategy
         schema = RecordSchema("MyRecordType", "my.namespace", fields=[], names=Names())
         expected = "my.namespace.MyRecordType"
-        actual = strategy.get_subject_name(None, None, schema)
+        actual = strategy(None, None, schema)
         assert expected == actual
 
     def test_topic_record_name(self):
-        strategy = strategies.TopicRecordNameStrategy()
+        strategy = strategies.topic_record_name_strategy
         topic = "some_legal_topic_name"
         schema = RecordSchema("MyRecordType", "my.namespace", fields=[], names=Names())
         expected = "some_legal_topic_name-my.namespace.MyRecordType"
-        actual = strategy.get_subject_name(topic, None, schema)
+        actual = strategy(topic, None, schema)
         assert expected == actual
 
     def test_default_subject_name_strategy(self):
         schema_registry = MockSchemaRegistryClient()
         producer = AvroProducer(config={}, schema_registry=schema_registry)
         serializer = producer._serializer
-        assert isinstance(serializer.key_subject_name_strategy, strategies.TopicNameStrategy)
-        assert isinstance(serializer.value_subject_name_strategy, strategies.TopicNameStrategy)
+        assert serializer.key_subject_name_strategy is strategies.topic_name_strategy
+        assert serializer.value_subject_name_strategy is strategies.topic_name_strategy
 
     def test_explicit_topic_subject_name_strategy(self):
         schema_registry = MockSchemaRegistryClient()
-        config = {
-            'value.subject.name.strategy': 'TopicName',
-            'key.subject.name.strategy': 'TopicName'
-        }
-        producer = AvroProducer(config=config, schema_registry=schema_registry)
+        producer = AvroProducer(config={}, schema_registry=schema_registry,
+                                key_subject_name_strategy=strategies.topic_name_strategy,
+                                value_subject_name_strategy=strategies.topic_name_strategy)
         serializer = producer._serializer
-        assert isinstance(serializer.key_subject_name_strategy, strategies.TopicNameStrategy)
-        assert isinstance(serializer.value_subject_name_strategy, strategies.TopicNameStrategy)
+        assert serializer.key_subject_name_strategy is strategies.topic_name_strategy
+        assert serializer.value_subject_name_strategy is strategies.topic_name_strategy
 
     def test_explicit_record_subject_name_strategy(self):
         schema_registry = MockSchemaRegistryClient()
-        config = {
-            'value.subject.name.strategy': 'RecordName',
-            'key.subject.name.strategy': 'RecordName'
-        }
-        producer = AvroProducer(config=config, schema_registry=schema_registry)
+        producer = AvroProducer(config={}, schema_registry=schema_registry,
+                                key_subject_name_strategy=strategies.record_name_strategy,
+                                value_subject_name_strategy=strategies.record_name_strategy)
         serializer = producer._serializer
-        assert isinstance(serializer.key_subject_name_strategy, strategies.RecordNameStrategy)
-        assert isinstance(serializer.value_subject_name_strategy, strategies.RecordNameStrategy)
+        assert serializer.key_subject_name_strategy is strategies.record_name_strategy
+        assert serializer.value_subject_name_strategy is strategies.record_name_strategy
 
     def test_explicit_topic_record_subject_name_strategy(self):
         schema_registry = MockSchemaRegistryClient()
-        config = {
-            'value.subject.name.strategy': 'TopicRecordName',
-            'key.subject.name.strategy': 'TopicRecordName'
-        }
-        producer = AvroProducer(config=config, schema_registry=schema_registry)
+        producer = AvroProducer(config={}, schema_registry=schema_registry,
+                                key_subject_name_strategy=strategies.topic_record_name_strategy,
+                                value_subject_name_strategy=strategies.topic_record_name_strategy)
         serializer = producer._serializer
-        assert isinstance(serializer.key_subject_name_strategy, strategies.TopicRecordNameStrategy)
-        assert isinstance(serializer.value_subject_name_strategy, strategies.TopicRecordNameStrategy)
+        assert serializer.key_subject_name_strategy is strategies.topic_record_name_strategy
+        assert serializer.value_subject_name_strategy is strategies.topic_record_name_strategy
 
     def test_differing_key_and_value_subject_name_strategies(self):
         schema_registry = MockSchemaRegistryClient()
-        config = {
-            'value.subject.name.strategy': 'RecordName',
-            'key.subject.name.strategy': 'TopicRecordName'
-        }
-        producer = AvroProducer(config=config, schema_registry=schema_registry)
+        producer = AvroProducer(config={}, schema_registry=schema_registry,
+                                key_subject_name_strategy=strategies.record_name_strategy,
+                                value_subject_name_strategy=strategies.topic_record_name_strategy)
         serializer = producer._serializer
-        assert isinstance(serializer.key_subject_name_strategy, strategies.RecordNameStrategy)
-        assert isinstance(serializer.value_subject_name_strategy, strategies.TopicRecordNameStrategy)
+        assert serializer.key_subject_name_strategy is strategies.record_name_strategy
+        assert serializer.value_subject_name_strategy is strategies.topic_record_name_strategy