Repository: dpkp/kafka-python

Issue:
KafkaConsumer.subscribe() should return copy
When calling ``consumer.subscription()`` a reference to the internal ``set()`` object ``self._subscription.subscription`` is returned.

If that is modified (eg to add a new topic) and then passed to ``consumer.subscribe(topics)`` then ``SubscriptionState.change_subscription`` will hit ``subscription unchanged by change_subscription`` and therefor not call ``self._group_subscription.update``.

It should not be so easy to modify the internal state of the ``SubscriptionState`` therefor consumer.subscription() should return a copy of ``self._subscription.subscription``

Example code, that does not work as expected:

```
consumer = KafkaConsumer()
topics = consumer.subscription()
topics.add('new-topic')
consumer.subscribe(topics)  # Will log that subscription is unchanged
```

Discussion:

Patch:
diff --git a/kafka/consumer/group.py b/kafka/consumer/group.py
index f2b1699..32f4556 100644
--- a/kafka/consumer/group.py
+++ b/kafka/consumer/group.py
@@ -819,7 +819,7 @@ class KafkaConsumer(six.Iterator):
         Returns:
             set: {topic, ...}
         """
-        return self._subscription.subscription
+        return self._subscription.subscription.copy()
 
     def unsubscribe(self):
         """Unsubscribe from all topics and clear all assigned partitions."""


Test patch:
diff --git a/test/test_consumer.py b/test/test_consumer.py
index 073a3af..e5dd946 100644
--- a/test/test_consumer.py
+++ b/test/test_consumer.py
@@ -24,6 +24,14 @@ class TestKafkaConsumer(unittest.TestCase):
         with self.assertRaises(KafkaConfigurationError):
             KafkaConsumer(bootstrap_servers='localhost:9092', fetch_max_wait_ms=41000, request_timeout_ms=40000)
 
+    def test_subscription_copy(self):
+        consumer = KafkaConsumer('foo', api_version=(0, 10))
+        sub = consumer.subscription()
+        assert sub is not consumer.subscription()
+        assert sub == set(['foo'])
+        sub.add('fizz')
+        assert consumer.subscription() == set(['foo'])
+
 
 class TestMultiProcessConsumer(unittest.TestCase):
     @unittest.skipIf(sys.platform.startswith('win'), 'test mocking fails on windows')