Instantiate error for each future.set_exception
@@ -174,11 +174,11 @@ class AIOKafkaConnection:
self._writer = self._reader = None
self._read_task.cancel()
self._read_task = None
+ for _, _, fut in self._requests:
+ if not fut.done():
error = Errors.ConnectionError(
"Connection at {0}:{1} closed".format(
self._host, self._port))
- for _, _, fut in self._requests:
- if not fut.done():
fut.set_exception(error)
self._requests = []
if self._on_close_cb is not None:
@@ -229,11 +229,11 @@ class AIOKafkaConnection:
# Update idle timer.
self._last_action = self._loop.time()
except (OSError, EOFError, ConnectionError) as exc:
+ for _, _, fut in self._requests:
conn_exc = Errors.ConnectionError(
"Connection at {0}:{1} broken".format(self._host, self._port))
conn_exc.__cause__ = exc
conn_exc.__context__ = exc
- for _, _, fut in self._requests:
fut.set_exception(conn_exc)
self.close(reason=CloseReason.CONNECTION_BROKEN)
except asyncio.CancelledError:
