[I] 5.3.2版本，延迟消息，投递失败，投递设置的开始offset有误 [rocketmq]

[via GitHub]

coolmoon101 opened a new issue, #9249:
URL: https://github.com/apache/rocketmq/issues/9249

   ### 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/discussions).
   
   - [x] I have searched the [GitHub 
Issues](https://github.com/apache/rocketmq/issues) and [GitHub 
Discussions](https://github.com/apache/rocketmq/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.
   
   
   ### Runtime platform environment
   
   mac
   
   ### RocketMQ version
   
   5.3.2
   
   ### JDK Version
   
   1.8
   
   ### Describe the Bug
   
   5.3.2 版本：
   类：ScheduleMessageService.DeliverDelayedMessageTimerTask#executeOnTimeUp
   
   ReferredIterator<CqUnit> bufferCQ = cq.iterateFrom(this.offset);
   long nextOffset = this.offset;
   try {
   while (bufferCQ.hasNext() && isStarted()) {
       .......
       long currOffset = cqUnit.getQueueOffset();
       assert cqUnit.getBatchNum() == 1;
       nextOffset = currOffset + cqUnit.getBatchNum();
       ......
       boolean deliverSuc;
       if (ScheduleMessageService.this.enableAsyncDeliver) {
           deliverSuc = this.asyncDeliver(msgInner, msgExt.getMsgId(), 
currOffset, offsetPy, sizePy);
       } else {
           deliverSuc = this.syncDeliver(msgInner, msgExt.getMsgId(), 
currOffset, offsetPy, sizePy);
       }
   
       if (!deliverSuc) {
           this.scheduleNextTimerTask(nextOffset, DELAY_FOR_A_WHILE);
           return;
       }
   
   } 
   
   5.3.2 版本是否有问题。
   投递失败时：this.scheduleNextTimerTask(nextOffset, DELAY_FOR_A_WHILE);
   这里第一个参数是应该为 currOffset 吧
   
   ### Steps to Reproduce
   
   无
   
   ### What Did You Expect to See?
   
   无
   
   ### What Did You See Instead?
   
   if (!deliverSuc) {
         this.scheduleNextTimerTask(currOffset, DELAY_FOR_A_WHILE);
         return;
     }
   
   ### Additional Context
   
   我对比了4.9.8的版本
   类：ScheduleMessageService.DeliverDelayedMessageTimerTask#executeOnTimeUp
   
   SelectMappedBufferResult bufferCQ = cq.getIndexBuffer(this.offset);
   int i = 0;
   ConsumeQueueExt.CqExtUnit cqExtUnit = new ConsumeQueueExt.CqExtUnit();
   for (; i < bufferCQ.getSize() && isStarted(); i += 
ConsumeQueue.CQ_STORE_UNIT_SIZE) {
       ......
       nextOffset = offset + (i / ConsumeQueue.CQ_STORE_UNIT_SIZE);    
       ......
       boolean deliverSuc;
       if (ScheduleMessageService.this.enableAsyncDeliver) {
           deliverSuc = this.asyncDeliver(msgInner, msgExt.getMsgId(), 
nextOffset, offsetPy, sizePy);
       } else {
           deliverSuc = this.syncDeliver(msgInner, msgExt.getMsgId(), 
nextOffset, offsetPy, sizePy);
       }
   
       if (!deliverSuc) {
           this.scheduleNextTimerTask(nextOffset, DELAY_FOR_A_WHILE);
           return;
       }
       ......
   }
   在这里投递和投递失败时的处理都是使用的是 nextOffset


-- 
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: commits-unsubscr...@rocketmq.apache.org.apache.org

For queries about this service, please contact Infrastructure at:
us...@infra.apache.org