diff --git a/src/framework/KafkaManager.cpp b/src/framework/KafkaManager.cpp index d2d30dff..d9036529 100644 --- a/src/framework/KafkaManager.cpp +++ b/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 MsgVec; while (Running_) { try { - std::vector 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_,