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

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

build

@@ -1 +1 @@
-86
+87

src/GWKafkaEvents.h

@@ -50,12 +50,22 @@ namespace OpenWifi {
 	class DeviceConfigurationChangeKafkaEvent : public GWKafkaEvents {
 	  public:
 		DeviceConfigurationChangeKafkaEvent(std::uint64_t serialNumber,
-											std::uint64_t timestamp, const Poco::JSON::Object::Ptr config)
+											std::uint64_t timestamp,
+											const Poco::JSON::Object::Ptr config)
 			: GWKafkaEvents(serialNumber, "unit.configuration_change", timestamp), config_(config) {
 		}
 
 		~DeviceConfigurationChangeKafkaEvent() {
-			payload_->set("configuration", *config_);
+			if(config_!= nullptr) {
+				std::ostringstream os;
+				config_->stringify(os);
+				if(os.str().size()> KafkaManager()->KafkaManagerMaximumPayloadSize()) {
+					payload_->set("configuration", "{}");
+					payload_->set("configurationTooBig", true);
+				} else {
+					payload_->set("configuration", *config_);
+				}
+			}
 			Send();
 		}
 

src/framework/KafkaManager.cpp

@@ -79,8 +79,10 @@ namespace OpenWifi {
 		Utils::SetThreadName("Kafka:Prod");
 		cppkafka::Configuration Config(
 			{{"client.id", MicroServiceConfigGetString("openwifi.kafka.client.id", "")},
-			 {"metadata.broker.list",
-			  MicroServiceConfigGetString("openwifi.kafka.brokerlist", "")}});
+			 {"metadata.broker.list",MicroServiceConfigGetString("openwifi.kafka.brokerlist", "")},
+			 {"send.buffer.bytes", KafkaManager()->KafkaManagerMaximumPayloadSize() }
+			}
+ 		);
 
 		AddKafkaSecurity(Config);
 
@@ -275,6 +277,7 @@ namespace OpenWifi {
 	int KafkaManager::Start() {
 		if (!KafkaEnabled_)
 			return 0;
+		MaxPayloadSize_ = MicroServiceConfigGetInt("openwifi.kafka.max.payload", 2500000);
 		ConsumerThr_.Start();
 		ProducerThr_.Start();
 		return 0;

src/framework/KafkaManager.h

@@ -94,11 +94,14 @@ namespace OpenWifi {
 			return ConsumerThr_.UnregisterTopicWatcher(Topic,Id);
 		}
 
+		std::uint64_t KafkaManagerMaximumPayloadSize() const { return MaxPayloadSize_; }
+
 	  private:
 		bool KafkaEnabled_ = false;
 		std::string SystemInfoWrapper_;
 		KafkaProducer ProducerThr_;
 		KafkaConsumer ConsumerThr_;
+		std::uint64_t MaxPayloadSize_ = 2500000;
 
 		void PartitionAssignment(const cppkafka::TopicPartitionList &partitions);
 		void PartitionRevocation(const cppkafka::TopicPartitionList &partitions);