From d4cf4a7e5f0cc0965aaf4747fd7bd6f6f244c529 Mon Sep 17 00:00:00 2001
From: Mike Siegel <msiegel@wayfair.com>
Date: Tue, 9 Oct 2018 14:08:02 -0400
Subject: [PATCH] forgot to flush

---
 parsedmarc/kafkaclient.py | 2 ++
 1 file changed, 2 insertions(+)

diff --git a/parsedmarc/kafkaclient.py b/parsedmarc/kafkaclient.py
index 7eaeb47..f81aff9 100644
--- a/parsedmarc/kafkaclient.py
+++ b/parsedmarc/kafkaclient.py
@@ -30,6 +30,7 @@ class KafkaClient(object):
 
         for report in aggregate_reports:
             self.producer.send(aggregate_topic, report)
+            self.producer.flush()
 
 
   def save_forensic_reports_to_kafka(self, forensic_reports, forensic_topic):
@@ -49,3 +50,4 @@ class KafkaClient(object):
 
             for report in forensic_reports:
                 self.producer.send(forensic_topic, report)
+                self.producer.flush()