Flush before close: If there are any messages pending, flush them bef…

…ore doing anything else
dpkp · Mar 17, 2021 · 5266f55 · 5266f55
1 parent f19e423
commit 5266f55
Showing 1 changed file with 2 additions and 0 deletions.
diff --git a/kafka/producer/kafka.py b/kafka/producer/kafka.py
@@ -469,6 +469,8 @@ def close(self, timeout=None):
             timeout (float, optional): timeout in seconds to wait for completion.
         """
 
+        # If there are any pending messages, send them now
+        self.flush(timeout)
         # drop our atexit handler now to avoid leaks
         self._unregister_cleanup()