Compare commits

...

7 Commits

Author SHA1 Message Date
Ivan Chvets
ea90cf4975 Merge branch 'master' into v3.1.0 2024-07-12 09:53:35 -04:00
i-chvets
5d89107827 Merge pull request #362 from Telecominfraproject/WIFI-13597-fix-kafka-producer-using-poll
WIFI-13857: fix: modified code to use flush() when internal queue is not loaded
2024-06-19 16:52:01 -04:00
Ivan Chvets
3c15c6dc4f fix: modified code to use flush() when internal queue is not loaded
https://telecominfraproject.atlassian.net/browse/WIFI-13597

Signed-off-by: Ivan Chvets <ivan.chvets@kinarasystems.com>
2024-06-19 16:29:24 -04:00
i-chvets
7b33a692b2 Merge pull request #361 from Telecominfraproject/WIFI-13597-fix-kafka-producer-using-poll
fix: added flush() for proper shutdown
2024-06-18 12:04:02 -04:00
Ivan Chvets
b118dcbcec fix: added flush() for proper shutdown
https://telecominfraproject.atlassian.net/browse/WIFI-13597

Signed-off-by: Ivan Chvets <ivan.chvets@kinarasystems.com>
2024-06-14 16:30:23 -04:00
Stephane Bourque
02a0eef44a Merge pull request #360 from Telecominfraproject/WIFI-13597-fix-kafka-producer-using-poll
WIFI-13597: fix: modified kafka manager to use poll in producer
2024-06-12 12:17:08 -07:00
Ivan Chvets
c7ed7fb264 fix: modified kafka manager to use poll in producer
https://telecominfraproject.atlassian.net/browse/WIFI-13597

Summary of changes:
- Modified code in KafkaManager to use poll instead of flush for every
  messages sent. flush is used only on empty internal notification queue
in idle times.

Signed-off-by: Ivan Chvets <ivan.chvets@kinarasystems.com>
2024-06-11 11:15:42 -04:00

View File

@@ -107,7 +107,16 @@ namespace OpenWifi {
NewMessage.partition(0);
NewMessage.payload(Msg->Payload());
Producer.produce(NewMessage);
Producer.flush();
if (Queue_.size() < 100) {
// use flush when internal queue is lightly loaded, i.e. flush after each
// message
Producer.flush();
}
else {
// use poll when internal queue is loaded to allow messages to be sent in
// batches
Producer.poll((std::chrono::milliseconds) 0);
}
}
} catch (const cppkafka::HandleException &E) {
poco_warning(Logger_,
@@ -117,8 +126,13 @@ namespace OpenWifi {
} catch (...) {
poco_error(Logger_, "std::exception");
}
if (Queue_.size() == 0) {
// message queue is empty, flush all previously sent messages
Producer.flush();
}
Note = Queue_.waitDequeueNotification();
}
Producer.flush();
poco_information(Logger_, "Stopped...");
}
@@ -324,4 +338,4 @@ namespace OpenWifi {
partitions.front().get_partition()));
}
} // namespace OpenWifi
} // namespace OpenWifi