Repository: dpkp/kafka-python

Issue:
Not catching exceptions/stuck on 'localhost:9092'
I am not getting any exception will creating the producer
- running pytest in a docker container on ubuntu
- the server is down to test bad connectivity

Here is a **successful** error handling when connecting to **127.0.0.1:9092**:

> [kafka.producer.kafka] Starting the Kafka producer
> [kafka.client] Bootstrapping cluster metadata from [('127.0.0.1', 9092, <AddressFamily.AF_INET: 2>)]
> [kafka.client] Attempting to bootstrap via node at 127.0.0.1:9092
> [kafka.conn] <BrokerConnection node_id=bootstrap host=127.0.0.1:9092 <disconnected> [IPv4 None]>: creating new socket
> [kafka.conn] <BrokerConnection node_id=bootstrap host=127.0.0.1:9092 <disconnected> [IPv4 ('127.0.0.1', 9092)]>: setting socket option (6, 1, 1)
> [kafka.conn] <BrokerConnection node_id=bootstrap host=127.0.0.1:9092 <connecting> [IPv4 ('127.0.0.1', 9092)]>: connecting to 127.0.0.1:9092 [('127.0.0.1', 9092) IPv4]
> [kafka.conn] Connect attempt to <BrokerConnection node_id=bootstrap host=127.0.0.1:9092 <connecting> [IPv4 ('127.0.0.1', 9092)]> returned error 111. Disconnecting.
> [kafka.conn] <BrokerConnection node_id=bootstrap host=127.0.0.1:9092 <connecting> [IPv4 ('127.0.0.1', 9092)]>: Closing connection. KafkaConnectionError: 111 ECONNREFUSED
> [kafka.conn] <BrokerConnection node_id=bootstrap host=127.0.0.1:9092 <disconnecting> [IPv4 ('127.0.0.1', 9092)]>: reconnect backoff 0.05721221285863193 after 1 failures
> [kafka.client] Unable to bootstrap from [('127.0.0.1', 9092, <AddressFamily.AF_INET: 2>)]
NoBrokersAvailable
> [kafka.producer.kafka] Kafka producer closed


Here is the same code stuck on **localhost:9092**, not catching anything or even trying to reconnect:

> [kafka.producer.kafka] Starting the Kafka producer
> [kafka.client] Bootstrapping cluster metadata from [('localhost', 9092, <AddressFamily.AF_UNSPEC: 0>)]
> [kafka.client] Attempting to bootstrap via node at localhost:9092
> [kafka.conn] <BrokerConnection node_id=bootstrap host=localhost:9092 <disconnected> [unspecified None]>: creating new socket
> [kafka.conn] <BrokerConnection node_id=bootstrap host=localhost:9092 <disconnected> [IPv4 ('127.0.0.1', 9092)]>: setting socket option (6, 1, 1)
> [kafka.conn] <BrokerConnection node_id=bootstrap host=localhost:9092 <connecting> [IPv4 ('127.0.0.1', 9092)]>: connecting to localhost:9092 [('127.0.0.1', 9092) IPv4]
> [kafka.conn] Connect attempt to <BrokerConnection node_id=bootstrap host=localhost:9092 <connecting> [IPv4 ('127.0.0.1', 9092)]> returned error 111. Disconnecting.
> [kafka.conn] <BrokerConnection node_id=bootstrap host=localhost:9092 <connecting> [IPv4 ('127.0.0.1', 9092)]>: Closing connection. KafkaConnectionError: 111 ECONNREFUSED

nothing else happens after that, it waits and don't try to reconnect.

it likes 127.0.0.1:9092 but can not handle localhost:9092, is there a reason?

Discussion:

Patch:
diff --git a/kafka/client_async.py b/kafka/client_async.py
index d608e6a..fdf5454 100644
--- a/kafka/client_async.py
+++ b/kafka/client_async.py
@@ -5,7 +5,9 @@ import copy
 import functools
 import logging
 import random
+import socket
 import threading
+import time
 import weakref
 
 # selectors in stdlib as of py3.4
@@ -15,9 +17,6 @@ except ImportError:
     # vendored backport module
     from kafka.vendor import selectors34 as selectors
 
-import socket
-import time
-
 from kafka.vendor import six
 
 from kafka.cluster import ClusterMetadata
@@ -56,7 +55,7 @@ class KafkaClient(object):
 
     Keyword Arguments:
         bootstrap_servers: 'host[:port]' string (or list of 'host[:port]'
-            strings) that the consumer should contact to bootstrap initial
+            strings) that the client should contact to bootstrap initial
             cluster metadata. This does not have to be the full node list.
             It just needs to have at least one broker that will respond to a
             Metadata API Request. Default port is 9092. If no servers are
@@ -223,76 +222,34 @@ class KafkaClient(object):
                                                self.config['metric_group_prefix'],
                                                weakref.proxy(self._conns))
 
-        self._bootstrap(collect_hosts(self.config['bootstrap_servers']))
+        self._num_bootstrap_hosts = len(collect_hosts(self.config['bootstrap_servers']))
 
         # Check Broker Version if not set explicitly
         if self.config['api_version'] is None:
             check_timeout = self.config['api_version_auto_timeout_ms'] / 1000
             self.config['api_version'] = self.check_version(timeout=check_timeout)
 
-    def _bootstrap(self, hosts):
-        log.info('Bootstrapping cluster metadata from %s', hosts)
-        # Exponential backoff if bootstrap fails
-        backoff_ms = self.config['reconnect_backoff_ms'] * 2 ** self._bootstrap_fails
+    def _can_bootstrap(self):
+        effective_failures = self._bootstrap_fails // self._num_bootstrap_hosts
+        backoff_factor = 2 ** effective_failures
+        backoff_ms = min(self.config['reconnect_backoff_ms'] * backoff_factor,
+                         self.config['reconnect_backoff_max_ms'])
+
+        backoff_ms *= random.uniform(0.8, 1.2)
+
         next_at = self._last_bootstrap + backoff_ms / 1000.0
-        self._refresh_on_disconnects = False
         now = time.time()
         if next_at > now:
-            log.debug("Sleeping %0.4f before bootstrapping again", next_at - now)
-            time.sleep(next_at - now)
-        self._last_bootstrap = time.time()
-
-        if self.config['api_version'] is None or self.config['api_version'] < (0, 10):
-            if self.config['bootstrap_topics_filter']:
-                metadata_request = MetadataRequest[0](list(self.config['bootstrap_topics_filter']))
-            else:
-                metadata_request = MetadataRequest[0]([])
-        else:
-            if self.config['bootstrap_topics_filter']:
-                metadata_request = MetadataRequest[1](list(self.config['bootstrap_topics_filter']))
-            else:
-                metadata_request = MetadataRequest[1](None)
-
-        for host, port, afi in hosts:
-            log.debug("Attempting to bootstrap via node at %s:%s", host, port)
-            cb = functools.partial(WeakMethod(self._conn_state_change), 'bootstrap')
-            bootstrap = BrokerConnection(host, port, afi,
-                                         state_change_callback=cb,
-                                         node_id='bootstrap',
-                                         **self.config)
-            if not bootstrap.connect_blocking():
-                bootstrap.close()
-                continue
-            future = bootstrap.send(metadata_request)
-            while not future.is_done:
-                self._selector.select(1)
-                for r, f in bootstrap.recv():
-                    f.success(r)
-            if future.failed():
-                bootstrap.close()
-                continue
-            self.cluster.update_metadata(future.value)
-            log.info('Bootstrap succeeded: found %d brokers and %d topics.',
-                     len(self.cluster.brokers()), len(self.cluster.topics()))
-
-            # A cluster with no topics can return no broker metadata
-            # in that case, we should keep the bootstrap connection
-            if not len(self.cluster.brokers()):
-                self._conns['bootstrap'] = bootstrap
-            else:
-                bootstrap.close()
-            self._bootstrap_fails = 0
-            break
-        # No bootstrap found...
-        else:
-            log.error('Unable to bootstrap from %s', hosts)
-            # Max exponential backoff is 2^12, x4000 (50ms -> 200s)
-            self._bootstrap_fails = min(self._bootstrap_fails + 1, 12)
-        self._refresh_on_disconnects = True
+            return False
+        return True
 
     def _can_connect(self, node_id):
         if node_id not in self._conns:
-            if self.cluster.broker_metadata(node_id):
+            # cluster.broker_metadata() is stateful when called w/ 'bootstrap'
+            # (it cycles through all of the bootstrap servers)
+            # so we short-circuit here and assume that we should always have
+            # some bootstrap_servers config to power bootstrap broker_metadata
+            if node_id == 'bootstrap' or self.cluster.broker_metadata(node_id):
                 return True
             return False
         conn = self._conns[node_id]
@@ -309,6 +266,9 @@ class KafkaClient(object):
                 except KeyError:
                     self._selector.modify(conn._sock, selectors.EVENT_WRITE)
 
+                if node_id == 'bootstrap':
+                    self._last_bootstrap = time.time()
+
             elif conn.connected():
                 log.debug("Node %s connected", node_id)
                 if node_id in self._connecting:
@@ -324,12 +284,12 @@ class KafkaClient(object):
 
                 self._idle_expiry_manager.update(node_id)
 
-                if 'bootstrap' in self._conns and node_id != 'bootstrap':
+                if node_id == 'bootstrap':
+                    self._bootstrap_fails = 0
+
+                elif 'bootstrap' in self._conns:
                     bootstrap = self._conns.pop('bootstrap')
-                    # XXX: make conn.close() require error to cause refresh
-                    self._refresh_on_disconnects = False
                     bootstrap.close()
-                    self._refresh_on_disconnects = True
 
             # Connection failures imply that our metadata is stale, so let's refresh
             elif conn.state is ConnectionStates.DISCONNECTING:
@@ -348,7 +308,10 @@ class KafkaClient(object):
                     idle_disconnect = True
                 self._idle_expiry_manager.remove(node_id)
 
-                if self._refresh_on_disconnects and not self._closed and not idle_disconnect:
+                if node_id == 'bootstrap':
+                    self._bootstrap_fails += 1
+
+                elif self._refresh_on_disconnects and not self._closed and not idle_disconnect:
                     log.warning("Node %s connection failed -- refreshing metadata", node_id)
                     self.cluster.request_update()
 
@@ -363,13 +326,40 @@ class KafkaClient(object):
             return True
         return False
 
+    def _should_recycle_connection(self, conn):
+        # Never recycle unless disconnected
+        if not conn.disconnected():
+            return False
+
+        # Always recycled disconnected bootstraps
+        elif conn.node_id == 'bootstrap':
+            return True
+
+        # Otherwise, only recycle when broker metadata has changed
+        broker = self.cluster.broker_metadata(conn.node_id)
+        if broker is None:
+            return False
+
+        host, _, afi = get_ip_port_afi(broker.host)
+        if conn.host != host or conn.port != broker.port:
+            log.info("Broker metadata change detected for node %s"
+                     " from %s:%s to %s:%s", conn.node_id, conn.host, conn.port,
+                     broker.host, broker.port)
+            return True
+
+        return False
+
     def _maybe_connect(self, node_id):
         """Idempotent non-blocking connection attempt to the given node id."""
         with self._lock:
-            broker = self.cluster.broker_metadata(node_id)
             conn = self._conns.get(node_id)
 
             if conn is None:
+                # Note that when bootstrapping, each call to broker_metadata may
+                # return a different host/port. So we need to be careful to only
+                # call when necessary to avoid skipping some possible bootstrap
+                # source.
+                broker = self.cluster.broker_metadata(node_id)
                 assert broker, 'Broker id %s not in current metadata' % (node_id,)
 
                 log.debug("Initiating connection to node %s at %s:%s",
@@ -383,17 +373,9 @@ class KafkaClient(object):
                 self._conns[node_id] = conn
 
             # Check if existing connection should be recreated because host/port changed
-            elif conn.disconnected() and broker is not None:
-                host, _, __ = get_ip_port_afi(broker.host)
-                if conn.host != host or conn.port != broker.port:
-                    log.info("Broker metadata change detected for node %s"
-                             " from %s:%s to %s:%s", node_id, conn.host, conn.port,
-                             broker.host, broker.port)
-
-                    # Drop old connection object.
-                    # It will be recreated on next _maybe_connect
-                    self._conns.pop(node_id)
-                    return False
+            elif self._should_recycle_connection(conn):
+                self._conns.pop(node_id)
+                return False
 
             elif conn.connected():
                 return True
@@ -611,7 +593,8 @@ class KafkaClient(object):
         return responses
 
     def _poll(self, timeout):
-        """Returns list of (response, future) tuples"""
+        # This needs to be locked, but since it is only called from within the
+        # locked section of poll(), there is no additional lock acquisition here
         processed = set()
 
         start_select = time.time()
@@ -713,7 +696,8 @@ class KafkaClient(object):
         This method will prefer a node with an existing connection and no
         in-flight-requests. If no such node is found, a node will be chosen
         randomly from disconnected nodes that are not "blacked out" (i.e.,
-        are not subject to a reconnect backoff).
+        are not subject to a reconnect backoff). If no node metadata has been
+        obtained, will return 'bootstrap' (subject to exponential backoff).
 
         Returns:
             node_id or None if no suitable node was found
@@ -740,12 +724,8 @@ class KafkaClient(object):
         if found is not None:
             return found
 
-        # some broker versions return an empty list of broker metadata
-        # if there are no topics created yet. the bootstrap process
-        # should detect this and keep a 'bootstrap' node alive until
-        # a non-bootstrap node is connected and non-empty broker
-        # metadata is available
-        elif 'bootstrap' in self._conns:
+        elif not nodes and self._can_bootstrap():
+            self._last_bootstrap = time.time()
             return 'bootstrap'
 
         return None
@@ -805,6 +785,9 @@ class KafkaClient(object):
 
         if self._can_send_request(node_id):
             topics = list(self._topics)
+            if not topics and node_id == 'bootstrap':
+                topics = list(self.config['bootstrap_topics_filter'])
+
             if self.cluster.need_all_topic_metadata or not topics:
                 topics = [] if self.config['api_version'] < (0, 10) else None
             api_version = 0 if self.config['api_version'] < (0, 10) else 1
@@ -825,9 +808,7 @@ class KafkaClient(object):
         # the client from unnecessarily connecting to additional nodes while a previous connection
         # attempt has not been completed.
         if self._connecting:
-            # Strictly the timeout we should return here is "connect timeout", but as we don't
-            # have such application level configuration, using request timeout instead.
-            return self.config['request_timeout_ms']
+            return self.config['reconnect_backoff_ms']
 
         if self.maybe_connect(node_id):
             log.debug("Initializing connection to node %s for metadata request", node_id)
diff --git a/kafka/cluster.py b/kafka/cluster.py
index 8078eb7..3d57ed2 100644
--- a/kafka/cluster.py
+++ b/kafka/cluster.py
@@ -9,6 +9,7 @@ import time
 from kafka.vendor import six
 
 from kafka import errors as Errors
+from kafka.conn import collect_hosts, dns_lookup
 from kafka.future import Future
 from kafka.structs import BrokerMetadata, PartitionMetadata, TopicPartition
 
@@ -29,10 +30,17 @@ class ClusterMetadata(object):
             which we force a refresh of metadata even if we haven't seen any
             partition leadership changes to proactively discover any new
             brokers or partitions. Default: 300000
+        bootstrap_servers: 'host[:port]' string (or list of 'host[:port]'
+            strings) that the client should contact to bootstrap initial
+            cluster metadata. This does not have to be the full node list.
+            It just needs to have at least one broker that will respond to a
+            Metadata API Request. Default port is 9092. If no servers are
+            specified, will default to localhost:9092.
     """
     DEFAULT_CONFIG = {
         'retry_backoff_ms': 100,
         'metadata_max_age_ms': 300000,
+        'bootstrap_servers': 'localhost',
     }
 
     def __init__(self, **configs):
@@ -42,7 +50,7 @@ class ClusterMetadata(object):
         self._groups = {}  # group_name -> node_id
         self._last_refresh_ms = 0
         self._last_successful_refresh_ms = 0
-        self._need_update = False
+        self._need_update = True
         self._future = None
         self._listeners = set()
         self._lock = threading.Lock()
@@ -56,6 +64,17 @@ class ClusterMetadata(object):
             if key in configs:
                 self.config[key] = configs[key]
 
+        self._bootstrap_brokers = self._generate_bootstrap_brokers()
+
+    def _generate_bootstrap_brokers(self):
+        # collect_hosts does not perform DNS, so we should be fine to re-use
+        bootstrap_hosts = collect_hosts(self.config['bootstrap_servers'])
+
+        while True:
+            for host, port, afi in bootstrap_hosts:
+                for _, __, ___, ____, sockaddr in dns_lookup(host, port, afi):
+                    yield BrokerMetadata('bootstrap', sockaddr[0], sockaddr[1], None)
+
     def brokers(self):
         """Get all BrokerMetadata
 
@@ -73,6 +92,9 @@ class ClusterMetadata(object):
         Returns:
             BrokerMetadata or None if not found
         """
+        if broker_id == 'bootstrap':
+            return next(self._bootstrap_brokers)
+
         return self._brokers.get(broker_id)
 
     def partitions_for_topic(self, topic):
diff --git a/kafka/conn.py b/kafka/conn.py
index 4f324c8..e857d0a 100644
--- a/kafka/conn.py
+++ b/kafka/conn.py
@@ -230,6 +230,9 @@ class BrokerConnection(object):
 
         self.node_id = self.config.pop('node_id')
 
+        if self.config['api_version'] is None:
+            self.config['api_version'] = self.DEFAULT_CONFIG['api_version']
+
         if self.config['receive_buffer_bytes'] is not None:
             self.config['socket_options'].append(
                 (socket.SOL_SOCKET, socket.SO_RCVBUF,
@@ -912,6 +915,9 @@ class BrokerConnection(object):
         return self._api_versions
 
     def get_api_versions(self):
+        if self._api_versions is not None:
+            return self._api_versions
+
         version = self.check_version()
         if version < (0, 10, 0):
             raise Errors.UnsupportedVersionError(


Test patch:
diff --git a/test/test_client_async.py b/test/test_client_async.py
index 1c8a50f..3588423 100644
--- a/test/test_client_async.py
+++ b/test/test_client_async.py
@@ -23,58 +23,34 @@ from kafka.structs import BrokerMetadata
 
 
 @pytest.fixture
-def cli(conn):
-    return KafkaClient(api_version=(0, 9))
-
-
-@pytest.mark.parametrize("bootstrap,expected_hosts", [
-    (None, [('localhost', 9092, socket.AF_UNSPEC)]),
-    ('foobar:1234', [('foobar', 1234, socket.AF_UNSPEC)]),
-    ('fizzbuzz', [('fizzbuzz', 9092, socket.AF_UNSPEC)]),
-    ('foo:12,bar:34', [('foo', 12, socket.AF_UNSPEC), ('bar', 34, socket.AF_UNSPEC)]),
-    (['fizz:56', 'buzz'], [('fizz', 56, socket.AF_UNSPEC), ('buzz', 9092, socket.AF_UNSPEC)]),
-])
-def test_bootstrap_servers(mocker, bootstrap, expected_hosts):
-    mocker.patch.object(KafkaClient, '_bootstrap')
-    if bootstrap is None:
-        KafkaClient(api_version=(0, 9)) # pass api_version to skip auto version checks
-    else:
-        KafkaClient(bootstrap_servers=bootstrap, api_version=(0, 9))
-
-    # host order is randomized internally, so resort before testing
-    (hosts,), _ = KafkaClient._bootstrap.call_args  # pylint: disable=no-member
-    assert sorted(hosts) == sorted(expected_hosts)
+def cli(mocker, conn):
+    mocker.patch('kafka.cluster.dns_lookup',
+                 return_value=[(socket.AF_INET, None, None, None, ('localhost', 9092))])
+    client = KafkaClient(api_version=(0, 9))
+    client.poll(future=client.cluster.request_update())
+    return client
 
 
-def test_bootstrap_success(conn):
+def test_bootstrap(mocker, conn):
     conn.state = ConnectionStates.CONNECTED
+    mocker.patch('kafka.cluster.dns_lookup',
+                 return_value=[(socket.AF_INET, None, None, None, ('localhost', 9092))])
     cli = KafkaClient(api_version=(0, 9))
+    future = cli.cluster.request_update()
+    cli.poll(future=future)
+
+    assert future.succeeded()
     args, kwargs = conn.call_args
     assert args == ('localhost', 9092, socket.AF_UNSPEC)
     kwargs.pop('state_change_callback')
     kwargs.pop('node_id')
     assert kwargs == cli.config
-    conn.connect_blocking.assert_called_with()
-    conn.send.assert_called_once_with(MetadataRequest[0]([]))
+    conn.send.assert_called_once_with(MetadataRequest[0]([]), blocking=False)
     assert cli._bootstrap_fails == 0
     assert cli.cluster.brokers() == set([BrokerMetadata(0, 'foo', 12, None),
                                          BrokerMetadata(1, 'bar', 34, None)])
 
 
-def test_bootstrap_failure(conn):
-    conn.connect_blocking.return_value = False
-    cli = KafkaClient(api_version=(0, 9))
-    args, kwargs = conn.call_args
-    assert args == ('localhost', 9092, socket.AF_UNSPEC)
-    kwargs.pop('state_change_callback')
-    kwargs.pop('node_id')
-    assert kwargs == cli.config
-    conn.connect_blocking.assert_called_with()
-    conn.close.assert_called_with()
-    assert cli._bootstrap_fails == 1
-    assert cli.cluster.brokers() == set()
-
-
 def test_can_connect(cli, conn):
     # Node is not in broker metadata - can't connect
     assert not cli._can_connect(2)
@@ -187,22 +163,26 @@ def test_is_ready(mocker, cli, conn):
 def test_close(mocker, cli, conn):
     mocker.patch.object(cli, '_selector')
 
-    # bootstrap connection should have been closed
-    assert conn.close.call_count == 1
+    call_count = conn.close.call_count
 
     # Unknown node - silent
     cli.close(2)
+    call_count += 0
+    assert conn.close.call_count == call_count
 
     # Single node close
     cli._maybe_connect(0)
-    assert conn.close.call_count == 1
+    assert conn.close.call_count == call_count
     cli.close(0)
-    assert conn.close.call_count == 2
+    call_count += 1
+    assert conn.close.call_count == call_count
 
     # All node close
     cli._maybe_connect(1)
     cli.close()
-    assert conn.close.call_count == 4
+    # +3 close: node 0, node 1, node bootstrap
+    call_count += 3
+    assert conn.close.call_count == call_count
 
 
 def test_is_disconnected(cli, conn):
@@ -249,7 +229,6 @@ def test_send(cli, conn):
 
 
 def test_poll(mocker):
-    mocker.patch.object(KafkaClient, '_bootstrap')
     metadata = mocker.patch.object(KafkaClient, '_maybe_refresh_metadata')
     _poll = mocker.patch.object(KafkaClient, '_poll')
     cli = KafkaClient(api_version=(0, 9))
@@ -309,7 +288,6 @@ def test_set_topics(mocker):
 
 @pytest.fixture
 def client(mocker):
-    mocker.patch.object(KafkaClient, '_bootstrap')
     _poll = mocker.patch.object(KafkaClient, '_poll')
 
     cli = KafkaClient(request_timeout_ms=9999999,
@@ -376,7 +354,7 @@ def test_maybe_refresh_metadata_cant_send(mocker, client):
     client._connecting.add('foobar')
     client._can_connect.reset_mock()
     client.poll(timeout_ms=12345678)
-    client._poll.assert_called_with(9999.999) # connection timeout (request timeout)
+    client._poll.assert_called_with(2.222) # connection timeout (reconnect timeout)
     assert not client._can_connect.called
 
     assert not client._metadata_refresh_in_progress
diff --git a/test/test_conn.py b/test/test_conn.py
index 953c112..66b8a0c 100644
--- a/test/test_conn.py
+++ b/test/test_conn.py
@@ -16,6 +16,13 @@ from kafka.protocol.produce import ProduceRequest
 import kafka.errors as Errors
 
 
+@pytest.fixture
+def dns_lookup(mocker):
+    return mocker.patch('kafka.conn.dns_lookup',
+                        return_value=[(socket.AF_INET,
+                                       None, None, None,
+                                       ('localhost', 9092))])
+
 @pytest.fixture
 def _socket(mocker):
     socket = mocker.MagicMock()
@@ -25,7 +32,7 @@ def _socket(mocker):
 
 
 @pytest.fixture
-def conn(_socket):
+def conn(_socket, dns_lookup):
     conn = BrokerConnection('localhost', 9092, socket.AF_INET)
     return conn