aio-libs / aio-libs/aiokafka

[QUESTION] aiokafka stucks after group rejoin

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

描述

### Description
We have long running service, which processes messages through the aiokafka, running in coroutine. Once in a few days `Failed fetch messages from 1: [Error 7] RequestTimedOutError` happens for some reason and after that no messages are consumed. It seems to rejoin the group properly after `RequestTimedOutError` but nothing happens.
### Environment
The service is deployed into `python:3.10-slim` Docker container
* aiokafka version: 0.7.2
* kafka-python version: 2.0.2

### Coroutine code used
```python3
async def listen_kafka():
consumer = AIOKafkaConsumer(
tasks_topic,
group_id=f"{tasks_topic}_group",
bootstrap_servers=kafka_addr,
session_timeout_ms=45000,
enable_auto_commit=False,
value_deserializer=lambda x: loads(x.decode('utf-8')))
try:
while True:
try:
await consumer.start()
except KafkaConnectionError as e:
await asyncio.sleep(5)
log.error(e)
else:
log.info("Connection to Kafka established")
break
try:
async for msg in consumer:
if type(msg) is Exception:
log.error(msg)
continue
await dispatch(msg.value)
await consumer.commit()

except Exception as e:
log.error(e)
finally:
await consumer.stop()

except asyncio.CancelledError:
log.info("Stopping consumer...")
await consumer.stop()
log.info("Listen Kafka task stopped")
```
Unfortunately, no decent text logs left yet we have a screenshot that shows the situation fairly enough.
![2022-07-19 10 47 02](https://user-images.githubusercontent.com/42657841/179700565-5c4b738c-4d9e-4c5f-9825-333596ecf602.png)
At the time the screenshot was taken plenty of tasks had been in kafka. They were successfully processed after service restart.

Do you have any ideas how we can overcome or at least debug the problem?

贡献指南

打开贡献指南

评估

这个 Issue 还没有评估数据。

把新 issue 发到你的邮箱

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