[GitHub] [rocketmq] leceshide opened a new issue #3171: How update queue offset when consumer occur Exception and return ConsumeConcurrentlyStatus.RECONSUME_LATER status

Thu, 22 Jul 2021 09:21:27 -0700 


leceshide opened a new issue #3171:
URL: https://github.com/apache/rocketmq/issues/3171


   
现象：当消费者消费数据时发生Exception异常且返回`ConsumeConcurrentlyStatus.RECONSUME_LATER`状态时，我们的监控系统就会发现这个队列的“消息积压”指标不断上升而报警。
   效果如下图：
   
![image](https://user-images.githubusercontent.com/2069461/126673455-14817cee-e4e0-496e-a365-1bccab5bb37c.png)
   
   
   
我阅读了`ConsumeMessageConcurrentlyService`这个实现类种对此类case的处理流程（processConsumeResult 
方法），有些疑问还请帮忙解答下哈：
   
   
![image](https://user-images.githubusercontent.com/2069461/126671969-ba1a6609-c4a0-4f6c-a384-054d686107b1.png)
   
   疑问点：
   1）当 `ackIndex = -1` 时，RocketMQ会把出错的 消息 放到重试队列中去，还是把整个queue中的消息放到重试队列？也就是 
`consumeRequest.getMsgs()` 放的消息是哪类消息？
   
![image](https://user-images.githubusercontent.com/2069461/126672773-c6015bf7-4011-4cc5-a7e2-90bf16491a65.png)
   
   2）关于 offset 的同步处理（上图中最后三行代码实现）
   当 `ackIndex = -1` 时，consumeRequest.getMsgs() 是不是已经空了，那么：
   `long offset = 
consumeRequest.getProcessQueue().removeMessage(consumeRequest.getMsgs());`
   返回的结果是？
   
   如果您有时间，还请帮忙给我指导指导，这个问题困扰我好久了，非常感谢：）


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: [email protected]

For queries about this service, please contact Infrastructure at:
[email protected]

Reply via email to