feat: remove confluent-kafka as a hard dependency

docs/decisions/0005-optional-import-of-confluent-kafka.rst

@@ -0,0 +1,60 @@
+4. Optional import of confluent-kafka
+#####################################
+
+Status
+******
+
+Accepted
+
+Context
+*******
+
+* `confluent-kafka`_ is a library written and maintained by Confluent, our managed instance provider (see :doc:`0004-kafka-managed-hosting`). The library abstracts out the work for sending and receiving events to and from the Kafka cluster and converting them into message objects.
+* confluent-kafka in turn is a wrapper around a C library called `librdkafka`_ (distributed as `librdkafka_dev`)
+* librdkafka-dev does not currently have a compiled binary available for Linux/aarch64
+
+As a result of the points above, if a package includes a dependency on confluent-kafka, developers will not be able to install the package in Linux/aarch64 environments.
+
+.. _confluent-kafka: https://github.com/confluentinc/confluent-kafka-python
+.. _librdkafka: https://github.com/edenhill/librdkafka
+
+Decision
+********
+
+Instead of requiring confluent-kafka directly in base.in, we will wrap all imports of `confluent_kafka`in a `try...catch` block. If the import fails, the library will log an informative message and any calls will fail gracefully.
+
+For example::
+
+    try:
+        import confluent_kafka
+        from confluent_kafka import DeserializingConsumer
+    except ImportError:
+        confluent_kafka = None
+
+Then, later on, before any usage of `DeserializingConsumer`::
+
+    if not confluent_kafka:
+        warn("Confluent_kafka not installed")
+        return None
+    ...do things with DeserializingConsumer
+
+Consequences
+************
+
+This will make developers or other users of the edx-event-bus-kafka library responsible for installing confluent-kafka in their own environments.
+
+For edx.org, we will install confluent-kafka as part of creating the docker containers that will run the services
+that use this library.
+
+Rejected Alternatives
+*********************
+
+* Make the entire `edx-event-bus-kafka` library an optional dependency in the services that use it (eg edx-platform, course-discovery)
+
+This would require developers to install `edx-event-bus-kafka` separately when setting up their environment. This means it would not be able to be updated with `make upgrade` in the same way we manage versions of all of our other packages. Moreover, this would require separate commits to update the version of the package and update the code that uses it, meaning we would have to use an expand-contract release model for every breaking change. This goes against best practices, being highly error-prone.
+
+We expect edx-event-bus-kafka to change more frequently than `confluent-kafka`, which is why we are more willing to adopt the optional dependency strategy for the latter.
+
+* Keep both `confluent-kafka` and `edx-event-bus-kafka` as a required dependencies
+
+While not necessarily causing problems for edx.org, this would break many community-hosted Open edX instances as well as many development environments.

docs/how_tos/manual_testing.rst

@@ -17,6 +17,7 @@ The producer can be tested manually against a Kafka running in devstack.
 
 #. In devstack, run ``make devpi-up studio-up-without-deps-shell`` to bring up Studio with a shell.
 #. In the Studio shell, run ``pip install -e /edx/src/event-bus-kafka``
+#. In the Studio shell, run ``pip install confluent-kafka`` (necessary external dependency)
 #. Test the producer:
 
    - Run the example command listed in the ``edx_event_bus_kafka.management.commands.produce_event.Command`` docstring

edx_event_bus_kafka/config.py

@@ -5,14 +5,24 @@
 import warnings
 from typing import Optional
 
-from confluent_kafka.schema_registry import SchemaRegistryClient
 from django.conf import settings
 
+try:
+    import confluent_kafka
+    from confluent_kafka.schema_registry import SchemaRegistryClient
+except ImportError:
+    confluent_kafka = None
 
-def create_schema_registry_client() -> Optional[SchemaRegistryClient]:
+
+# return type (Optional[SchemaRegistryClient]) removed for better error messaging when confluent-kafka is not available
+def create_schema_registry_client():
     """
     Create a schema registry client from common settings.
     """
+    if not confluent_kafka:
+        warnings.warn('Library confluent-kafka not available. Cannot create schema registry client.')
+        return None
+
     url = getattr(settings, 'EVENT_BUS_KAFKA_SCHEMA_REGISTRY_URL', None)
     if url is None:
         warnings.warn("Cannot configure event-bus-kafka: Missing setting EVENT_BUS_KAFKA_SCHEMA_REGISTRY_URL")

edx_event_bus_kafka/consumer/event_consumer.py

@@ -4,8 +4,6 @@
 
 import logging
 
-from confluent_kafka import DeserializingConsumer, KafkaError
-from confluent_kafka.schema_registry.avro import AvroDeserializer
 from django.conf import settings
 from django.core.management.base import BaseCommand
 from django.dispatch import receiver
@@ -19,6 +17,13 @@
 
 logger = logging.getLogger(__name__)
 
+try:
+    import confluent_kafka
+    from confluent_kafka import DeserializingConsumer, KafkaError
+    from confluent_kafka.schema_registry.avro import AvroDeserializer
+except ImportError:
+    confluent_kafka = None
+
 # .. toggle_name: EVENT_BUS_KAFKA_CONSUMERS_ENABLED
 # .. toggle_implementation: SettingToggle
 # .. toggle_default: False
@@ -44,12 +49,17 @@ class KafkaEventConsumer:
     """
 
     def __init__(self, topic, group_id, signal):
-        self.topic = topic
-        self.group_id = group_id
-        self.signal = signal
-        self.consumer = self._create_consumer()
+        if confluent_kafka:
+            self.topic = topic
+            self.group_id = group_id
+            self.signal = signal
+            self.consumer = self._create_consumer()
+        else:
+            raise Exception('Library confluent-kafka not available. Cannot create event consumer.')
 
-    def _create_consumer(self) -> DeserializingConsumer:
+    # return type (Optional[DeserializingConsumer]) removed for better error messaging when confluent-kafka is not
+    # available
+    def _create_consumer(self):
         """
         Create a DeserializingConsumer for events of the given signal instance.
         """
@@ -165,6 +175,7 @@ class ConsumeEventsCommand(BaseCommand):
 
     example:
         python3 manage.py cms consume_events -t user-event-debug -g user-event-consumers
+            -s org.openedx.learning.auth.session.login.completed.v1
 
     # TODO (EventBus): Add pointer to relevant future docs around topics and consumer groups, and potentially
     update example topic and group names to follow any future naming conventions.

edx_event_bus_kafka/consumer/tests/test_event_consumer.py

@@ -6,7 +6,6 @@
 from contextlib import contextmanager
 from unittest.mock import Mock, patch
 
-from confluent_kafka.serialization import StringSerializer
 from django.core.management import call_command
 from django.test import TestCase
 from django.test.utils import override_settings
@@ -16,6 +15,12 @@
 from edx_event_bus_kafka.consumer.event_consumer import KafkaEventConsumer
 from edx_event_bus_kafka.management.commands.consume_events import Command
 
+try:
+    import confluent_kafka
+    from confluent_kafka.serialization import StringSerializer
+except ImportExcept:
+    confluent_kafka = None
+
 
 class FakeMessage:
     """

edx_event_bus_kafka/management/commands/produce_event.py

@@ -19,8 +19,7 @@ class Command(BaseCommand):
     Produce a single test event.
     """
     help = """
-    This starts a Kafka event consumer that listens to the specified topic and logs all messages it receives. Topic
-    is required.
+    This starts a Kafka event consumer that produces a test event with the given data to the specified topic.
 
     example:
         python3 manage.py cms produce_event --signal openedx_events.learning.signals.SESSION_LOGIN_COMPLETED \

edx_event_bus_kafka/publishing/event_producer.py

@@ -7,17 +7,22 @@
 import json
 import logging
 from functools import lru_cache
-from typing import Any, List, Optional
+from typing import Any, List
 
-from confluent_kafka import SerializingProducer
-from confluent_kafka.schema_registry.avro import AvroSerializer
 from openedx_events.event_bus.avro.serializer import AvroSignalSerializer
 from openedx_events.tooling import OpenEdxPublicSignal
 
 from edx_event_bus_kafka.config import create_schema_registry_client, load_common_settings
 
 logger = logging.getLogger(__name__)
 
+try:
+    import confluent_kafka
+    from confluent_kafka import SerializingProducer
+    from confluent_kafka.schema_registry.avro import AvroSerializer
+except ImportError:
+    confluent_kafka = None
+
 # CloudEvent standard name for the event type header, see
 # https://github.com/cloudevents/spec/blob/v1.0.1/kafka-protocol-binding.md#325-example
 EVENT_TYPE_HEADER_KEY = "ce_type"
@@ -118,8 +123,10 @@ def get_serializer(signal: OpenEdxPublicSignal) -> AvroSignalSerializer:
 # fall out of scope and be garbage-collected, destroying the
 # outbound-message queue and threads. The use of this cache allows the
 # producers to be long-lived.
+
+# return type (Optional[SerializingProducer]) removed for better error messaging when confluent-kafka is not available
 @lru_cache
-def get_producer_for_signal(signal: OpenEdxPublicSignal, event_key_field: str) -> Optional[SerializingProducer]:
+def get_producer_for_signal(signal: OpenEdxPublicSignal, event_key_field: str):
     """
     Create the producer for a signal and a key field path.
 
@@ -135,6 +142,10 @@ def get_producer_for_signal(signal: OpenEdxPublicSignal, event_key_field: str) -
         remote-config (and in particular does not result in mixed cache/uncached configuration).
         This complexity is being deferred until this becomes a performance issue.
     """
+    if not confluent_kafka:
+        logger.warning('Library confluent-kafka not available. Cannot create event producer.')
+        return None
+
     schema_registry_client = create_schema_registry_client()
     if schema_registry_client is None:
         return None

edx_event_bus_kafka/publishing/test_event_producer.py

@@ -8,13 +8,17 @@
 
 import openedx_events.learning.signals
 import pytest
-from confluent_kafka import SerializingProducer
 from django.test import override_settings
 from openedx_events.event_bus.avro.serializer import AvroSignalSerializer
 from openedx_events.learning.data import UserData, UserPersonalData
 
 import edx_event_bus_kafka.publishing.event_producer as ep
 
+try:
+    from confluent_kafka import SerializingProducer
+except ImportError:
+    confluent_kafka = None
+
 
 class TestEventProducer(TestCase):
     """Test producer."""

requirements/base.in

@@ -2,7 +2,6 @@
 -c constraints.txt
 
 Django                                  # Web application framework
-confluent_kafka[avro,schema-registry]   # Kafka client
 openedx-events                          # Events API
 edx_django_utils
 edx_toggles

requirements/base.txt

@@ -6,25 +6,17 @@
 #
 asgiref==3.5.2
     # via django
-attrs==21.4.0
+attrs==22.1.0
     # via openedx-events
-avro==1.10.0
-    # via confluent-kafka
-certifi==2022.6.15
-    # via requests
 cffi==1.15.1
     # via pynacl
-charset-normalizer==2.1.0
-    # via requests
 click==8.1.3
     # via
     #   code-annotations
     #   edx-django-utils
 code-annotations==1.3.0
     # via edx-toggles
-confluent-kafka[avro,schema-registry]==1.9.0
-    # via -r requirements/base.in
-django==3.2.14
+django==3.2.15
     # via
     #   -c https://raw.githubusercontent.com/edx/edx-lint/master/edx_lint/files/common_constraints.txt
     #   -r requirements/base.in
@@ -48,21 +40,17 @@ edx-opaque-keys[django]==2.3.0
     # via openedx-events
 edx-toggles==5.0.0
     # via -r requirements/base.in
-fastavro==1.5.3
-    # via
-    #   confluent-kafka
-    #   openedx-events
-idna==3.3
-    # via requests
+fastavro==1.6.0
+    # via openedx-events
 jinja2==3.1.2
     # via code-annotations
 markupsafe==2.1.1
     # via jinja2
-newrelic==7.14.0.177
+newrelic==8.0.0.179
     # via edx-django-utils
-openedx-events==0.10.0
+openedx-events==0.12.0
     # via -r requirements/base.in
-pbr==5.9.0
+pbr==5.10.0
     # via stevedore
 psutil==5.9.1
     # via edx-django-utils
@@ -74,12 +62,10 @@ pynacl==1.5.0
     # via edx-django-utils
 python-slugify==6.1.2
     # via code-annotations
-pytz==2022.1
+pytz==2022.2.1
     # via django
 pyyaml==6.0
     # via code-annotations
-requests==2.28.1
-    # via confluent-kafka
 sqlparse==0.4.2
     # via django
 stevedore==4.0.0
@@ -89,5 +75,3 @@ stevedore==4.0.0
     #   edx-opaque-keys
 text-unidecode==1.3
     # via python-slugify
-urllib3==1.26.10
-    # via requests

requirements/ci.txt

@@ -10,11 +10,11 @@ charset-normalizer==2.1.0
     # via requests
 codecov==2.1.12
     # via -r requirements/ci.in
-coverage==6.4.2
+coverage==6.4.4
     # via codecov
 distlib==0.3.5
     # via virtualenv
-filelock==3.7.1
+filelock==3.8.0
     # via
     #   tox
     #   virtualenv
@@ -33,14 +33,12 @@ pyparsing==3.0.9
 requests==2.28.1
     # via codecov
 six==1.16.0
-    # via
-    #   tox
-    #   virtualenv
+    # via tox
 toml==0.10.2
     # via tox
 tox==3.25.1
     # via -r requirements/ci.in
-urllib3==1.26.10
+urllib3==1.26.11
     # via requests
-virtualenv==20.15.1
+virtualenv==20.16.3
     # via tox