apache / apache/rocketmq-clients
[Bug] [Python] PushConsumer __execute_receive_later infinite recursion causes RecursionError stack overflow
- Dominant language
- Java
- Stars
- 505
- Forks
- 313
- Avg merge
- 11h 28m
- Merged PRs (30d)
- 6
Description
### Before Creating the Bug Report
- [x] I found a bug, not just asking a question, which should be created in [GitHub Discussions](https://github.com/apache/rocketmq-clients/discussions).
- [x] I have searched the [GitHub Issues](https://github.com/apache/rocketmq-clients/issues) and [GitHub Discussions](https://github.com/apache/rocketmq-clients/discussions) of this repository and believe that this is not a duplicate.
- [x] I have confirmed that this bug belongs to the current repository, not other repositories of RocketMQ.
### Programming Language of the Client
Python
### Runtime Platform Environment
### Environment
- **SDK**: `rocketmq-python-client==5.1.1`
- **Python**: 3.8
- **RocketMQ Server**: 5.x
- **OS**: Linux (CentOS 7 / Ubuntu 20.04)
### Describe the bug
When `PushConsumer`'s process queue cache is full and messages keep arriving faster than they can be consumed, `__execute_receive` and `__execute_receive_later` enter a synchronous mutual recursion. Each call adds a stack
frame, and after ~800-1000 seconds (~13-16 minutes), Python's default recursion limit (1000) is exceeded and the process crashes with `RecursionError: maximum recursion depth exceeded`.
### Reproducer
Run a PushConsumer with slow message processing against a topic with sustained high throughput. For example, a video-processing consumer that takes 30-90 seconds per message, with the default `consumption_thread_count=20` and `receive_batch_size=32`.
### Stack trace
Thread 0x00007f0abe7fc700 (most recent call first):
File "logging/init.py", line 424, in usesTime
File "logging/init.py", line 633, in usesTime
File "logging/init.py", line 665, in format
File "logging/init.py", line 925, in format
File "logging/init.py", line 1081, in emit
File "logging/init.py", line 1183, in emit
File "logging/handlers.py", line 71, in emit
File "logging/init.py", line 950, in handle
File "logging/init.py", line 1649, in callHandlers
File "logging/init.py", line 1587, in handle
File "logging/init.py", line 1577, in _log
File "logging/init.py", line 1446, in warning
File "logging/init.py", line 1451, in warn
File "rocketmq/v5/model/process_queue.py", line 66, in is_cache_full
File "rocketmq/v5/consumer/push/push_consumer.py", line 235, in __execute_receive
File "rocketmq/v5/consumer/push/push_consumer.py", line 271, in __execute_receive_later
File "rocketmq/v5/consumer/push/push_consumer.py", line 237, in __execute_receive
File "rocketmq/v5/consumer/push/push_consumer.py", line 271, in __execute_receive_later
File "rocketmq/v5/consumer/push/push_consumer.py", line 237, in __execute_receive
File "rocketmq/v5/consumer/push/push_consumer.py", line 271, in __execute_receive_later
... (hundreds more identical frames) ...
### Root Cause
The call chain forms unbounded synchronous recursion:
__execute_receive (line 237)
→ is_cache_full() returns True
→ __execute_receive_later (line 271)
→ sleep(1s)
→ __execute_receive (line 237) ← back to start, +1 stack frame
1 stack frame is added per second. After ~800-1000 seconds → `RecursionError` crash.
Additionally, `process_queue.py:66` `is_cache_full()` calls `logging.warn()` on each check, whose `emit → format → usesTime` path consumes extra stack frames per iteration, accelerating the overflow.
### Suggested fix
Replace the recursive call in `__execute_receive_later` with an iterative `while` loop:
```python
# Before
def __execute_receive_later(self, message_queue, process_queue, attempt_id):
time.sleep(PushConsumer.RECEIVE_RETRY_DELAY)
self.__execute_receive(message_queue, process_queue, attempt_id)
# After
def __execute_receive_later(self, message_queue, process_queue, attempt_id):
while self.is_running and not process_queue.dropped:
time.sleep(PushConsumer.RECEIVE_RETRY_DELAY)
if not self.is_running or process_queue.dropped:
return
if not process_queue.is_cache_full(
self.__queue_threshold(self.__max_cache_message_count),
self.__queue_threshold(self.__max_cache_message_size)):
self.__execute_receive(message_queue, process_queue, attempt_id)
return
This preserves identical behavior (sleep + retry) but uses O(1) stack space.
I'm happy to submit a PR with this fix if the approach looks good.
---
### RocketMQ Version of the Client/Server
rocketmq-python-client==5.1.1
### Run or Compiler Version
_No response_
### Describe the Bug
PushConsumer 在消息处理速度跟不上拉取速度时,`__execute_receive` 和 `__execute_receive_later` 两个方法会形成同步递归。每秒钟增加一层栈帧,约 13-16 分钟后 Python 栈溢出(RecursionError),进程崩溃。
具体链路:
1. `__execute_receive` 检查 `is_cache_full()` → 缓存满了
2. 调用 `__execute_receive_later` → sleep 1秒 → 回调 `__execute_receive`
3. 缓存仍满 → 再次调用 `__execute_receive_later` → 无限循环
4. 每次循环 `is_cache_full` 还调用 `logging.warn()`,额外消耗栈帧
5. 约 800-1000 次递归后超出 Python 默认递归限制 1000,崩溃
线程堆栈中可见数百个重复的:
`push_consumer.py:237 __execute_receive → push_consumer.py:271 __execute_receive_later → push_consumer.py:237 __execute_receive → ...`
### Steps to Reproduce
1. 启动 RocketMQ 5.x 服务端
2. 创建一个 PushConsumer,使用默认配置(consumption_thread_count=20, max_cache_message_count=1024)
3. 在 MessageListener.consume() 中加入较长的处理耗时(如 `time.sleep(5)` 模拟慢消费)
4. 向对应 topic 持续高频发送消息(如每秒 100 条)
5. 运行约 15-20 分钟后,进程崩溃并抛出 `RecursionError: maximum recursion depth exceeded`
环境信息:
- SDK: rocketmq-python-client==5.1.1
- Python: 3.8
- OS: Linux
### What Did You Expect to See?
PushConsumer 在缓存满时应该使用迭代式循环等待(while + sleep),而不是递归调用。进程应该稳定运行,无论缓存满多久都不会因栈溢出而崩溃
### What Did You See Instead?
进程运行约 15-20 分钟后崩溃,抛出 RecursionError: maximum recursion depth exceeded。
线程堆栈中可见数百个重复帧:
push_consumer.py:237 __execute_receive
push_consumer.py:271 __execute_receive_later
push_consumer.py:237 __execute_receive
...(重复数百次)
process_queue.py:66 is_cache_full → logging.warn
logging/__init__.py emit → format → usesTime(日志调用耗尽最后栈空间)
### Additional Context
_No response_
Contributor guide
No contributing guide indexed for this repository
Research direction
Start in rocketmq/v5/consumer/push/push_consumer.py at __execute_receive and __execute_receive_later, then inspect rocketmq/v5/model/process_queue.py:is_cache_full. Reproduce with a slow PushConsumer and sustained high throughput; done means retries no longer grow the call stack or raise RecursionError while cache-full behavior remains intact.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- python
- Domain
- distributed-systems
- Issue type
- Bug
- Difficulty
- 2/5
- Estimated time
- 1-3 hours
- Activity status
- Active
- Clarity
- Clearly specified
- Newbie friendliness
- 78/100