apache / apache/rocketmq-clients

[Bug] [Python] PushConsumer __execute_receive_later infinite recursion causes RecursionError stack overflow

Open Beginner friendly
#1,331 11 comments 2 reactions 0 assignees View on GitHub
type/bug
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

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.