Rocketmq: About consumers stopping pulling messages

[xiaoyuzxcasd]

Background:

• consumers use DefaultMQPushConsumer with only one instance (i.e., only one consumer)
• multiple producers and consumers were started in the application service in question
• the application service in question involves multiple topics configured in two clusters
• the version used is 4.4.0
• what I use is ConsumeMessageConcurrentlyService.

---

Problem description:

1. In the vicinity of 2020-04-04 05:46:26, the application service experienced network disconnection, connection failure and heartbeat packet failure.
   The failure lasted about two minutes, then returned to normal, the connection was successful, the heartbeat packet was successful
   
2. After the network is restored, the consumers cannot work normally, the consumption stops, and the message accumulation increases
3. After network recovery, the log of doRebalance appears continuously and never stops (it has lasted for more than 24h)
   The topics and queues shown in the figure below are just examples of problems. In fact, all the topics and queues involved in the application service have the same situation and log information
   
   
4. In the consumer log, both "add a new pull request" and "because pull is pause, so try to fix it" were printed
5. Search for exception, timeout and drop in the consumer log. These keywords do not exist
6. The thread stack related to the problem is as follows:
   
   

---

the problem investigation into the bottleneck, help ~~
Thank you

[duhenglucky]

@xiaoyuzxcasd could you provide more information about this issue? and did you used the orderly consume message service? and it seems that all these issues happened in the orderly consumer.
and it seems that the pull message request(broker-01,queueId=1) always pull message expired(org.apache.rocketmq.client.impl.consumer.RebalanceImpl#updateProcessQueueTableInRebalance, org.apache.rocketmq.client.impl.consumer.ProcessQueue#isPullExpired) and lead to removed after rebalance.

[xiaoyuzxcasd]

@xiaoyuzxcasd could you provide more information about this issue? and did you used the orderly consume message service? and it seems that all these issues happened in the orderly consumer.
and it seems that the pull message request(broker-01,queueId=1) always pull message expired(org.apache.rocketmq.client.impl.consumer.RebalanceImpl#updateProcessQueueTableInRebalance, org.apache.rocketmq.client.impl.consumer.ProcessQueue#isPullExpired) and lead to removed after rebalance.

What I use is ConsumeMessageConcurrentlyService.
The log "add a new pull request" means that the PullRequest's self-looping begins.
The log "because pull is pause, so try to fix it" means that the PullRequest's self-looping stops.
This is where I get most confused
Any code that can jump out of the loop will have corresponding log output.But I didn't find These logs, including what I explained earlier :"Search for exception, timeout and drop in the consumer log. These keywords do not exist".

[duhenglucky]

@xiaoyuzxcasd "add a new pull request" means the rebalance allocated result contains this message queue and started a scheduled service to pull the message, but the pull request expired, so this ProcessQueue was set dropped, so maybe we can try to find what exactly caused the expiration. how about providing more log about this issue?

[duhenglucky]

@xiaoyuzxcasd ping

[xiaoyuzxcasd]

@xiaoyuzxcasd ping

First of all, thank you for your attention to this issue.
Due to the restrictions of the company's system, I cannot take out the complete log file from the company's computer and send it to you.
The only way to show you is to take a photo to capture the key part.
But I think the most critical logs have been intercepted before, and I don't know what else to offer.
The following is the log from the network failure to the network recovery, I hope it can give you some discovery:

[raftonsea]

I also meet the same problem. My consumer seems like not stopped , but it refuse consme data . In my scenario , I have a topic ( vacc_top ) with 48 messageQueues among 3 brokers , I found that it stopped consume in mq4 which reside in broker-a . But the consumer has not stopped , I send a message to the topic by mq-console , it works normal . It just delays some messages . When I restart my application or start a new consumer the delay disappear and everything is fine .

In my opinion , it seems like the MQ-PullRequest exit by some reason so the mq cannot be consumed . I debugged the consumer code , I found the code below in DefaultMQPushConsumerImpl

if (processQueue.isDropped()) { log.info("the pull request[{}] is dropped.", pullRequest.toString()); return; }

if (pullResult.getNextBeginOffset() < prevRequestOffset || firstMsgOffset < prevRequestOffset) { log.warn( "[BUG] pull message result maybe data wrong, nextBeginOffset: {} firstMsgOffset: {} prevRequestOffset: {}", pullResult.getNextBeginOffset(), firstMsgOffset, prevRequestOffset); }

It seems that :
1 , the processQueue is stopped ( perhaps because of expired , when doRebalance rocketmq will evaluate the last pull timestamp of the processqueue to decide weather to expir the processqueue ) so the mq stop consume .
2 , the pullRequest get wrong offset from broker ( the next pull offset from the pullResult is less than the previous pull offset) , so the pullRequest exit .

As everything become ok after I restart my consumer , so I prefer the first reason .
I will inspect my log carefully and hope my case will be helpful .
Thanks .

[xiaoyuzxcasd]

@raftonsea
Do you have this log: "the pull request[{}] is dropped."?
If so, your problem is clear.
But i can`t find any keyword about exception, timeout or drop in the consumer log.
I don't know why the PullRequest stopped, this is the most troubling thing for me!

[gaoyf]

@xiaoyuzxcasd 最近翻了一issue，发现了你这个问题。我之前也遇到过类似的，不知道你日志还保存着没有，有的话你可以搜一下OutOfMemoryError，看看异常堆栈有没有关于PullMessageService的。
主要是你贴图里有好几处这样的日志：

下面说一下你引起这个问题的原因：

1 这个图对应的代码如下：

if (pq.isPullExpired()) {
    switch (this.consumeType()) {
        case CONSUME_ACTIVELY:
            break;
        case CONSUME_PASSIVELY:
            pq.setDropped(true);
            if (this.removeUnnecessaryMessageQueue(mq, pq)) {
                it.remove();
                changed = true;
                log.error("[BUG]doRebalance, {}, remove unnecessary mq, {}, because pull is pause, so try to fixed it",
                    consumerGroup, mq);
            }
            break;
        default:
            break;
    }
}

即pq.isPullExpired()为true，而其对应的代码如下：

public boolean isPullExpired() {
    return (System.currentTimeMillis() - this.lastPullTimestamp) > PULL_MAX_IDLE_TIME;
}

即lastPullTimestamp一直没有更新，那么lastPullTimestamp在哪更新的呢？参考代码DefaultMQPushConsumerImpl.pullMessage()：

2 而执行该方法的线程是PullMessageService。
其实说到这里，你应该也明白原因了，即消息拉取的线程因为OOM停止了，所以不会更新lastPullTimestamp字段。

3 为啥会OOM?

3.1 看你的贴图貌似有两个broker，broker上topic有8个队列，那么总共算来就是16个队列，默认rocketmq限流是每个队列最大拉取1000条消息，16*1000=16000，而单条消息最大为4M，那么客户端最大缓存64000M，即62.5G。
当然，一般情况下单条消息通常小于512K，那么按此计算最大缓存8000M，即7.8G。
所以，如果jvm设置堆不大的情况下，是存在可能把内存打爆的。

3.2 当然还有一种可能，就是业务接到消息后执行的逻辑，需要从数据源大量获取数据到内存。我之前遇到的就是这种情况，业务接到消息后，一次性拉取到内存数据量过大，直接oom，导致拉消息线程退出。

另外，@duhenglucky 建议把PullMessageService.run()捕获的异常改为Throwable，当然，改下异常捕获并不能解决根本问题，但是对于问题定位大有裨益。

[xiaoyuzxcasd]

@gaoyf
非常感谢你对此问题的关注与答复
问题的原因的确是如你所说那样的，PullMessageService线程挂了
由于我的应用里启动了两个PullMessageService线程，只是挂掉了其中一个，还保留着一个，导致没有发现这个问题
另外，当前保留的日志里没发现PullMessageService的oom报错
也许是在更早的日志里，已无法追查了
但问题原因基本上是明确了，再次感谢~~