aio-libs / aio-libs/aiokafka

Closing of connection from server don't caught for uvloop

未关闭
#1,038 0 条评论 0 个 reaction 已指派 0 人 在 GitHub 查看
主要语言
Python
星标
1.4k
派生
269
平均合并
1 天 1 小时
30 天内合并 PR
6

描述

**Describe the bug**
We are using aiokafka and uvloop. When connection closed from server and aiokafka do attempt to send something to it, RuntimeError is raised (but only OSError is caught).

Traceback

```
Traceback (most recent call last):

File \"/opt/pysetup/.venv/lib/python3.11/site-packages/aiokafka/consumer/group_coordinator.py\", line 554, in _coordination_routine
await self.__coordination_routine()

File \"/opt/pysetup/.venv/lib/python3.11/site-packages/aiokafka/consumer/group_coordinator.py\", line 612, in __coordination_routine
wait_timeout = await self._maybe_do_autocommit(assignment)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^

File \"/opt/pysetup/.venv/lib/python3.11/site-packages/aiokafka/consumer/group_coordinator.py\", line 907, in _maybe_do_autocommit
await self._do_commit_offsets(

File \"/opt/pysetup/.venv/lib/python3.11/site-packages/aiokafka/consumer/group_coordinator.py\", line 995, in _do_commit_offsets
response = await self._send_req(request)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^

File \"/opt/pysetup/.venv/lib/python3.11/site-packages/aiokafka/consumer/group_coordinator.py\", line 284, in _send_req
resp = await self._client.send(
^^^^^^^^^^^^^^^^^^^^^^^^

File \"/opt/pysetup/.venv/lib/python3.11/site-packages/aiokafka/client.py\", line 502, in send
future = self._conns[(node_id, group)].send(
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^

File \"/opt/pysetup/.venv/lib/python3.11/site-packages/aiokafka/conn.py\", line 439, in send
self._writer.write(size + message)

File \"/usr/local/lib/python3.11/asyncio/streams.py\", line 346, in write
self._transport.write(data)

File \"uvloop/handles/stream.pyx\", line 674, in uvloop.loop.UVStream.write

File \"uvloop/handles/handle.pyx\", line 159, in uvloop.loop.UVHandle._ensure_alive

RuntimeError: unable to perform operation on ; the handler is closed

During handling of the above exception, another exception occurred:

Traceback (most recent call last):

File \"/usr/src/consumers/kafka/consumer.py\", line 68, in run
async for message in self.consumer:

File \"/opt/pysetup/.venv/lib/python3.11/site-packages/opentelemetry/instrumentation/aiokafka/utils.py\", line 207, in _traced_next
record = await func(*args, **kwargs)
^^^^^^^^^^^^^^^^^^^^^^^^^^^

File \"/opt/pysetup/.venv/lib/python3.11/site-packages/aiokafka/consumer/consumer.py\", line 1263, in __anext__
return (await self.getone())
^^^^^^^^^^^^^^^^^^^

File \"/opt/pysetup/.venv/lib/python3.11/site-packages/aiokafka/consumer/consumer.py\", line 1146, in getone
self._coordinator.check_errors()

File \"/opt/pysetup/.venv/lib/python3.11/site-packages/aiokafka/consumer/group_coordinator.py\", line 299, in check_errors
self._coordination_task.result()

aiokafka.errors.KafkaError: KafkaError: Unexpected error during coordination RuntimeError('unable to perform operation on ; the handler is closed
```

**Expected behaviour**
Aiokafka catches this error and recreate connection.

**Environment (please complete the following information):**
- aiokafka version (`python -c "import aiokafka; print(aiokafka.__version__)"`): 0.10.0
- uvloop: 0.19.0
- Kafka Broker version (`kafka-topics.sh --version`): 3.5

**Reproducible example**
```python
class TestClosedSocket:
@pytest.fixture(
params=(
asyncio.DefaultEventLoopPolicy(),
uvloop.EventLoopPolicy(),
),
)
def event_loop(
self, request: pytest.FixtureRequest
) -> Iterable[asyncio.AbstractEventLoop]:
loop: asyncio.AbstractEventLoop = request.param.new_event_loop()
yield loop
loop.close()

@pytest.fixture()
def server(self, unused_tcp_port: int) -> Iterable[Tuple[str, int, socket.socket]]:
host = "localhost"
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.bind((host, unused_tcp_port))
sock.listen(8)
sock.setblocking(False)

yield host, unused_tcp_port, sock

sock.close()

@pytest_asyncio.fixture()
async def conn(
self, server: Tuple[str, int, socket.socket]
) -> AsyncIterable[AIOKafkaConnection]:
host, port, _ = server

conn = AIOKafkaConnection(host=host, port=port, request_timeout_ms=1000)
conn._create_reader_task = mock.Mock()

yield conn

fut = conn.close()
if fut:
await fut

@pytest.mark.asyncio
async def test_send_to_closed_socket(
self, server: Tuple[str, int, socket.socket], conn: AIOKafkaConnection
) -> None:
_, _, sock = server

request = MetadataRequest([])

with pytest.raises(KafkaConnectionError):
await conn.send(request)

await conn.connect()

sock.close()
await asyncio.sleep(0.1)

with pytest.raises(KafkaConnectionError):
await conn.send(request)

```

贡献指南

打开贡献指南

评估

这个 Issue 还没有评估数据。

把新 issue 发到你的邮箱

精选适合新手参与的 GitHub issue 摘要。