[rocketmq] branch develop updated: #3015 when slaveAckOffset greater than local commitLog offset, break connection and return (#3016)

[duhengforever]

This is an automated email from the ASF dual-hosted git repository.

duhengforever pushed a commit to branch develop
in repository https://gitbox.apache.org/repos/asf/rocketmq.git


The following commit(s) were added to refs/heads/develop by this push:
     new 84cde5f  #3015 when slaveAckOffset greater than local commitLog 
offset, break connection and return (#3016)
84cde5f is described below

commit 84cde5f5beddfc16621f6b5dba31883796d2c216
Author: makabakaboom <402879...@qq.com>
AuthorDate: Thu Jan 20 13:37:01 2022 +0800

    #3015 when slaveAckOffset greater than local commitLog offset, break 
connection and return (#3016)
    
    when slaveAckOffset greater than local commitLog offset, break connection 
and return
---
 store/src/main/java/org/apache/rocketmq/store/ha/HAConnection.java | 6 ++++++
 1 file changed, 6 insertions(+)

diff --git a/store/src/main/java/org/apache/rocketmq/store/ha/HAConnection.java 
b/store/src/main/java/org/apache/rocketmq/store/ha/HAConnection.java
index 11af1e2..dd68c73 100644
--- a/store/src/main/java/org/apache/rocketmq/store/ha/HAConnection.java
+++ b/store/src/main/java/org/apache/rocketmq/store/ha/HAConnection.java
@@ -168,6 +168,12 @@ public class HAConnection {
                             if (HAConnection.this.slaveRequestOffset < 0) {
                                 HAConnection.this.slaveRequestOffset = 
readOffset;
                                 log.info("slave[" + 
HAConnection.this.clientAddr + "] request offset " + readOffset);
+                            } else if (HAConnection.this.slaveAckOffset > 
HAConnection.this.haService.getDefaultMessageStore().getMaxPhyOffset()){
+                                log.warn("slave[{}] request offset={} greater 
than local commitLog offset={}. ",
+                                        HAConnection.this.clientAddr,
+                                        HAConnection.this.slaveAckOffset,
+                                        
HAConnection.this.haService.getDefaultMessageStore().getMaxPhyOffset());
+                                return false;
                             }
 
                             
HAConnection.this.haService.notifyTransferSome(HAConnection.this.slaveAckOffset);