log recover issue

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

@@ -23,6 +23,7 @@ import java.util.List;
 import java.util.Map;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
+
 import org.apache.rocketmq.common.ServiceThread;
 import org.apache.rocketmq.common.UtilAll;
 import org.apache.rocketmq.common.constant.LoggerName;
@@ -158,7 +159,7 @@ public class CommitLog {
     /**
      * When the normal exit, data recovery, all memory data have been flush
      */
-    public void recoverNormally() {
+    public void recoverNormally(long maxPhyOffsetOfConsumeQueue) {
         boolean checkCRCOnRecover = this.defaultMessageStore.getMessageStoreConfig().isCheckCRCOnRecover();
         final List<MappedFile> mappedFiles = this.mappedFileQueue.getMappedFiles();
         if (!mappedFiles.isEmpty()) {
@@ -171,6 +172,7 @@ public class CommitLog {
             ByteBuffer byteBuffer = mappedFile.sliceByteBuffer();
             long processOffset = mappedFile.getFileFromOffset();
             long mappedFileOffset = 0;
+            boolean isDamaged = false;
             while (true) {
                 DispatchRequest dispatchRequest = this.checkMessageAndReturnSize(byteBuffer, checkCRCOnRecover);
                 int size = dispatchRequest.getMsgSize();
@@ -198,6 +200,7 @@ public class CommitLog {
                 // Intermediate file read error
                 else if (!dispatchRequest.isSuccess()) {
                     log.info("recover physics file end, " + mappedFile.getFileName());
+                    isDamaged = true;
                     break;
                 }
             }
@@ -206,6 +209,12 @@ public class CommitLog {
             this.mappedFileQueue.setFlushedWhere(processOffset);
             this.mappedFileQueue.setCommittedWhere(processOffset);
             this.mappedFileQueue.truncateDirtyFiles(processOffset);
+
+            // Clear ConsumeQueue redundant data
+            if (isDamaged && maxPhyOffsetOfConsumeQueue >= processOffset) {
+                log.warn("maxPhyOffsetOfConsumeQueue({}) >= processOffset({}), truncate dirty logic files", maxPhyOffsetOfConsumeQueue, processOffset);
+                this.defaultMessageStore.truncateDirtyLogicFiles(processOffset);
+            }
         }
     }
 
@@ -390,7 +399,7 @@ public class CommitLog {
         this.confirmOffset = phyOffset;
     }
 
-    public void recoverAbnormally() {
+    public void recoverAbnormally(long maxPhyOffsetOfConsumeQueue) {
         // recover by the minimum time stamp
         boolean checkCRCOnRecover = this.defaultMessageStore.getMessageStoreConfig().isCheckCRCOnRecover();
         final List<MappedFile> mappedFiles = this.mappedFileQueue.getMappedFiles();
@@ -418,6 +427,7 @@ public class CommitLog {
                 DispatchRequest dispatchRequest = this.checkMessageAndReturnSize(byteBuffer, checkCRCOnRecover);
                 int size = dispatchRequest.getMsgSize();
 
+                if (dispatchRequest.isSuccess()) {
                     // Normal data
                     if (size > 0) {
                         mappedFileOffset += size;
@@ -430,11 +440,6 @@ public class CommitLog {
                             this.defaultMessageStore.doDispatch(dispatchRequest);
                         }
                     }
-                // Intermediate file read error
-                else if (size == -1) {
-                    log.info("recover physics file end, " + mappedFile.getFileName());
-                    break;
-                }
                     // Come the end of the file, switch to the next file
                     // Since the return 0 representatives met last hole, this can
                     // not be included in truncate offset
@@ -453,6 +458,10 @@ public class CommitLog {
                             log.info("recover next physics file, " + mappedFile.getFileName());
                         }
                     }
+                } else {
+                    log.info("recover physics file end, " + mappedFile.getFileName());
+                    break;
+                }
             }
 
             processOffset += mappedFileOffset;
@@ -461,8 +470,11 @@ public class CommitLog {
             this.mappedFileQueue.truncateDirtyFiles(processOffset);
 
             // Clear ConsumeQueue redundant data
+            if (maxPhyOffsetOfConsumeQueue >= processOffset) {
+                log.warn("maxPhyOffsetOfConsumeQueue({}) >= processOffset({}), truncate dirty logic files", maxPhyOffsetOfConsumeQueue, processOffset);
                 this.defaultMessageStore.truncateDirtyLogicFiles(processOffset);
             }
+        }
         // Commitlog case files are deleted
         else {
             this.mappedFileQueue.setFlushedWhere(0);

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

@@ -35,6 +35,7 @@ import java.util.concurrent.Executors;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicLong;
+
 import org.apache.rocketmq.common.BrokerConfig;
 import org.apache.rocketmq.common.MixAll;
 import org.apache.rocketmq.common.ServiceThread;
@@ -1276,12 +1277,12 @@ public class DefaultMessageStore implements MessageStore {
     }
 
     private void recover(final boolean lastExitOK) {
-        this.recoverConsumeQueue();
+        long maxPhyOffsetOfConsumeQueue = this.recoverConsumeQueue();
 
         if (lastExitOK) {
-            this.commitLog.recoverNormally();
+            this.commitLog.recoverNormally(maxPhyOffsetOfConsumeQueue);
         } else {
-            this.commitLog.recoverAbnormally();
+            this.commitLog.recoverAbnormally(maxPhyOffsetOfConsumeQueue);
         }
 
         this.recoverTopicQueueTable();
@@ -1306,12 +1307,18 @@ public class DefaultMessageStore implements MessageStore {
         }
     }
 
-    private void recoverConsumeQueue() {
+    private long recoverConsumeQueue() {
+        long maxPhysicOffset = -1;
         for (ConcurrentMap<Integer, ConsumeQueue> maps : this.consumeQueueTable.values()) {
             for (ConsumeQueue logic : maps.values()) {
                 logic.recover();
+                if (logic.getMaxPhysicOffset() > maxPhysicOffset) {
+                    maxPhysicOffset = logic.getMaxPhysicOffset();
+                }
             }
         }
+
+        return maxPhysicOffset;
     }
 
     private void recoverTopicQueueTable() {

store/src/test/java/org/apache/rocketmq/store/DefaultMessageStoreTest.java

@@ -18,9 +18,12 @@
 package org.apache.rocketmq.store;
 
 import java.io.File;
+import java.io.RandomAccessFile;
 import java.net.InetAddress;
 import java.net.InetSocketAddress;
 import java.net.SocketAddress;
+import java.nio.MappedByteBuffer;
+import java.nio.channels.FileChannel;
 import java.nio.channels.OverlappingFileLockException;
 import java.util.Map;
 import java.util.concurrent.atomic.AtomicInteger;
@@ -29,6 +32,7 @@ import org.apache.rocketmq.common.BrokerConfig;
 import org.apache.rocketmq.common.UtilAll;
 import org.apache.rocketmq.store.config.FlushDiskType;
 import org.apache.rocketmq.store.config.MessageStoreConfig;
+import org.apache.rocketmq.store.config.StorePathConfigHelper;
 import org.junit.After;
 import org.apache.rocketmq.store.stats.BrokerStatsManager;
 import org.junit.Before;
@@ -171,6 +175,120 @@ public class DefaultMessageStoreTest {
         assertThat(getMessageResult45.getMessageBufferList().size()).isEqualTo(10);
     }
 
+    @Test
+    public void testRecover() throws Exception {
+        String topic = "recoverTopic";
+        MessageBody = StoreMessage.getBytes();
+        for (int i = 0; i < 100; i++) {
+            MessageExtBrokerInner messageExtBrokerInner = buildMessage();
+            messageExtBrokerInner.setTopic(topic);
+            messageExtBrokerInner.setQueueId(0);
+            messageStore.putMessage(messageExtBrokerInner);
+        }
+
+        Thread.sleep(100);//wait for build consumer queue
+        long maxPhyOffset = messageStore.getMaxPhyOffset();
+        long maxCqOffset = messageStore.getMaxOffsetInQueue(topic, 0);
+
+        //1.just reboot
+        messageStore.shutdown();
+        messageStore = buildMessageStore();
+        boolean load = messageStore.load();
+        assertTrue(load);
+        messageStore.start();
+        assertTrue(maxPhyOffset == messageStore.getMaxPhyOffset());
+        assertTrue(maxCqOffset == messageStore.getMaxOffsetInQueue(topic, 0));
+
+        //2.damage commitlog and reboot normal
+        for (int i = 0; i < 100; i++) {
+            MessageExtBrokerInner messageExtBrokerInner = buildMessage();
+            messageExtBrokerInner.setTopic(topic);
+            messageExtBrokerInner.setQueueId(0);
+            messageStore.putMessage(messageExtBrokerInner);
+        }
+        Thread.sleep(100);
+        long secondLastPhyOffset = messageStore.getMaxPhyOffset();
+        long secondLastCqOffset = messageStore.getMaxOffsetInQueue(topic, 0);
+
+        MessageExtBrokerInner messageExtBrokerInner = buildMessage();
+        messageExtBrokerInner.setTopic(topic);
+        messageExtBrokerInner.setQueueId(0);
+        messageStore.putMessage(messageExtBrokerInner);
+
+        messageStore.shutdown();
+
+        //damage last message
+        damageCommitlog(secondLastPhyOffset);
+
+        //reboot
+        messageStore = buildMessageStore();
+        load = messageStore.load();
+        assertTrue(load);
+        messageStore.start();
+        assertTrue(secondLastPhyOffset == messageStore.getMaxPhyOffset());
+        assertTrue(secondLastCqOffset == messageStore.getMaxOffsetInQueue(topic, 0));
+
+        //3.damage commitlog and reboot abnormal
+        for (int i = 0; i < 100; i++) {
+            messageExtBrokerInner = buildMessage();
+            messageExtBrokerInner.setTopic(topic);
+            messageExtBrokerInner.setQueueId(0);
+            messageStore.putMessage(messageExtBrokerInner);
+        }
+        Thread.sleep(100);
+        secondLastPhyOffset = messageStore.getMaxPhyOffset();
+        secondLastCqOffset = messageStore.getMaxOffsetInQueue(topic, 0);
+
+        messageExtBrokerInner = buildMessage();
+        messageExtBrokerInner.setTopic(topic);
+        messageExtBrokerInner.setQueueId(0);
+        messageStore.putMessage(messageExtBrokerInner);
+        messageStore.shutdown();
+
+        //damage last message
+        damageCommitlog(secondLastPhyOffset);
+        //add abort file
+        String fileName = StorePathConfigHelper.getAbortFile(((DefaultMessageStore) messageStore).getMessageStoreConfig().getStorePathRootDir());
+        File file = new File(fileName);
+        MappedFile.ensureDirOK(file.getParent());
+        file.createNewFile();
+
+        messageStore = buildMessageStore();
+        load = messageStore.load();
+        assertTrue(load);
+        messageStore.start();
+        assertTrue(secondLastPhyOffset == messageStore.getMaxPhyOffset());
+        assertTrue(secondLastCqOffset == messageStore.getMaxOffsetInQueue(topic, 0));
+
+        //message write again
+        for (int i = 0; i < 100; i++) {
+            messageExtBrokerInner = buildMessage();
+            messageExtBrokerInner.setTopic(topic);
+            messageExtBrokerInner.setQueueId(0);
+            messageStore.putMessage(messageExtBrokerInner);
+        }
+    }
+
+    private void damageCommitlog(long offset) throws Exception {
+        MessageStoreConfig messageStoreConfig = new MessageStoreConfig();
+        File file = new File(messageStoreConfig.getStorePathCommitLog() + File.separator + "00000000000000000000");
+
+        FileChannel fileChannel = new RandomAccessFile(file, "rw").getChannel();
+        MappedByteBuffer mappedByteBuffer = fileChannel.map(FileChannel.MapMode.READ_WRITE, 0, 1024 * 1024 * 10);
+
+        int bodyLen = mappedByteBuffer.getInt((int) offset + 84);
+        int topicLenIndex = (int) offset + 84 + bodyLen + 4;
+        mappedByteBuffer.position(topicLenIndex);
+        mappedByteBuffer.putInt(0);
+        mappedByteBuffer.putInt(0);
+        mappedByteBuffer.putInt(0);
+        mappedByteBuffer.putInt(0);
+
+        mappedByteBuffer.force();
+        fileChannel.force(true);
+        fileChannel.close();
+    }
+
     private class MyMessageArrivingListener implements MessageArrivingListener {
         @Override
         public void arriving(String topic, int queueId, long logicOffset, long tagsCode, long msgStoreTime,