Merge remote-tracking branch 'parsely/master' into feature/rdkafka_extension

The producer-futures feature was backed out of master, which means the
expected interface for RdKafkaProducer._produce() has changed back, too.
I've addressed all merge conflicts here - the change in the _produce()
interface will be addressed in the next commit.

* parsely/master: (26 commits)
  changelog updates for 2.0.3, dev version
  increment version
  Catch IOError in recvall_into util.
  Catch IOError during connection response.
  re-import weakref
  Revert 52ae7a1
  Link #334 to README.
  drop autocommit logging to DEBUG level. fixes #337
  update after socket error in offset manager discovery
  remove unused condition
  unconditionally update partition leaders on update
  Load all topic values on values method.
  fix typo causing interpreter error in reset_offsets`
  fix outdenting error
  clarify functools import
  catch all exceptions when removing from zookeeper
  be very specific about the error we expect
  producer: minimal changes for gc'ability
  balancedconsumer: minimal changes for gc'ability (RFC)
  add logging, fix some retry/reconnect/update logic in simpleconsumer
  ...

Signed-off-by: Yung-Chin Oei <[email protected]>

Conflicts:
	.travis.yml
	pykafka/simpleconsumer.py
	tests/pykafka/test_producer.py

Author: yungchin

.travis.yml

@@ -23,7 +23,7 @@ notifications:
     - [email protected]
 
 install:
-    - pip install codecov kazoo tox testinstances futures
+    - pip install codecov kazoo tox testinstances
     - wget https://github.com/edenhill/librdkafka/archive/0.8.6.tar.gz
     - tar -xzf 0.8.6.tar.gz
     - cd librdkafka-0.8.6/ && ./configure --prefix=$HOME

CHANGES.rst

@@ -1,6 +1,32 @@
 Changelog
 =========
 
+2.0.3 (2015-11-10)
+------------------
+
+`Compare 2.0.3`_
+
+.. _Compare 2.0.3: https://github.com/Parsely/pykafka/compare/2.0.2...bd844cd66e79b3e0a56dd92a2aae4579a9046e8e
+
+Features
+********
+
+* Raise exceptions from worker threads to the main thread in `BalancedConsumer`
+* Call `stop()` when `BalancedConsumer` is finalized to minimize zombie threads
+
+Bug Fixes
+*********
+
+* Use weak references in `BalancedConsumer` workers to avoid zombie threads creating
+  memory leaks
+* Stabilize `BalancedConsumer.start()`
+* Fix a bug in `TopicDict.values()` causing topics to be listed as `None`
+* Handle `IOError` in `BrokerConnection` and `socket.recvall_into`
+* Unconditionally update partitions' leaders after metadata requests
+* Fix thread-related memory leaks in `Producer`
+* Handle connection errors during offset commits
+* Fix an interpreter error in `SimpleConsumer`
+
 2.0.2 (2015-10-29)
 ------------------
 

README.rst

@@ -58,44 +58,10 @@ producing messages.
 
 .. sourcecode:: python
 
-    >>> with topic.get_sync_producer() as producer:
+    >>> with topic.get_producer() as producer:
     ...     for i in range(4):
     ...         producer.produce('test message ' + i ** 2)
 
-The example above would produce to kafka synchronously, that is, the call only
-returns after we have confirmation that the message made it to the cluster.
-
-To achieve higher throughput however, we recommend using the ``Producer`` in
-asynchronous mode.  In that configuration, ``produce()`` calls will return a
-``concurrent.futures.Future`` (`docs`_), which you may evaluate later (or, if
-reliable delivery is not a concern, you're free to discard it unevaluated).
-Here's a rough usage example:
-
-.. sourcecode:: python
-
-    >>> with topic.get_producer() as producer:
-    ...     count = 0
-    ...     pending = []
-    ...     while True:
-    ...         count += 1
-    ...         future = producer.produce('test message',
-    ...                                   partition_key='{}'.format(count))
-    ...         pending.append(future)
-    ...         if count % 10**5 == 0:  # adjust this or bring lots of RAM ;)
-    ...             done, not_done = concurrent.futures.wait(pending,
-                                                             timeout=.001)
-    ...             for future in done:
-    ...                 message_key = future.kafka_msg.partition_key
-    ...                 if future.exception() is not None:
-    ...                     print 'Failed to deliver message {}: {}'.format(
-    ...                         message_key, repr(future.exception()))
-    ...                 else:
-    ...                     print 'Successfully delivered message {}'.format(
-    ...                         message_key)
-    ...             pending = list(not_done)
-
-.. _docs: https://pythonhosted.org/futures/#future-objects
-
 You can also consume messages from this topic using a `Consumer` instance.
 
 .. sourcecode:: python
@@ -179,6 +145,12 @@ previous versions will always be available in this repo.
 
 .. _PyPI package: https://pypi.python.org/pypi/samsa/0.3.11
 
+pykafka or kafka-python?
+------------------------
+
+These are two different projects.
+See `the discussion here <https://github.com/Parsely/pykafka/issues/334>`_.
+
 Support
 -------
 

pykafka/__init__.py

@@ -7,7 +7,7 @@
 from .client import KafkaClient
 from .balancedconsumer import BalancedConsumer
 
-__version__ = '2.0.3-dev'
+__version__ = '2.0.4-dev'
 
 
 __all__ = ["Broker", "SimpleConsumer", "Cluster", "Partition", "Producer",

pykafka/balancedconsumer.py

@@ -18,10 +18,13 @@
 limitations under the License.
 """
 __all__ = ["BalancedConsumer"]
+import functools
 import itertools
 import logging
 import socket
+import sys
 import time
+import traceback
 from uuid import uuid4
 import weakref
 
@@ -44,6 +47,18 @@
 log = logging.getLogger(__name__)
 
 
+def _catch_thread_exception(fn):
+    """Sets self._worker_exception when fn raises an exception"""
+    def wrapped(self, *args, **kwargs):
+        try:
+            ret = fn(self, *args, **kwargs)
+        except Exception:
+            self._worker_exception = sys.exc_info()
+        else:
+            return ret
+    return wrapped
+
+
 class BalancedConsumer():
     """
     A self-balancing consumer for Kafka that uses ZooKeeper to communicate
@@ -173,6 +188,8 @@ def __init__(self,
         self._zookeeper_connection_timeout_ms = zookeeper_connection_timeout_ms
         self._reset_offset_on_start = reset_offset_on_start
         self._running = False
+        self._worker_exception = None
+        self._worker_trace_logged = False
 
         if not rdkafka and use_rdkafka:
             raise ImportError("use_rdkafka requires rdkafka to be installed")
@@ -200,6 +217,10 @@ def __init__(self,
         if auto_start is True:
             self.start()
 
+    def __del__(self):
+        log.debug("Finalising {}".format(self))
+        self.stop()
+
     def __repr__(self):
         return "<{module}.{name} at {id_} (consumer_group={group})>".format(
             module=self.__class__.__module__,
@@ -208,15 +229,33 @@ def __repr__(self):
             group=self._consumer_group
         )
 
+    def _raise_worker_exceptions(self):
+        """Raises exceptions encountered on worker threads"""
+        if self._worker_exception is not None:
+            _, ex, tb = self._worker_exception
+            if not self._worker_trace_logged:
+                self._worker_trace_logged = True
+                log.error("Exception encountered in worker thread:\n%s",
+                          "".join(traceback.format_tb(tb)))
+            raise ex
+
     def _setup_checker_worker(self):
         """Start the zookeeper partition checker thread"""
+        self = weakref.proxy(self)
+
         def checker():
             while True:
-                if not self._running:
+                try:
+                    if not self._running:
+                        break
+                    time.sleep(120)
+                    if not self._check_held_partitions():
+                        self._rebalance()
+                except Exception as e:
+                    if not isinstance(e, ReferenceError):
+                        # surface all exceptions to the main thread
+                        self._worker_exception = sys.exc_info()
                     break
-                time.sleep(120)
-                if not self._check_held_partitions():
-                    self._rebalance()
             log.debug("Checker thread exiting")
         log.debug("Starting checker thread")
         return self._cluster.handler.spawn(checker)
@@ -242,17 +281,21 @@ def held_offsets(self):
 
     def start(self):
         """Open connections and join a cluster."""
-        if self._zookeeper is None:
-            self._setup_zookeeper(self._zookeeper_connect,
-                                  self._zookeeper_connection_timeout_ms)
-        self._zookeeper.ensure_path(self._topic_path)
-        self._zk_state_listener = self._get_zk_state_listener()
-        self._zookeeper.add_listener(self._zk_state_listener)
-        self._add_self()
-        self._running = True
-        self._set_watches()
-        self._rebalance()
-        self._setup_checker_worker()
+        try:
+            if self._zookeeper is None:
+                self._setup_zookeeper(self._zookeeper_connect,
+                                      self._zookeeper_connection_timeout_ms)
+            self._zookeeper.ensure_path(self._topic_path)
+            self._zk_state_listener = self._get_zk_state_listener()
+            self._zookeeper.add_listener(self._zk_state_listener)
+            self._add_self()
+            self._running = True
+            self._set_watches()
+            self._rebalance()
+            self._setup_checker_worker()
+        except Exception:
+            log.error("Stopping consumer in response to error")
+            self.stop()
 
     def stop(self):
         """Close the zookeeper connection and stop consuming.
@@ -273,10 +316,13 @@ def stop(self):
             self._zookeeper.stop()
         else:
             self._remove_partitions(self._get_held_partitions())
-            self._zookeeper.delete(self._path_self)
-            # additionally we'd want to remove watches here, but there are no
-            # facilities for that in ChildrenWatch - as a workaround we check
-            # self._running in the watcher callbacks (see further down)
+            try:
+                self._zookeeper.delete(self._path_self)
+            except:
+                pass
+        # additionally we'd want to remove watches here, but there are no
+        # facilities for that in ChildrenWatch - as a workaround we check
+        # self._running in the watcher callbacks (see further down)
 
     def _setup_zookeeper(self, zookeeper_connect, timeout):
         """Open a connection to a ZooKeeper host.
@@ -411,13 +457,18 @@ def _set_watches(self):
         consumer group remains up-to-date with the current state of the
         cluster.
         """
+        proxy = weakref.proxy(self)
+        _brokers_changed = functools.partial(BalancedConsumer._brokers_changed, proxy)
+        _topics_changed = functools.partial(BalancedConsumer._topics_changed, proxy)
+        _consumers_changed = functools.partial(BalancedConsumer._consumers_changed, proxy)
+
         self._setting_watches = True
         # Set all our watches and then rebalance
         broker_path = '/brokers/ids'
         try:
             self._broker_watcher = ChildrenWatch(
                 self._zookeeper, broker_path,
-                self._brokers_changed
+                _brokers_changed
             )
         except NoNodeException:
             raise Exception(
@@ -428,12 +479,12 @@ def _set_watches(self):
         self._topics_watcher = ChildrenWatch(
             self._zookeeper,
             '/brokers/topics',
-            self._topics_changed
+            _topics_changed
         )
 
         self._consumer_watcher = ChildrenWatch(
             self._zookeeper, self._consumer_id_path,
-            self._consumers_changed
+            _consumers_changed
         )
         self._setting_watches = False
 
@@ -602,6 +653,7 @@ def listener(zk_state):
 
         return listener
 
+    @_catch_thread_exception
     def _brokers_changed(self, brokers):
         if not self._running:
             return False  # `False` tells ChildrenWatch to disable this watch
@@ -611,6 +663,7 @@ def _brokers_changed(self, brokers):
             self._consumer_id))
         self._rebalance()
 
+    @_catch_thread_exception
     def _consumers_changed(self, consumers):
         if not self._running:
             return False  # `False` tells ChildrenWatch to disable this watch
@@ -620,6 +673,7 @@ def _consumers_changed(self, consumers):
             self._consumer_id))
         self._rebalance()
 
+    @_catch_thread_exception
     def _topics_changed(self, topics):
         if not self._running:
             return False  # `False` tells ChildrenWatch to disable this watch
@@ -641,6 +695,7 @@ def reset_offsets(self, partition_offsets=None):
         :type partition_offsets: Iterable of
             (:class:`pykafka.partition.Partition`, int)
         """
+        self._raise_worker_exceptions()
         if not self._consumer:
             raise ConsumerStoppedException("Internal consumer is stopped")
         self._consumer.reset_offsets(partition_offsets=partition_offsets)
@@ -663,6 +718,7 @@ def consumer_timed_out():
         message = None
         self._last_message_time = time.time()
         while message is None and not consumer_timed_out():
+            self._raise_worker_exceptions()
             try:
                 message = self._consumer.consume(block=block)
             except ConsumerStoppedException:
@@ -690,4 +746,5 @@ def commit_offsets(self):
 
         Uses the offset commit/fetch API
         """
+        self._raise_worker_exceptions()
         return self._consumer.commit_offsets()

pykafka/cluster.py

@@ -44,6 +44,9 @@ def __init__(self, cluster, exclude_internal_topics, *args, **kwargs):
         self._cluster = weakref.ref(cluster)
         self._exclude_internal_topics = exclude_internal_topics
 
+    def values(self):
+        return [self[key] for key in self]
+
     def __getitem__(self, key):
         if self._should_exclude_topic(key):
             raise KeyError("You have configured KafkaClient/Cluster to hide "
@@ -259,10 +262,13 @@ def _update_brokers(self, broker_metadata):
                 try:
                     self._brokers[id_].connect()
                 except socket.error:
-                    log.info('Failed to re-establish connection with broker id %s: %s:%s', id_, meta.host, meta.port)
+                    log.info('Failed to re-establish connection with broker id %s: %s:%s',
+                             id_, meta.host, meta.port)
             else:
                 broker = self._brokers[id_]
                 if meta.host == broker.host and meta.port == broker.port:
+                    log.info('Broker %s:%s metadata unchanged. Continuing.',
+                             broker.host, broker.port)
                     continue  # no changes
                 # TODO: Can brokers update? Seems like a problem if so.
                 #       Figure out and implement update/disconnect/reconnect if
@@ -298,9 +304,10 @@ def get_offset_manager(self, consumer_group):
                 if i == MAX_RETRIES - 1:
                     raise
             except SocketDisconnectedError:
-                raise KafkaException("Socket disconnected during offset manager "
-                                     "discovery. This can happen when using PyKafka "
-                                     "with a Kafka version lower than 0.8.2.")
+                log.error("Socket disconnected during offset manager "
+                          "discovery. This can happen when using PyKafka "
+                          "with a Kafka version lower than 0.8.2.")
+                self.update()
             else:
                 coordinator = self.brokers.get(res.coordinator_id, None)
                 if coordinator is None:

pykafka/common.py

@@ -33,7 +33,7 @@ class Message(object):
     :ivar key: (optional) Message key
     :ivar offset: Message offset
     """
-    __slots__ = []
+    pass
 
 
 class CompressionType(object):

pykafka/connection.py

@@ -112,8 +112,11 @@ def response(self):
         size = bytes()
         expected_len = 4  # Size => int32
         while len(size) != expected_len:
-            r = self._socket.recv(expected_len - len(size))
-            if len(r) == 0:
+            try:
+                r = self._socket.recv(expected_len - len(size))
+            except IOError:
+                r = None
+            if r is None or len(r) == 0:
                 # Happens when broker has shut down
                 self.disconnect()
                 raise SocketDisconnectedError