librgw/notifications: initialize kafka and amqp

Fixes: https://tracker.ceph.com/issues/49738
Signed-off-by: Yuval Lifshitz <ylifshit@redhat.com>

diff --git a/src/rgw/librgw.cc b/src/rgw/librgw.cc
index 012cc54c3be8509cf1088e0db0d48978b3dc201c..a5351dbe7c13a20601f86b0360cd56beca9b957a 100644 (file)
--- a/src/rgw/librgw.cc
+++ b/src/rgw/librgw.cc
@@ -53,6 +53,12 @@
 #include "rgw_http_client.h"
 #include "rgw_http_client_curl.h"
 #include "rgw_perf_counters.h"
+#ifdef WITH_RADOSGW_AMQP_ENDPOINT
+#include "rgw_amqp.h"
+#endif
+#ifdef WITH_RADOSGW_KAFKA_ENDPOINT
+#include "rgw_kafka.h"
+#endif
 
 #include "services/svc_zone.h"
 
@@ -617,6 +623,17 @@ namespace rgw {
       /* ignore error */
     }
 
+#ifdef WITH_RADOSGW_AMQP_ENDPOINT
+    if (!rgw::amqp::init(cct.get())) {
+      derr << "ERROR: failed to initialize AMQP manager" << dendl;
+    }
+#endif
+#ifdef WITH_RADOSGW_KAFKA_ENDPOINT
+    if (!rgw::kafka::init(cct.get())) {
+      derr << "ERROR: failed to initialize Kafka manager" << dendl;
+    }
+#endif
+
     return 0;
   } /* RGWLib::init() */
 
@@ -645,6 +662,12 @@ namespace rgw {
     rgw_shutdown_resolver();
     rgw_http_client_cleanup();
     rgw::curl::cleanup_curl();
+#ifdef WITH_RADOSGW_AMQP_ENDPOINT
+    rgw::amqp::shutdown();
+#endif
+#ifdef WITH_RADOSGW_KAFKA_ENDPOINT
+    rgw::kafka::shutdown();
+#endif
 
     rgw_perf_stop(g_ceph_context);