set socket timeout for the wake_w

Author: flaneur2020

kafka/client_async.py

@@ -151,6 +151,7 @@ class KafkaClient(object):
         'bootstrap_servers': 'localhost',
         'client_id': 'kafka-python-' + __version__,
         'request_timeout_ms': 30000,
+        'max_block_ms': 60000,
         'connections_max_idle_ms': 9 * 60 * 1000,
         'reconnect_backoff_ms': 50,
         'reconnect_backoff_max_ms': 1000,
@@ -198,6 +199,7 @@ def __init__(self, **configs):
         self._bootstrap_fails = 0
         self._wake_r, self._wake_w = socket.socketpair()
         self._wake_r.setblocking(False)
+        self._wake_w.settimeout(self.config['max_block_ms'] / 1000.0)
         self._wake_lock = threading.Lock()
 
         self._lock = threading.RLock()

kafka/producer/sender.py

@@ -29,6 +29,7 @@ class Sender(threading.Thread):
         'acks': 1,
         'retries': 0,
         'request_timeout_ms': 30000,
+        'max_block_ms': 60000,
         'guarantee_message_order': False,
         'client_id': 'kafka-python-' + __version__,
         'api_version': (0, 8, 0),