https://telecominfraproject.atlassian.net/browse/WIFI-12954

Signed-off-by: stephb9959 <[email protected]>

src/framework/KafkaManager.cpp

@@ -160,17 +160,19 @@ namespace OpenWifi {
 		});
 
 		bool AutoCommit = MicroServiceConfigGetBool("openwifi.kafka.auto.commit", false);
-		auto BatchSize = MicroServiceConfigGetInt("openwifi.kafka.consumer.batchsize", 20);
+		auto BatchSize = MicroServiceConfigGetInt("openwifi.kafka.consumer.batchsize", 100);
 
 		Types::StringVec Topics;
 		KafkaManager()->Topics(Topics);
 		Consumer.subscribe(Topics);
 
 		Running_ = true;
+		std::vector<cppkafka::Message> MsgVec;
 		while (Running_) {
 			try {
-				std::vector<cppkafka::Message> MsgVec =
-					Consumer.poll_batch(BatchSize, std::chrono::milliseconds(100));
+				MsgVec.clear();
+				MsgVec.reserve(BatchSize);
+				MsgVec = Consumer.poll_batch(BatchSize, std::chrono::milliseconds(1000));
 				for (auto const &Msg : MsgVec) {
 					if (!Msg)
 						continue;
@@ -180,12 +182,12 @@ namespace OpenWifi {
 									   fmt::format("Error: {}", Msg.get_error().to_string()));
 						}
 						if (!AutoCommit)
-							Consumer.async_commit(Msg);
+							Consumer.commit(Msg);
 						continue;
 					}
 					KafkaManager()->Dispatch(Msg.get_topic().c_str(), Msg.get_key(), Msg.get_payload());
 					if (!AutoCommit)
-						Consumer.async_commit(Msg);
+						Consumer.commit(Msg);
 				}
 			} catch (const cppkafka::HandleException &E) {
 				poco_warning(Logger_,