Raise AssertionError if consumer closed in poll() (#1978)

Author: dpkp

kafka/consumer/group.py

@@ -638,6 +638,7 @@ def poll(self, timeout_ms=0, max_records=None, update_offsets=True):
             max_records = self.config['max_poll_records']
         assert isinstance(max_records, int), 'max_records must be an integer'
         assert max_records > 0, 'max_records must be positive'
+        assert not self._closed, 'KafkaConsumer is closed'
 
         # Poll for new data until the timeout expires
         start = time.time()
@@ -1173,6 +1174,8 @@ def __iter__(self):  # pylint: disable=non-iterator-returned
         return self
 
     def __next__(self):
+        if self._closed:
+            raise StopIteration('KafkaConsumer closed')
         # Now that the heartbeat thread runs in the background
         # there should be no reason to maintain a separate iterator
         # but we'll keep it available for a few releases just in case