diff --git a/app.py b/app.py
index d50ea77..ef8733a 100644
--- a/app.py
+++ b/app.py
@@ -79,7 +79,7 @@ class XProcess(Process):
             else:
                 continue
 
-            transmitter.run()
+            transmitter.run(kafka_client)
 
         while True:
             time.sleep(5)
diff --git a/v2/transmitter.py b/v2/transmitter.py
index b739e27..447b880 100644
--- a/v2/transmitter.py
+++ b/v2/transmitter.py
@@ -130,7 +130,7 @@ class Transmitter:
             for key in del_keys:
                 del item[key]
 
-    def run(self):
+    def run(self, kafka_client):
         for tb, buffer in self.check_send():
             try:
                 data = [self.flat_data(x) for x in buffer.values()]
@@ -141,3 +141,7 @@ class Transmitter:
             except Exception as e:
                 self.log.error(traceback.format_exc())
             buffer.clear()
+            try:
+                kafka_client.commit()
+            except Exception as e:
+                self.log.error(e)