feat!: update event_bus_kafka and add event_bus_redis

feat: add event bus redis as dependency refactor: add openedx_events in installed apps to use consume_events command chore: upgrade dependencies
openedx · May 16, 2023 · 1faa3c0 · 1faa3c0
1 parent 7019422
commit 1faa3c0
Show file tree

Hide file tree

Showing 9 changed files with 45 additions and 31 deletions.
diff --git a/course_discovery/apps/course_metadata/management/commands/consume_events.py b/course_discovery/apps/course_metadata/management/commands/consume_events.py
diff --git a/course_discovery/apps/course_metadata/signals.py b/course_discovery/apps/course_metadata/signals.py
@@ -354,10 +354,10 @@ def update_course_data_from_event(**kwargs):
     if event_metadata:
         event_timestamp = event_metadata.time
         time_diff = datetime.now(tz=timezone.utc) - event_timestamp
-        if time_diff.seconds < settings.EVENT_BUS_KAFKA_MESSAGE_DELAY_THRESHOLD_SECONDS:
+        if time_diff.seconds < settings.EVENT_BUS_MESSAGE_DELAY_THRESHOLD_SECONDS:
             logger.debug(f"COURSE_CATALOG_INFO_CHANGED event received within the delay "
                          f"applicable window for course run {course_data.course_key}.")
-            time.sleep(settings.EVENT_BUS_KAFKA_PROCESSING_DELAY_SECONDS)
+            time.sleep(settings.EVENT_BUS_PROCESSING_DELAY_SECONDS)
 
     # Handle optional fields.
     schedule_data = course_data.schedule_data

diff --git a/course_discovery/apps/course_metadata/tests/test_signals.py b/course_discovery/apps/course_metadata/tests/test_signals.py
@@ -865,7 +865,7 @@ def test_event_processing_delay(self, sleep_patch):
         applicable time window.
         """
         metadata = EventsMetadata(event_type='catalog-data-changed', minorversion=0)
-        with override_settings(EVENT_BUS_KAFKA_MESSAGE_DELAY_THRESHOLD_SECONDS=120):
+        with override_settings(EVENT_BUS_MESSAGE_DELAY_THRESHOLD_SECONDS=120):
             with self.assertLogs(LOGGER_NAME, level="DEBUG") as logger:
                 update_course_data_from_event(
                     catalog_info=self.catalog_data,
@@ -874,4 +874,4 @@ def test_event_processing_delay(self, sleep_patch):
                 assert f"COURSE_CATALOG_INFO_CHANGED event received within the " \
                        f"delay applicable window for course run {self.course_key}." in logger.output[0]
 
-        sleep_patch.assert_called_once_with(settings.EVENT_BUS_KAFKA_PROCESSING_DELAY_SECONDS)
+        sleep_patch.assert_called_once_with(settings.EVENT_BUS_PROCESSING_DELAY_SECONDS)
diff --git a/course_discovery/settings/base.py b/course_discovery/settings/base.py
@@ -66,6 +66,7 @@
     'taxonomy',
     'django_object_actions',
     'nested_admin',
+    'openedx_events',
 ]
 
 ALGOLIA = {
@@ -721,8 +722,8 @@
     },
 }
 
-EVENT_BUS_KAFKA_PROCESSING_DELAY_SECONDS = 60
-EVENT_BUS_KAFKA_MESSAGE_DELAY_THRESHOLD_SECONDS = 60
+EVENT_BUS_PROCESSING_DELAY_SECONDS = 60
+EVENT_BUS_MESSAGE_DELAY_THRESHOLD_SECONDS = 60
 
 ALGOLIA_INDEX_EXCLUDED_SOURCES = []
 

diff --git a/course_discovery/settings/devstack.py b/course_discovery/settings/devstack.py
@@ -86,6 +86,11 @@
 
 CELERY_TASK_ALWAYS_EAGER = False
 
+EVENT_BUS_CONSUMER = 'edx_event_bus_redis.RedisEventConsumer'
+EVENT_BUS_PRODUCER = 'edx_event_bus_redis.create_producer'
+EVENT_BUS_REDIS_CONNECTION_URL = 'redis://:[email protected]:6379/'
+EVENT_BUS_TOPIC_PREFIX = 'dev'
+
 #####################################################################
 # Lastly, see if the developer has any local overrides.
 if os.path.isfile(join(dirname(abspath(__file__)), 'private.py')):

diff --git a/requirements/base.in b/requirements/base.in
@@ -49,8 +49,9 @@ edx-django-release-util
 edx-django-sites-extensions
 edx-django-utils
 edx-drf-extensions
-# edx-event-bus-kafka 3.7.0 introduces `reset_offsets_and_sleep_indefinitely`
-edx-event-bus-kafka>=3.7.1
+# edx-event-bus-kafka 4.0.0 adds support for configurable consumer API
+edx-event-bus-kafka>=4.0.1
+edx-event-bus-redis
 edx-opaque-keys
 edx-rest-api-client
 elasticsearch

diff --git a/requirements/constraints.txt b/requirements/constraints.txt
@@ -44,6 +44,3 @@ cairocffi < 1.5.0
 
 # 1.0.0 has breaking changes
 requests-toolbelt==0.10.1
-
-# 4.0.0 introduces some breaking changes  which will be resolved in a followup PR
-edx-event-bus-kafka==3.9.6
diff --git a/requirements/local.txt b/requirements/local.txt
@@ -210,6 +210,7 @@ distro==1.8.0
     #   edx-django-utils
     #   edx-drf-extensions
     #   edx-event-bus-kafka
+    #   edx-event-bus-redis
     #   edx-i18n-tools
     #   edx-toggles
     #   jsonfield
@@ -375,6 +376,7 @@ edx-django-utils==5.4.0
     #   django-config-models
     #   edx-drf-extensions
     #   edx-event-bus-kafka
+    #   edx-event-bus-redis
     #   edx-rest-api-client
     #   edx-toggles
     #   getsmarter-api-clients
@@ -383,10 +385,10 @@ edx-drf-extensions==8.2.0
     # via
     #   -c requirements/constraints.txt
     #   -r requirements/base.in
-edx-event-bus-kafka==3.9.6
-    # via
-    #   -c requirements/constraints.txt
-    #   -r requirements/base.in
+edx-event-bus-kafka==4.0.1
+    # via -r requirements/base.in
+edx-event-bus-redis==0.1.1
+    # via -r requirements/base.in
 edx-i18n-tools==0.9.2
     # via -r requirements/local.in
 edx-lint==5.3.4
@@ -403,7 +405,9 @@ edx-rest-api-client==5.5.0
     #   -r requirements/base.in
     #   taxonomy-connector
 edx-toggles==5.0.0
-    # via edx-event-bus-kafka
+    # via
+    #   edx-event-bus-kafka
+    #   edx-event-bus-redis
 elasticsearch==7.13.4
     # via
     #   -c requirements/common_constraints.txt
@@ -556,6 +560,7 @@ openai==0.27.6
 openedx-events==7.2.0
     # via
     #   edx-event-bus-kafka
+    #   edx-event-bus-redis
     #   taxonomy-connector
 oscrypto==1.3.0
     # via snowflake-connector-python
@@ -742,7 +747,9 @@ pyyaml==5.4.1
 rcssmin==1.1.1
     # via django-compressor
 redis==4.5.5
-    # via -r requirements/base.in
+    # via
+    #   -r requirements/base.in
+    #   walrus
 requests==2.30.0
     # via
     #   -r requirements/base.in
@@ -954,6 +961,8 @@ vine==5.0.0
     #   kombu
 virtualenv==20.23.0
     # via tox
+walrus==0.9.2
+    # via edx-event-bus-redis
 wcmatch==8.4.1
     # via semgrep
 wcwidth==0.2.6

diff --git a/requirements/production.txt b/requirements/production.txt
@@ -155,6 +155,7 @@ django==3.2.19
     #   edx-django-utils
     #   edx-drf-extensions
     #   edx-event-bus-kafka
+    #   edx-event-bus-redis
     #   edx-toggles
     #   jsonfield
     #   openedx-events
@@ -304,6 +305,7 @@ edx-django-utils==5.4.0
     #   django-config-models
     #   edx-drf-extensions
     #   edx-event-bus-kafka
+    #   edx-event-bus-redis
     #   edx-rest-api-client
     #   edx-toggles
     #   getsmarter-api-clients
@@ -312,10 +314,10 @@ edx-drf-extensions==8.2.0
     # via
     #   -c requirements/constraints.txt
     #   -r requirements/base.in
-edx-event-bus-kafka==3.9.6
-    # via
-    #   -c requirements/constraints.txt
-    #   -r requirements/base.in
+edx-event-bus-kafka==4.0.1
+    # via -r requirements/base.in
+edx-event-bus-redis==0.1.1
+    # via -r requirements/base.in
 edx-opaque-keys[django]==2.3.0
     # via
     #   -r requirements/base.in
@@ -328,7 +330,9 @@ edx-rest-api-client==5.5.0
     #   -r requirements/base.in
     #   taxonomy-connector
 edx-toggles==5.0.0
-    # via edx-event-bus-kafka
+    # via
+    #   edx-event-bus-kafka
+    #   edx-event-bus-redis
 elasticsearch==7.13.4
     # via
     #   -c requirements/common_constraints.txt
@@ -446,6 +450,7 @@ openai==0.27.6
 openedx-events==7.2.0
     # via
     #   edx-event-bus-kafka
+    #   edx-event-bus-redis
     #   taxonomy-connector
 oscrypto==1.3.0
     # via snowflake-connector-python
@@ -547,7 +552,9 @@ pyyaml==6.0
 rcssmin==1.1.1
     # via django-compressor
 redis==4.5.5
-    # via -r requirements/base.in
+    # via
+    #   -r requirements/base.in
+    #   walrus
 requests==2.30.0
     # via
     #   -r requirements/base.in
@@ -670,6 +677,8 @@ vine==5.0.0
     #   amqp
     #   celery
     #   kombu
+walrus==0.9.2
+    # via edx-event-bus-redis
 wcwidth==0.2.6
     # via prompt-toolkit
 webencodings==0.5.1