KafkaProducer: Flush pending records before close() (#2537)

Author: dpkp

kafka/producer/kafka.py

@@ -506,6 +506,7 @@ def __getattr__(self, name):
             assert timeout >= 0
 
         log.info("Closing the Kafka producer with %s secs timeout.", timeout)
+        self.flush(timeout)
         invoked_from_callback = bool(threading.current_thread() is self._sender)
         if timeout > 0:
             if invoked_from_callback: