Compare commits

...

7 Commits

Author SHA1 Message Date
TIP Automation User
9fc047d866 Chg: update image tag in helm values to v3.2.0-RC1 2024-09-11 17:50:58 +00:00
i-chvets
6249cb914b Merge pull request #59 from Telecominfraproject/WIFI-13857-fix-svc-kafka-startup
WIFI-13857: fix: modified code to use flush() when internal queue is not loaded
2024-06-19 16:51:32 -04:00
Ivan Chvets
574ea606c2 fix: modified code to use flush() when internal queue is not loaded
https://telecominfraproject.atlassian.net/browse/WIFI-13857

NOTE: This is port of https://github.com/Telecominfraproject/wlan-cloud-ucentralgw/pull/362

Signed-off-by: Ivan Chvets <ivan.chvets@kinarasystems.com>
2024-06-19 16:45:35 -04:00
i-chvets
a88014dab8 Merge pull request #58 from Telecominfraproject/WIFI-13597-fix-kafka-producer-using-poll
fix: added flush() for proper shutdown
2024-06-18 12:03:30 -04:00
Ivan Chvets
280ae24182 fix: added flush() for proper shutdown
https://telecominfraproject.atlassian.net/browse/WIFI-13597

Signed-off-by: Ivan Chvets <ivan.chvets@kinarasystems.com>
2024-06-18 10:46:42 -04:00
i-chvets
79e9a98725 Merge pull request #57 from Telecominfraproject/WIFI-13597-fix-kafka-producer-using-poll
fix: modified kafka manager to use poll in producer
2024-06-13 14:35:07 -04:00
Ivan Chvets
5ddcf9a2bb fix: modified kafka manager to use poll in producer
https://telecominfraproject.atlassian.net/browse/WIFI-13597

NOTE: This fix is port of https://github.com/Telecominfraproject/wlan-cloud-ucentralgw/pull/360

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-13 13:20:29 -04:00
2 changed files with 17 additions and 3 deletions

View File

@@ -9,7 +9,7 @@ fullnameOverride: ""
images:
owanalytics:
repository: tip-tip-wlan-cloud-ucentral.jfrog.io/owanalytics
tag: main
tag: v3.2.0-RC1
pullPolicy: Always
# regcred:
# registry: tip-tip-wlan-cloud-ucentral.jfrog.io

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