From 5266f55317ce2659a0dac7fa48b7a202ca55145a Mon Sep 17 00:00:00 2001
From: Moshe Zadka <mzadka@surveymonkey.com>
Date: Wed, 17 Mar 2021 13:05:06 -0700
Subject: [PATCH] Flush before close: If there are any messages pending, flush
 them before doing anything else

---
 kafka/producer/kafka.py | 2 ++
 1 file changed, 2 insertions(+)

diff --git a/kafka/producer/kafka.py b/kafka/producer/kafka.py
index 1f64536b6..9e966f503 100644
--- 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()