From 9bc01657ed9402b502f7156ae95764029436eab3 Mon Sep 17 00:00:00 2001 From: Dana Powers Date: Wed, 30 Dec 2015 11:49:22 -0800 Subject: Resolve delayed task futures in KafkaClient.poll --- kafka/client_async.py | 9 ++++++--- 1 file changed, 6 insertions(+), 3 deletions(-) (limited to 'kafka/client_async.py') diff --git a/kafka/client_async.py b/kafka/client_async.py index 386b0cb..ca81214 100644 --- a/kafka/client_async.py +++ b/kafka/client_async.py @@ -243,11 +243,14 @@ class KafkaClient(object): metadata_timeout = self._maybe_refresh_metadata() # Send scheduled tasks - for task in self._delayed_tasks.pop_ready(): + for task, future in self._delayed_tasks.pop_ready(): try: - task() + result = task() except Exception as e: log.error("Task %s failed: %s", task, e) + future.failure(e) + else: + future.success(result) timeout = min(timeout_ms, metadata_timeout, self.config['request_timeout_ms']) @@ -450,7 +453,7 @@ class DelayedTaskQueue(object): else: task, future = maybe_task del self._task_map[task] - return task + return (task, future) def next_at(self): """Number of seconds until next task is ready""" -- cgit v1.2.1