diff options
| author | Dana Powers <dana.powers@gmail.com> | 2018-02-02 16:36:30 -0800 |
|---|---|---|
| committer | GitHub <noreply@github.com> | 2018-02-02 16:36:30 -0800 |
| commit | 618c5051493693c1305aa9f08e8a0583d5fcf0e3 (patch) | |
| tree | 3a2fcec8260915a83f19a603671c4a0e5461cca0 /kafka/consumer/fetcher.py | |
| parent | 08a7fb7b754a754c6c64e96d4ba5c4f56cf38a5f (diff) | |
| download | kafka-python-618c5051493693c1305aa9f08e8a0583d5fcf0e3.tar.gz | |
KAFKA-3949: Avoid race condition when subscription changes during rebalance (#1364)
Diffstat (limited to 'kafka/consumer/fetcher.py')
| -rw-r--r-- | kafka/consumer/fetcher.py | 6 |
1 files changed, 0 insertions, 6 deletions
diff --git a/kafka/consumer/fetcher.py b/kafka/consumer/fetcher.py index afb8f52..f9fcb37 100644 --- a/kafka/consumer/fetcher.py +++ b/kafka/consumer/fetcher.py @@ -326,9 +326,6 @@ class Fetcher(six.Iterator): max_records = self.config['max_poll_records'] assert max_records > 0 - if self._subscriptions.needs_partition_assignment: - return {}, False - drained = collections.defaultdict(list) records_remaining = max_records @@ -397,9 +394,6 @@ class Fetcher(six.Iterator): def _message_generator(self): """Iterate over fetched_records""" - if self._subscriptions.needs_partition_assignment: - raise StopIteration('Subscription needs partition assignment') - while self._next_partition_records or self._completed_fetches: if not self._next_partition_records: |