How to handle message accumulation - ApsaraMQ for RocketMQ

Symptoms

An alert for message accumulation on a Message Queue for RocketMQ instance indicates a potential issue. After you log on to the Message Queue for RocketMQ console, you may see the following symptoms:

On the Group Details page, the value of Real-time Accumulated Messages for the Group ID is higher than expected.
In the navigation pane, select Message Tracing, click Create Query Task, and then select Query by Message ID. After you enter the required information, you find that some messages are sent to the broker but not delivered to downstream consumers.

Possible causes

In Message Queue for RocketMQ, after messages are sent to a broker, a client configured with a Group ID pulls messages from the broker based on the current consumer offset and consumes the messages locally. Message accumulation typically does not occur when the client pulls messages from the broker. Message accumulation is usually caused by insufficient client processing capacity, which can result from factors like long consumption time or low consumer concurrency. For more information about the consumption mechanism and the causes of message accumulation, see Message accumulation and latency issues.

Solution

If message accumulation occurs, follow these steps to resolve the issue.

Determine whether the message accumulation occurs on the Message Queue for RocketMQ server or on the client.
Check the local client log file ons.log and search for the following message:
```
the cached message count exceeds the threshold
```
- If this log message appears, the client's local buffer queue is full and messages are accumulating on the client. In this case, proceed to Step 2.
- If this log message is not found, the message accumulation does not occur on the client. In this exceptional case, contact Alibaba Cloud technical support.
Check whether the message consumption time is reasonable.
- If the consumption time is too long, check the client stack trace to troubleshoot your business logic. In this case, proceed to Step 3.
- If the consumption time is normal, the message accumulation might be caused by insufficient consumer concurrency. To resolve this issue, gradually increase the number of consumer threads or scale out your consumer nodes.
You can check the message consumption time in the following ways:
- Log on to the Message Queue for RocketMQ console to view the message trace. In the Consumer section, you can view the consumption time of a single message. For more information, see Query message traces. To obtain the consumption time for a single message, query the message trace and view the Message Processing Time in the consumer delivery results on the trace details page.
- Log on to the Message Queue for RocketMQ console to view the consumer status. In the client connection information, check the Response Time to obtain the average consumption time. For more information, see View consumer status. In the Consumption Statistics section of the Java Client Real-time Data pop-up window, you can view metrics for each topic, such as Response Time (ms/message), Successful Messages (messages/s), Failed Messages (messages/s), and Message Accumulation. If the Response Time value is too high (for example, 5003.94 ms/message) and the message accumulation continuously increases, a consumption latency issue exists.
- Use other monitoring products, such as Application Real-Time Monitoring Service (ARMS), to instrument your application and collect data about message consumption time.

View the client stack trace. Focus only on threads that are named ConsumeMessageThread. These threads contain the business logic for message consumption. You can refer to the official Java documentation to determine the thread state and modify your business logic accordingly.

You can obtain the client stack trace in the following ways:

Log on to the Message Queue for RocketMQ console and view the consumer status. In the client connection information, find the View Stack Information option. For more information, see View consumer status.
Use the jstack tool to print the stack trace.
1. See View consumer status to find the IP address of the host running the consumer instance with message accumulation. Then, log on to the host.
2. Run one of the following commands to view and record the process ID (PID) of the Java process.
```
ps -ef 
|grep javajps -lm
```
3. Run the following command to view the stack trace.
```
jstack -l pid > /tmp/pid.jstack
```
4. Run the following command to view information about ConsumeMessageThread threads.
```
cat /tmp/pid.jstack|grep ConsumeMessageThread -A 10 --color
```

The following examples show common abnormal stack traces:

Example 1: An idle stack trace with no accumulation.

When a consumer is idle, its consumer threads are in the WAITING state and waiting to retrieve messages from the consumption task queue.

For example, the stack trace for the ConsumeMessageThread_7 thread shows TID: 53 STATE: WAITING, and the call chain is ThreadPoolExecutor.runWorker → ThreadPoolExecutor.getTask → LinkedBlockingQueue.take → LockSupport.park → sun.misc.Unsafe.park.

Example 2: The consumption logic involves lock contention or sleep states.

The consumer thread is blocked by an internal sleep or wait operation, which slows consumption.

Thread: ConsumeMessageThread_16
Stack:
TID: 51 STATE: TIMED_WAITING
java.lang.Thread.sleep(Native Method)
mqtest.DelayTest$1.consume(DelayTest.java:51)
com.aliyun.openservices.ons.api.impl.rocketmq.ConsumerImpl$MessageListenerImpl.consumeMessage(ConsumerImpl.java:101)
com.aliyun.openservices.shade.com.alibaba.rocketmq.client.impl.consumer.ConsumeMessageConcurrentlyService$ConsumeRequest.run(ConsumeMessageConcurrentlyService.java:415)
java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
java.util.concurrent.FutureTask.run(FutureTask.java:266)
java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
java.lang.Thread.run(Thread.java:748)

Example 3: The consumption logic is stuck on an operation with an external storage system, such as a database.

The consumer thread is blocked by an external HTTP call, which slows consumption.

ConsumeMessageThread_3
TID: 54 STATE: RUNNABLE
java.lang.ClassLoader.loadClass(ClassLoader.java:404)
sun.misc.Launcher$AppClassLoader.loadClass(Launcher.java:349)
java.lang.ClassLoader.loadClass(ClassLoader.java:357)
org.apache.http.impl.conn.PoolingHttpClientConnectionManager.<init>(PoolingHttpClientConnectionManager.java:174)
org.apache.http.impl.conn.PoolingHttpClientConnectionManager.<init>(PoolingHttpClientConnectionManager.java:158)
org.apache.http.impl.conn.PoolingHttpClientConnectionManager.<init>(PoolingHttpClientConnectionManager.java:149)
org.apache.http.impl.conn.PoolingHttpClientConnectionManager.<init>(PoolingHttpClientConnectionManager.java:125)
refactor.base.Tools.getHttpsClient(Tools.java:138)
refactor.base.Tools.httpsPost(Tools.java:257)
mqtest.DelayTest$1.consume(DelayTest.java:58)
com.aliyun.openservices.ons.api.impl.rocketmq.ConsumerImpl$MessageListenerImpl.consumeMessage(ConsumerImpl.java:101)
com.aliyun.openservices.shade.com.alibaba.rocketmq.client.impl.consumer.ConsumeMessageConcurrentlyService$ConsumeRequest.run(ConsumeMessageConcurrentlyService.java:415)
java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
java.util.concurrent.FutureTask.run(FutureTask.java:266)
java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
java.lang.Thread.run(Thread.java:748)

If message accumulation affects your services and the messages can be safely skipped, use the Reset Consumer Offsets feature. This feature lets you skip accumulated messages and resume consumption from the latest offset to quickly restore your services. For more information, see Reset Consumer Offsets.