aio-libs / aio-libs/aiokafka

Closing of connection from server don't caught for uvloop

Abierto
#1,038 0 comentarios 0 reacciones 0 asignados Ver en GitHub
Lenguaje dominante
Python
Estrellas
1.4k
Forks
269
Merge medio
1 d 1 h
PR fusionados (30 d)
6

Descripción

**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)

```

Guía de contribución

Abrir la guía de contribución

Evaluación

Este issue todavía no se ha evaluado.

Recibe los nuevos issues en tu correo

Un resumen breve de issues de GitHub para principiantes.