add dispatchKafkaCallbacks

Author: orianar

src/app/contextBroker/contextBroker.cpp

@@ -1321,7 +1321,8 @@ int main(int argC, char* argV[])
   while (1)
   {
     // Sleep periodically during a minute
-    sleep(60);
+    sleep(1);
+    kafkaMgr.dispatchKafkaCallbacks();
 
     // At the present moment MQTT max age checking is the only one periodic process we need to do
     // If some other is introduced in the future, this part should be adapted.
@@ -1331,12 +1332,12 @@ int main(int argC, char* argV[])
     // checking should be moved to a separate thread or, the other way arround, the cache sync
     // process be included as part of this sleep loop
     times++;
-    if (times == mqttMaxAge)
+    if (times == mqttMaxAge * 60)
     {
       times = 0;
       mqttMgr.cleanup(mqttMaxAge*60);
     }
-    if (times == kafkaMaxAge)
+    if (times == kafkaMaxAge * 60)
     {
       times = 0;
       kafkaMgr.cleanup(kafkaMaxAge*60);

src/lib/kafka/KafkaConnectionManager.cpp

@@ -95,7 +95,6 @@ void KafkaConnectionManager::init(long _timeout)
   }
 
   semInit();
-  startPollingThread();
 }
 
 
@@ -108,8 +107,6 @@ void KafkaConnectionManager::teardown(void)
 {
   LM_T(LmtKafkaNotif, ("Teardown KAFKA connections"));
 
-  stopPollingThread();
-
   for (std::map<std::string, KafkaConnection*>::iterator iter = connections.begin(); iter != connections.end(); ++iter)
   {
     std::string endpoint = iter->first;
@@ -198,46 +195,30 @@ void KafkaConnectionManager::semGive(void)
 
 /* ****************************************************************************
 *
-* KafkaConnectionManager::startPollingThread -
+* KafkaConnectionManager::dispatchKafkaCallbacks -
 */
-void KafkaConnectionManager::startPollingThread() {
-  stopPolling = false;
-  pollingThread = std::thread([this]() {
-      while (!stopPolling.load()) {
-        // Aquí llamamos a poll con un timeout de, digamos, 100 ms
-        semTake();
-        // Copiar referencias para minimizar tiempo de bloqueo
-        std::vector<rd_kafka_t*> producers;
-        for (auto& kv : connections) {
-            producers.push_back(kv.second->producer);
-        }
-        semGive();
-
-        // Hacer poll fuera del bloque sincronizado
-        for (auto* producer : producers) {
-            rd_kafka_poll(producer, 0);
-        }
-          // Pequeña pausa para no consumir CPU innecesariamente
-          std::this_thread::sleep_for(std::chrono::milliseconds(100));
-      }
-  });
-}
-
+void KafkaConnectionManager::dispatchKafkaCallbacks()
+{
+  // We copy the producers under the semaphore to minimize critical section time.
+  std::vector<rd_kafka_t*> producers;
+  producers.reserve(connections.size());
 
+  semTake();
+  for (auto& kv : connections) {
+    KafkaConnection* k = kv.second;
+    if (k && k->producer)
+      producers.push_back(k->producer);
+  }
+  semGive();
 
-/* ****************************************************************************
-*
-* KafkaConnectionManager::stopPollingThread -
-*/
-void KafkaConnectionManager::stopPollingThread() {
-  stopPolling = true;
-  if (pollingThread.joinable()) {
-    pollingThread.join();
+  // We fire callbacks outside the critical section
+  for (auto* p : producers) {
+    // 0 ms: does not block; if you call frequently, this is sufficient
+    rd_kafka_poll(p, 0);
   }
 }
 
 
-
 /* ****************************************************************************
 *
 * KafkaConnectionManager::disconnect -

src/lib/kafka/KafkaConnectionManager.h

@@ -32,8 +32,6 @@
 #include <string>
 #include <map>
 #include <rdkafka.h>
-#include <thread>
-#include <atomic>
 
 
 /* ****************************************************************************
@@ -79,18 +77,14 @@ class KafkaConnectionManager
  );
 
  void cleanup(double maxAge);  // Cleaning inactive connections
-
- void stopPollingThread();
- void startPollingThread();
+ void dispatchKafkaCallbacks();
 
 private:
  void disconnect(rd_kafka_t* producer, const std::string& endpoint);
  void semInit(void);
  void semTake(void);
  void semGive(void);
  KafkaConnection* getConnection(const std::string& endpoint);
- std::thread pollingThread;
- std::atomic<bool> stopPolling{false};
 };
 
 #endif  // SRC_LIB_KAFKA_KAFKACONNECTIONMANAGER_H_

test/functionalTest/cases/4666_kafka/kafka_subscription_multibroker.test

@@ -45,9 +45,9 @@ accumulatorStart --pretty-print --bootstrapServers "${KAFKA_BOOTSTRAP_A}" --kafk
 # 04. Create KAFKA sub4 at localhost:9095 (2)
 # 05. Create entity E9092:9094 to stimulate connections to localhost:9092,localhost:9094 (one created, one reused)
 # 06. Sleep 60 seconds, then create entity E1884 to stimulate connections to localhost:9095 (one created, one reused)
-# 07. Sleep 60 seconds, get logs and see localhost:9092,localhost:9094 connection went away
+# 07. Sleep 70 seconds, get logs and see localhost:9092,localhost:9094 connection went away
 # 08. Update entity E9092:9094 to re-create connections to localhost:9092,localhost:9094
-# 09. Sleep 60 seconds, get logs and see localhost:9095 connection went away
+# 09. Sleep 70 seconds, get logs and see localhost:9095 connection went away
 # 10. Dump accumulator for localhost:9092,localhost:9094 and see 4 notifications
 #
 
@@ -178,9 +178,9 @@ echo
 echo
 
 
-echo "07. Sleep 60 seconds, get logs and see localhost:9092,localhost:9094 connection went away"
+echo "07. Sleep 70 seconds, get logs and see localhost:9092,localhost:9094 connection went away"
 echo "========================================================================================="
-sleep 60s
+sleep 70s
 cat /tmp/contextBroker.log | grep 'DEBUG\|ERROR\|WARN' | grep -v 'start command line' | awk -F 'msg=' '{print $2}'
 echo
 echo
@@ -198,9 +198,9 @@ orionCurl --url /v2/entities/E9092:9094/attrs --payload "$payload"
 echo
 echo
 
-echo "09. Sleep 60 seconds, get logs and see localhost:9095 connection went away"
+echo "09. Sleep 70 seconds, get logs and see localhost:9095 connection went away"
 echo "=========================================================================="
-sleep 60s
+sleep 70s
 cat /tmp/contextBroker.log | grep 'DEBUG\|ERROR\|WARN' | grep -v 'start command line' | awk -F 'msg=' '{print $2}'
 echo
 echo
@@ -280,7 +280,7 @@ Content-Length: 0
 
 
 
-07. Sleep 60 seconds, get logs and see localhost:9092,localhost:9094 connection went away
+07. Sleep 70 seconds, get logs and see localhost:9092,localhost:9094 connection went away
 =========================================================================================
 #SORT_START
 Initializing KAFKA library
@@ -312,7 +312,7 @@ Fiware-Correlator: REGEX([0-9a-f\-]{36})
 
 
 
-09. Sleep 60 seconds, get logs and see localhost:9095 connection went away
+09. Sleep 70 seconds, get logs and see localhost:9095 connection went away
 ==========================================================================
 #SORT_START
 Initializing KAFKA library