diff options
| author | Dana Powers <dana.powers@rd.io> | 2016-01-04 00:18:58 -0800 |
|---|---|---|
| committer | Dana Powers <dana.powers@rd.io> | 2016-01-04 00:18:58 -0800 |
| commit | 16c834119b6113450a59a05ce627669ba5752f37 (patch) | |
| tree | f0d559f515bddfe50e349432c71109f0da4b6e1b /kafka | |
| parent | bb433e27f020597f3807cb4058d1e45a671cea6e (diff) | |
| download | kafka-python-16c834119b6113450a59a05ce627669ba5752f37.tar.gz | |
Fix future redefine bug in client.poll
Diffstat (limited to 'kafka')
| -rw-r--r-- | kafka/client_async.py | 6 | ||||
| -rw-r--r-- | kafka/coordinator/base.py | 1 |
2 files changed, 4 insertions, 3 deletions
diff --git a/kafka/client_async.py b/kafka/client_async.py index 914afec..747a85f 100644 --- a/kafka/client_async.py +++ b/kafka/client_async.py @@ -292,14 +292,14 @@ class KafkaClient(object): metadata_timeout = self._maybe_refresh_metadata() # Send scheduled tasks - for task, future in self._delayed_tasks.pop_ready(): + for task, task_future in self._delayed_tasks.pop_ready(): try: result = task() except Exception as e: log.error("Task %s failed: %s", task, e) - future.failure(e) + task_future.failure(e) else: - future.success(result) + task_future.success(result) timeout = min(timeout_ms, metadata_timeout, self.config['request_timeout_ms']) diff --git a/kafka/coordinator/base.py b/kafka/coordinator/base.py index 4f16bb0..bcd5889 100644 --- a/kafka/coordinator/base.py +++ b/kafka/coordinator/base.py @@ -236,6 +236,7 @@ class BaseCoordinator(object): self.needs_join_prepare = True self.heartbeat_task.reset() else: + assert future.failed() exception = future.exception if isinstance(exception, (Errors.UnknownMemberIdError, Errors.RebalanceInProgressError, |