diff --git a/cms/djangoapps/contentstore/signals/handlers.py b/cms/djangoapps/contentstore/signals/handlers.py
index fec46b6b7484..20c14089e0a6 100644
--- a/cms/djangoapps/contentstore/signals/handlers.py
+++ b/cms/djangoapps/contentstore/signals/handlers.py
@@ -13,13 +13,7 @@
 from edx_toggles.toggles import SettingToggle
 from opaque_keys.edx.keys import CourseKey
 from openedx_events.content_authoring.data import CourseCatalogData, CourseScheduleData
-from openedx_events.content_authoring.signals import (
-    COURSE_CATALOG_INFO_CHANGED,
-    XBLOCK_DELETED,
-    XBLOCK_DUPLICATED,
-    XBLOCK_PUBLISHED,
-)
-from openedx_events.event_bus import get_producer
+from openedx_events.content_authoring.signals import COURSE_CATALOG_INFO_CHANGED
 from pytz import UTC
 
 from cms.djangoapps.contentstore.courseware_index import (
@@ -159,60 +153,6 @@ def listen_for_course_publish(sender, course_key, **kwargs):  # pylint: disable=
     transaction.on_commit(lambda: emit_catalog_info_changed_signal(course_key))
 
 
-@receiver(COURSE_CATALOG_INFO_CHANGED)
-def listen_for_course_catalog_info_changed(sender, signal, **kwargs):
-    """
-    Publish COURSE_CATALOG_INFO_CHANGED signals onto the event bus.
-    """
-    get_producer().send(
-        signal=COURSE_CATALOG_INFO_CHANGED, topic='course-catalog-info-changed',
-        event_key_field='catalog_info.course_key', event_data={'catalog_info': kwargs['catalog_info']},
-        event_metadata=kwargs['metadata'],
-    )
-
-
-@receiver(XBLOCK_PUBLISHED)
-def listen_for_xblock_published(sender, signal, **kwargs):
-    """
-    Publish XBLOCK_PUBLISHED signals onto the event bus.
-    """
-    if settings.FEATURES.get("ENABLE_SEND_XBLOCK_EVENTS_OVER_BUS"):
-        topic = getattr(settings, "EVENT_BUS_XBLOCK_LIFECYCLE_TOPIC", "course-authoring-xblock-lifecycle")
-        get_producer().send(
-            signal=XBLOCK_PUBLISHED, topic=topic,
-            event_key_field='xblock_info.usage_key', event_data={'xblock_info': kwargs['xblock_info']},
-            event_metadata=kwargs['metadata'],
-        )
-
-
-@receiver(XBLOCK_DELETED)
-def listen_for_xblock_deleted(sender, signal, **kwargs):
-    """
-    Publish XBLOCK_DELETED signals onto the event bus.
-    """
-    if settings.FEATURES.get("ENABLE_SEND_XBLOCK_EVENTS_OVER_BUS"):
-        topic = getattr(settings, "EVENT_BUS_XBLOCK_LIFECYCLE_TOPIC", "course-authoring-xblock-lifecycle")
-        get_producer().send(
-            signal=XBLOCK_DELETED, topic=topic,
-            event_key_field='xblock_info.usage_key', event_data={'xblock_info': kwargs['xblock_info']},
-            event_metadata=kwargs['metadata'],
-        )
-
-
-@receiver(XBLOCK_DUPLICATED)
-def listen_for_xblock_duplicated(sender, signal, **kwargs):
-    """
-    Publish XBLOCK_DUPLICATED signals onto the event bus.
-    """
-    if settings.FEATURES.get("ENABLE_SEND_XBLOCK_EVENTS_OVER_BUS"):
-        topic = getattr(settings, "EVENT_BUS_XBLOCK_LIFECYCLE_TOPIC", "course-authoring-xblock-lifecycle")
-        get_producer().send(
-            signal=XBLOCK_DUPLICATED, topic=topic,
-            event_key_field='xblock_info.usage_key', event_data={'xblock_info': kwargs['xblock_info']},
-            event_metadata=kwargs['metadata'],
-        )
-
-
 @receiver(SignalHandler.course_deleted)
 def listen_for_course_delete(sender, course_key, **kwargs):  # pylint: disable=unused-argument
     """
diff --git a/cms/envs/common.py b/cms/envs/common.py
index 12b982d95d9b..e511a9115e0f 100644
--- a/cms/envs/common.py
+++ b/cms/envs/common.py
@@ -1789,6 +1789,9 @@
 
     # Blockstore
     'blockstore.apps.bundles',
+
+    # Openedx events
+    'openedx_events',
 ]
 
 
@@ -2725,3 +2728,29 @@
 
 DISCUSSIONS_INCONTEXT_FEEDBACK_URL = ''
 DISCUSSIONS_INCONTEXT_LEARNMORE_URL = ''
+
+######################## Event bus producer config ########################
+
+# .. setting_name: EVENT_BUS_PRODUCER_CONFIG
+# .. setting_default: {}
+# .. setting_description: Dictionary of event_types mapped to lists of dictionaries containing topic related configuration.
+#    Each topic configuration dictionary contains
+#    * a topic/stream name called `topic` where the event will be pushed to.
+#    * a flag called `enabled` denoting whether the event will be published to the topic.
+#    * `event_key_field` which is a period-delimited string path to event data field to use as event key.
+#    Note: The topic names should not include environment prefix as it will be dynamically added based on
+#    EVENT_BUS_TOPIC_PREFIX setting.
+EVENT_BUS_PRODUCER_CONFIG = {
+    'org.openedx.content_authoring.course.catalog_info.changed.v1': [
+        {'topic': 'course-catalog-info-changed', 'event_key_field': 'catalog_info.course_key', 'enabled': True},
+    ],
+    'org.openedx.content_authoring.xblock.published.v1': [
+        {'topic': 'content-authoring-xblock-lifecycle', 'event_key_field': 'xblock_info.usage_key', 'enabled': True},
+    ],
+    'org.openedx.content_authoring.xblock.deleted.v1': [
+        {'topic': 'content-authoring-xblock-lifecycle', 'event_key_field': 'xblock_info.usage_key', 'enabled': True},
+    ],
+    'org.openedx.content_authoring.xblock.duplicated.v1': [
+        {'topic': 'content-authoring-xblock-lifecycle', 'event_key_field': 'xblock_info.usage_key', 'enabled': True},
+    ],
+}
diff --git a/cms/envs/devstack.py b/cms/envs/devstack.py
index 4170af3d390e..3771ee06443d 100644
--- a/cms/envs/devstack.py
+++ b/cms/envs/devstack.py
@@ -293,21 +293,10 @@ def should_show_debug_toolbar(request):  # lint-amnesty, pylint: disable=missing
 CREDENTIALS_PUBLIC_SERVICE_URL = 'http://localhost:18150'
 
 #################### Event bus backend ########################
-# .. toggle_name: FEATURES['ENABLE_SEND_XBLOCK_EVENTS_OVER_BUS']
-# .. toggle_implementation: DjangoSetting
-# .. toggle_default: False
-# .. toggle_description: Temporary configuration which enables sending xblock events over the event bus.
-# .. toggle_use_cases: open_edx
-# .. toggle_creation_date: 2023-02-21
-# .. toggle_warning: For consistency in user experience, keep the value in sync with the setting of the same name
-#   in the LMS and CMS.
-# .. toggle_tickets: 'https://github.com/openedx/edx-platform/pull/31813'
-FEATURES['ENABLE_SEND_XBLOCK_EVENTS_OVER_BUS'] = True
 EVENT_BUS_PRODUCER = 'edx_event_bus_redis.create_producer'
 EVENT_BUS_REDIS_CONNECTION_URL = 'redis://:password@edx.devstack.redis:6379/'
 EVENT_BUS_TOPIC_PREFIX = 'dev'
 EVENT_BUS_CONSUMER = 'edx_event_bus_redis.RedisEventConsumer'
-EVENT_BUS_XBLOCK_LIFECYCLE_TOPIC = 'course-authoring-xblock-lifecycle'
 
 ################# New settings must go ABOVE this line #################
 ########################################################################
diff --git a/lms/djangoapps/certificates/signals.py b/lms/djangoapps/certificates/signals.py
index 676c10fb9c8f..28e68e340253 100644
--- a/lms/djangoapps/certificates/signals.py
+++ b/lms/djangoapps/certificates/signals.py
@@ -6,12 +6,10 @@
 
 from django.db.models.signals import post_save
 from django.dispatch import receiver
-from openedx_events.event_bus import get_producer
 
 from common.djangoapps.course_modes import api as modes_api
 from common.djangoapps.student.models import CourseEnrollment
 from common.djangoapps.student.signals import ENROLLMENT_TRACK_UPDATED
-from lms.djangoapps.certificates.config import SEND_CERTIFICATE_CREATED_SIGNAL
 from lms.djangoapps.certificates.generation_handler import (
     CertificateGenerationNotAllowed,
     generate_allowlist_certificate_task,
@@ -32,7 +30,6 @@
     COURSE_GRADE_NOW_PASSED,
     LEARNER_NOW_VERIFIED
 )
-from openedx_events.learning.signals import CERTIFICATE_CREATED
 
 log = logging.getLogger(__name__)
 
@@ -159,18 +156,3 @@ def _listen_for_enrollment_mode_change(sender, user, course_key, mode, **kwargs)
                 course_key,
             )
             return False
-
-
-@receiver(CERTIFICATE_CREATED)
-def listen_for_certificate_created_event(sender, signal, **kwargs):
-    """
-    Publish `CERTIFICATE_CREATED` events to the event bus.
-    """
-    if SEND_CERTIFICATE_CREATED_SIGNAL.is_enabled():
-        get_producer().send(
-            signal=CERTIFICATE_CREATED,
-            topic='learning-certificate-lifecycle',
-            event_key_field='certificate.course.course_key',
-            event_data={'certificate': kwargs['certificate']},
-            event_metadata=kwargs['metadata']
-        )
diff --git a/lms/djangoapps/certificates/tests/test_signals.py b/lms/djangoapps/certificates/tests/test_signals.py
index b5993eb623c3..320eec4d61d0 100644
--- a/lms/djangoapps/certificates/tests/test_signals.py
+++ b/lms/djangoapps/certificates/tests/test_signals.py
@@ -3,12 +3,9 @@
 and disabling for instructor-paced courses.
 """
 
-from datetime import datetime, timezone
 from unittest import mock
-from uuid import uuid4
 
 import ddt
-from django.test.utils import override_settings
 from edx_toggles.toggles.testutils import override_waffle_flag, override_waffle_switch
 from xmodule.modulestore.tests.django_utils import ModuleStoreTestCase
 from xmodule.modulestore.tests.factories import CourseFactory
@@ -21,14 +18,10 @@
     CertificateGenerationConfiguration,
     GeneratedCertificate
 )
-from lms.djangoapps.certificates.signals import listen_for_certificate_created_event
 from lms.djangoapps.certificates.tests.factories import CertificateAllowlistFactory, GeneratedCertificateFactory
 from lms.djangoapps.grades.course_grade_factory import CourseGradeFactory
 from lms.djangoapps.grades.tests.utils import mock_passing_grade
 from lms.djangoapps.verify_student.models import SoftwareSecurePhotoVerification
-from openedx_events.data import EventsMetadata
-from openedx_events.learning.signals import CERTIFICATE_CREATED
-from openedx_events.learning.data import CourseData, UserData, UserPersonalData, CertificateData
 
 
 class SelfGeneratedCertsSignalTest(ModuleStoreTestCase):
@@ -440,78 +433,3 @@ def test_verified_to_audit(self):
         ) as mock_allowlist_task:
             self.verified_enrollment.change_mode('audit')
             mock_allowlist_task.assert_not_called()
-
-
-class CertificateEventBusTests(ModuleStoreTestCase):
-    """
-    Tests for Certificate events that interact with the event bus.
-    """
-    def setUp(self):
-        super().setUp()
-        self.user = UserFactory.create()
-        self.name = f'{self.user.first_name} {self.user.last_name}'
-        self.course = CourseFactory.create(self_paced=True)
-        self.enrollment = CourseEnrollmentFactory(
-            user=self.user,
-            course_id=self.course.id,
-            is_active=True,
-            mode='verified',
-        )
-
-    @override_settings(SEND_CERTIFICATE_CREATED_SIGNAL=False)
-    @mock.patch('lms.djangoapps.certificates.signals.get_producer', autospec=True)
-    def test_event_disabled(self, mock_producer):
-        """
-        Test to verify that we do not push `CERTIFICATE_CREATED` events to the event bus if the
-        `SEND_CERTIFICATE_CREATED_SIGNAL` setting is disabled.
-        """
-        listen_for_certificate_created_event(None, CERTIFICATE_CREATED)
-        mock_producer.assert_not_called()
-
-    @override_settings(SEND_CERTIFICATE_CREATED_SIGNAL=True)
-    @mock.patch('lms.djangoapps.certificates.signals.get_producer', autospec=True)
-    def test_event_enabled(self, mock_producer):
-        """
-        Test to verify that we push `CERTIFICATE_CREATED` events to the event bus if the
-        `SEND_CERTIFICATE_CREATED_SIGNAL` setting is enabled.
-        """
-        expected_course_data = CourseData(course_key=self.course.id)
-        expected_user_data = UserData(
-            pii=UserPersonalData(
-                username=self.user.username,
-                email=self.user.email,
-                name=self.name,
-            ),
-            id=self.user.id,
-            is_active=self.user.is_active
-        )
-        expected_certificate_data = CertificateData(
-            user=expected_user_data,
-            course=expected_course_data,
-            mode='verified',
-            grade='',
-            current_status='downloadable',
-            download_url='',
-            name='',
-        )
-        event_metadata = EventsMetadata(
-            event_type=CERTIFICATE_CREATED.event_type,
-            id=uuid4(),
-            minorversion=0,
-            source='openedx/lms/web',
-            sourcehost='lms.test',
-            time=datetime.now(timezone.utc)
-        )
-
-        event_kwargs = {
-            'certificate': expected_certificate_data,
-            'metadata': event_metadata
-        }
-
-        listen_for_certificate_created_event(None, CERTIFICATE_CREATED, **event_kwargs)
-        # verify that the data sent to the event bus matches what we expect
-        data = mock_producer.return_value.send.call_args.kwargs
-        assert data['signal'].event_type == CERTIFICATE_CREATED.event_type
-        assert data['event_data']['certificate'] == expected_certificate_data
-        assert data['topic'] == 'learning-certificate-lifecycle'
-        assert data['event_key_field'] == 'certificate.course.course_key'
diff --git a/lms/envs/common.py b/lms/envs/common.py
index 5cd3a0c06c2d..656ea4717889 100644
--- a/lms/envs/common.py
+++ b/lms/envs/common.py
@@ -3274,6 +3274,9 @@ def _make_locale_paths(settings):  # pylint: disable=missing-function-docstring
 
     # Notifications
     'openedx.core.djangoapps.notifications',
+
+    # Openedx events
+    'openedx_events',
 ]
 
 ######################### CSRF #########################################
@@ -5334,3 +5337,20 @@ def _make_locale_paths(settings):  # pylint: disable=missing-function-docstring
 ############## NOTIFICATIONS EXPIRY ##############
 NOTIFICATIONS_EXPIRY = 60
 EXPIRED_NOTIFICATIONS_DELETE_BATCH_SIZE = 10000
+
+######################## Event bus producer config ########################
+
+# .. setting_name: EVENT_BUS_PRODUCER_CONFIG
+# .. setting_default: {}
+# .. setting_description: Dictionary of event_types mapped to lists of dictionaries containing topic related configuration.
+#    Each topic configuration dictionary contains
+#    * a topic/stream name called `topic` where the event will be pushed to.
+#    * a flag called `enabled` denoting whether the event will be published to the topic.
+#    * `event_key_field` which is a period-delimited string path to event data field to use as event key.
+#    Note: The topic names should not include environment prefix as it will be dynamically added based on
+#    EVENT_BUS_TOPIC_PREFIX setting.
+EVENT_BUS_PRODUCER_CONFIG = {
+    'org.openedx.learning.certificate.created.v1': [
+        {'topic': 'learning-certificate-lifecycle', 'event_key_field': 'certificate.course.course_key', 'enabled': True},
+    ],
+}
diff --git a/requirements/edx/base.txt b/requirements/edx/base.txt
index 67250d9d05ad..c6c1e04d7787 100644
--- a/requirements/edx/base.txt
+++ b/requirements/edx/base.txt
@@ -762,7 +762,7 @@ openedx-django-require==2.1.0
     # via -r requirements/edx/kernel.in
 openedx-django-wiki==2.0.0
     # via -r requirements/edx/kernel.in
-openedx-events==8.3.0
+openedx-events @ git+https://github.com/open-craft/openedx-events@navin/configurable-handler
     # via
     #   -r requirements/edx/kernel.in
     #   edx-event-bus-kafka
diff --git a/requirements/edx/development.txt b/requirements/edx/development.txt
index 743938fb1773..1f8068248319 100644
--- a/requirements/edx/development.txt
+++ b/requirements/edx/development.txt
@@ -1294,7 +1294,7 @@ openedx-django-wiki==2.0.0
     # via
     #   -r requirements/edx/doc.txt
     #   -r requirements/edx/testing.txt
-openedx-events==8.3.0
+openedx-events @ git+https://github.com/open-craft/openedx-events@navin/configurable-handler
     # via
     #   -r requirements/edx/doc.txt
     #   -r requirements/edx/testing.txt
@@ -2138,6 +2138,7 @@ walrus==0.9.3
     #   edx-event-bus-redis
 watchdog==3.0.0
     # via
+    #   -r requirements/edx/development.in
     #   -r requirements/edx/doc.txt
     #   -r requirements/edx/testing.txt
 wcwidth==0.2.6
diff --git a/requirements/edx/doc.txt b/requirements/edx/doc.txt
index 3b584a4bee5e..4b41bc133483 100644
--- a/requirements/edx/doc.txt
+++ b/requirements/edx/doc.txt
@@ -556,7 +556,7 @@ edx-drf-extensions==8.8.0
     #   edx-rbac
     #   edx-when
     #   edxval
-edx-enterprise==4.0.6
+edx-enterprise==4.0.7
     # via
     #   -c requirements/edx/../constraints.txt
     #   -r requirements/edx/base.txt
@@ -905,7 +905,7 @@ openedx-django-require==2.1.0
     # via -r requirements/edx/base.txt
 openedx-django-wiki==2.0.0
     # via -r requirements/edx/base.txt
-openedx-events==8.3.0
+openedx-events @ git+https://github.com/open-craft/openedx-events@navin/configurable-handler
     # via
     #   -r requirements/edx/base.txt
     #   edx-event-bus-kafka
diff --git a/requirements/edx/kernel.in b/requirements/edx/kernel.in
index 8929520ea819..8cdf504893ac 100644
--- a/requirements/edx/kernel.in
+++ b/requirements/edx/kernel.in
@@ -113,7 +113,8 @@ oauthlib                            # OAuth specification support for authentica
 olxcleaner
 openedx-calc                        # Library supporting mathematical calculations for Open edX
 openedx-django-require
-openedx-events>=8.3.0               # Open edX Events from Hooks Extension Framework (OEP-50)
+# openedx-events>=8.3.0               # Open edX Events from Hooks Extension Framework (OEP-50)
+git+https://github.com/open-craft/openedx-events@navin/configurable-handler
 openedx-filters                     # Open edX Filters from Hooks Extension Framework (OEP-50)
 openedx-learning<=0.1
 openedx-mongodbproxy
diff --git a/requirements/edx/testing.txt b/requirements/edx/testing.txt
index 7165c677dc35..e7c4ab1e59b4 100644
--- a/requirements/edx/testing.txt
+++ b/requirements/edx/testing.txt
@@ -974,7 +974,7 @@ openedx-django-require==2.1.0
     # via -r requirements/edx/base.txt
 openedx-django-wiki==2.0.0
     # via -r requirements/edx/base.txt
-openedx-events==8.3.0
+openedx-events @ git+https://github.com/open-craft/openedx-events@navin/configurable-handler
     # via
     #   -r requirements/edx/base.txt
     #   edx-event-bus-kafka