Repository: dpkp/kafka-python

Issue:
KAFKA-3117: handle metadata updates during consumer rebalance

Discussion:

Patch:
diff --git a/kafka/consumer/group.py b/kafka/consumer/group.py
index 9ebfe02..db0022d 100644
--- a/kafka/consumer/group.py
+++ b/kafka/consumer/group.py
@@ -176,6 +176,10 @@ class KafkaConsumer(six.Iterator):
         selector (selectors.BaseSelector): Provide a specific selector
             implementation to use for I/O multiplexing.
             Default: selectors.DefaultSelector
+        exclude_internal_topics (bool): Whether records from internal topics
+            (such as offsets) should be exposed to the consumer. If set to True
+            the only way to receive records from an internal topic is
+            subscribing to it. Requires 0.10+ Default: True
 
     Note:
         Configuration parameters are described in more detail at
@@ -222,6 +226,7 @@ class KafkaConsumer(six.Iterator):
         'metrics_num_samples': 2,
         'metrics_sample_window_ms': 30000,
         'selector': selectors.DefaultSelector,
+        'exclude_internal_topics': True,
     }
 
     def __init__(self, *topics, **configs):
diff --git a/kafka/coordinator/consumer.py b/kafka/coordinator/consumer.py
index 083a36a..a18329c 100644
--- a/kafka/coordinator/consumer.py
+++ b/kafka/coordinator/consumer.py
@@ -36,6 +36,7 @@ class ConsumerCoordinator(BaseCoordinator):
         'heartbeat_interval_ms': 3000,
         'retry_backoff_ms': 100,
         'api_version': (0, 9),
+        'exclude_internal_topics': True,
     }
 
     def __init__(self, client, subscription, metrics, metric_group_prefix,
@@ -70,6 +71,10 @@ class ConsumerCoordinator(BaseCoordinator):
                 using Kafka's group managementment facilities. Default: 30000
             retry_backoff_ms (int): Milliseconds to backoff when retrying on
                 errors. Default: 100.
+            exclude_internal_topics (bool): Whether records from internal topics
+                (such as offsets) should be exposed to the consumer. If set to
+                True the only way to receive records from an internal topic is
+                subscribing to it. Requires 0.10+. Default: True
         """
         super(ConsumerCoordinator, self).__init__(client, **configs)
         self.config = copy.copy(self.DEFAULT_CONFIG)
@@ -81,7 +86,8 @@ class ConsumerCoordinator(BaseCoordinator):
             assert self.config['assignors'], 'Coordinator requires assignors'
 
         self._subscription = subscription
-        self._partitions_per_topic = {}
+        self._metadata_snapshot = {}
+        self._assignment_snapshot = None
         self._cluster = client.cluster
         self._cluster.request_update()
         self._cluster.add_listener(WeakMethod(self._handle_metadata_update))
@@ -131,13 +137,12 @@ class ConsumerCoordinator(BaseCoordinator):
 
     def _handle_metadata_update(self, cluster):
         # if we encounter any unauthorized topics, raise an exception
-        # TODO
-        #if self._cluster.unauthorized_topics:
-        #    raise TopicAuthorizationError(self._cluster.unauthorized_topics)
+        if cluster.unauthorized_topics:
+            raise Errors.TopicAuthorizationFailedError(cluster.unauthorized_topics)
 
         if self._subscription.subscribed_pattern:
             topics = []
-            for topic in cluster.topics():
+            for topic in cluster.topics(self.config['exclude_internal_topics']):
                 if self._subscription.subscribed_pattern.match(topic):
                     topics.append(topic)
 
@@ -146,7 +151,7 @@ class ConsumerCoordinator(BaseCoordinator):
 
         # check if there are any changes to the metadata which should trigger
         # a rebalance
-        if self._subscription_metadata_changed():
+        if self._subscription_metadata_changed(cluster):
 
             if (self.config['api_version'] >= (0, 9)
                 and self.config['group_id'] is not None):
@@ -159,20 +164,20 @@ class ConsumerCoordinator(BaseCoordinator):
                 self._subscription.assign_from_subscribed([
                     TopicPartition(topic, partition)
                     for topic in self._subscription.subscription
-                    for partition in self._partitions_per_topic[topic]
+                    for partition in self._metadata_snapshot[topic]
                 ])
 
-    def _subscription_metadata_changed(self):
+    def _subscription_metadata_changed(self, cluster):
         if not self._subscription.partitions_auto_assigned():
             return False
 
-        old_partitions_per_topic = self._partitions_per_topic
-        self._partitions_per_topic = {}
+        metadata_snapshot = {}
         for topic in self._subscription.group_subscription():
-            partitions = self._cluster.partitions_for_topic(topic) or []
-            self._partitions_per_topic[topic] = set(partitions)
+            partitions = cluster.partitions_for_topic(topic) or []
+            metadata_snapshot[topic] = set(partitions)
 
-        if self._partitions_per_topic != old_partitions_per_topic:
+        if self._metadata_snapshot != metadata_snapshot:
+            self._metadata_snapshot = metadata_snapshot
             return True
         return False
 
@@ -184,8 +189,15 @@ class ConsumerCoordinator(BaseCoordinator):
 
     def _on_join_complete(self, generation, member_id, protocol,
                           member_assignment_bytes):
+        # if we were the assignor, then we need to make sure that there have
+        # been no metadata updates since the rebalance begin. Otherwise, we
+        # won't rebalance again until the next metadata change
+        if self._assignment_snapshot and self._assignment_snapshot != self._metadata_snapshot:
+            self._subscription.mark_for_reassignment()
+            return
+
         assignor = self._lookup_assignor(protocol)
-        assert assignor, 'invalid assignment protocol: %s' % protocol
+        assert assignor, 'Coordinator selected invalid assignment protocol: %s' % protocol
 
         assignment = ConsumerProtocol.ASSIGNMENT.decode(member_assignment_bytes)
 
@@ -235,6 +247,11 @@ class ConsumerCoordinator(BaseCoordinator):
         self._subscription.group_subscribe(all_subscribed_topics)
         self._client.set_topics(self._subscription.group_subscription())
 
+        # keep track of the metadata used for assignment so that we can check
+        # after rebalance completion whether anything has changed
+        self._cluster.request_update()
+        self._assignment_snapshot = self._metadata_snapshot
+
         log.debug("Performing assignment for group %s using strategy %s"
                   " with subscriptions %s", self.group_id, assignor.name,
                   member_metadata)
@@ -264,6 +281,7 @@ class ConsumerCoordinator(BaseCoordinator):
                               " for group %s failed on_partitions_revoked",
                               self._subscription.listener, self.group_id)
 
+        self._assignment_snapshot = None
         self._subscription.mark_for_reassignment()
 
     def need_rejoin(self):


Test patch:
diff --git a/test/test_coordinator.py b/test/test_coordinator.py
index 3435292..280fa70 100644
--- a/test/test_coordinator.py
+++ b/test/test_coordinator.py
@@ -85,7 +85,7 @@ def test_pattern_subscription(coordinator, api_version):
     coordinator.config['api_version'] = api_version
     coordinator._subscription.subscribe(pattern='foo')
     assert coordinator._subscription.subscription == set([])
-    assert coordinator._subscription_metadata_changed() is False
+    assert coordinator._subscription_metadata_changed({}) is False
     assert coordinator._subscription.needs_partition_assignment is False
 
     cluster = coordinator._client.cluster