[ISSUE #6728] Compute the confirmOffset without considering new connections

store/src/main/java/org/apache/rocketmq/store/CommitLog.java

@@ -553,15 +553,35 @@ private void setBatchSizeIfNeeded(Map<String, String> propertiesMap, DispatchReq
         }
     }
 
+    // Fetch and compute the newest confirmOffset.
+    // Even if it is just inited.
     public long getConfirmOffset() {
         if (this.defaultMessageStore.getBrokerConfig().isEnableControllerMode()) {
             if (this.defaultMessageStore.getMessageStoreConfig().getBrokerRole() != BrokerRole.SLAVE && !this.defaultMessageStore.getRunningFlags().isFenced()) {
                 if (((AutoSwitchHAService) this.defaultMessageStore.getHaService()).getLocalSyncStateSet().size() == 1) {
                     return this.defaultMessageStore.getMaxPhyOffset();
                 }
-                // First time compute confirmOffset.
+                // First time it will compute the confirmOffset.
                 if (this.confirmOffset <= 0) {
                     setConfirmOffset(((AutoSwitchHAService) this.defaultMessageStore.getHaService()).computeConfirmOffset());
+                    log.info("Init the confirmOffset to {}.", this.confirmOffset);
+                }
+            }
+            return this.confirmOffset;
+        } else if (this.defaultMessageStore.getMessageStoreConfig().isDuplicationEnable()) {
+            return this.confirmOffset;
+        } else {
+            return getMaxOffset();
+        }
+    }
+
+    // Fetch the original confirmOffset's value.
+    // Without checking and re-computing.
+    public long getConfirmOffsetDirectly() {
+        if (this.defaultMessageStore.getBrokerConfig().isEnableControllerMode()) {
+            if (this.defaultMessageStore.getMessageStoreConfig().getBrokerRole() != BrokerRole.SLAVE && !this.defaultMessageStore.getRunningFlags().isFenced()) {
+                if (((AutoSwitchHAService) this.defaultMessageStore.getHaService()).getLocalSyncStateSet().size() == 1) {
+                    return this.defaultMessageStore.getMaxPhyOffset();
                 }
             }
             return this.confirmOffset;

store/src/main/java/org/apache/rocketmq/store/DefaultMessageStore.java

@@ -1589,11 +1589,19 @@ public boolean resetWriteOffset(long phyOffset) {
         }
     }
 
+    // Fetch and compute the newest confirmOffset.
+    // Even if it is just inited.
     @Override
     public long getConfirmOffset() {
         return this.commitLog.getConfirmOffset();
     }
 
+    // Fetch the original confirmOffset's value.
+    // Without checking and re-computing.
+    public long getConfirmOffsetDirectly() {
+        return this.commitLog.getConfirmOffsetDirectly();
+    }
+
     @Override
     public void setConfirmOffset(long phyOffset) {
         this.commitLog.setConfirmOffset(phyOffset);

store/src/main/java/org/apache/rocketmq/store/ha/autoswitch/AutoSwitchHAService.java

@@ -421,14 +421,21 @@ public long computeConfirmOffset() {
         for (Long syncId : currentSyncStateSet) {
             if (!idList.contains(syncId) && this.brokerControllerId != null && !Objects.equals(syncId, this.brokerControllerId)) {
                 LOGGER.warn("Slave {} is still in syncStateSet, but has lost its connection. So new offset can't be compute.", syncId);
-                return this.defaultMessageStore.getConfirmOffset();
+                // Without check and re-compute, return the confirmOffset's value directly.
+                return this.defaultMessageStore.getConfirmOffsetDirectly();
             }
         }
 
         for (HAConnection connection : this.connectionList) {
             final Long slaveId = ((AutoSwitchHAConnection) connection).getSlaveId();
             if (currentSyncStateSet.contains(slaveId)) {
-                newConfirmOffset = Math.min(newConfirmOffset, connection.getSlaveAckOffset());
+                long slaveAckOffset = connection.getSlaveAckOffset();
+                if (slaveAckOffset <= 0) {
+                    // Slave's connection is just inited, the ack hasn't been calculated.
+                    // So skip this ackOffset.
+                    continue;
+                }
+                newConfirmOffset = Math.min(newConfirmOffset, slaveAckOffset);
             }
         }
         return newConfirmOffset;