From 736e813a7a1cd1f0f509832023ec20d04d202694 Mon Sep 17 00:00:00 2001 From: Rui <1685901819@qq.com> Date: Sun, 2 Aug 2026 21:24:40 +0800 Subject: [PATCH] [ISSUE #10755] Fix ConsumeQueueExt truncation cleanup Signed-off-by: Rui <1685901819@qq.com> --- .../apache/rocketmq/store/ConsumeQueue.java | 117 +++-- .../rocketmq/store/ConsumeQueueExt.java | 34 ++ .../rocketmq/store/ConsumeQueueTest.java | 494 ++++++++++++++++++ 3 files changed, 607 insertions(+), 38 deletions(-) diff --git a/store/src/main/java/org/apache/rocketmq/store/ConsumeQueue.java b/store/src/main/java/org/apache/rocketmq/store/ConsumeQueue.java index 0d698dacfe1..611cfd4620b 100644 --- a/store/src/main/java/org/apache/rocketmq/store/ConsumeQueue.java +++ b/store/src/main/java/org/apache/rocketmq/store/ConsumeQueue.java @@ -136,7 +136,15 @@ public boolean load() { @Override public void recover() { final List mappedFiles = this.mappedFileQueue.getMappedFiles(); - if (!mappedFiles.isEmpty()) { + if (mappedFiles.isEmpty()) { + if (isExtReadEnable()) { + this.consumeQueueExt.recover(); + if (!this.consumeQueueExt.truncateAll()) { + log.warn("Failed to truncate all consume queue ext data during recovery, topic={}, queueId={}", + this.topic, this.queueId); + } + } + } else { int index = mappedFiles.size() - 3; if (index < 0) { @@ -423,10 +431,12 @@ public void truncateDirtyLogicFiles(long phyOffset, boolean deleteFile) { this.setMaxPhysicOffset(phyOffset); long maxExtAddr = 1; - boolean shouldDeleteFile = false; + boolean hasRetainedExt = false; + boolean cqFileDeletionFailed = false; while (true) { MappedFile mappedFile = this.mappedFileQueue.getLastMappedFile(); if (mappedFile != null) { + boolean shouldDeleteFile = false; ByteBuffer byteBuffer = mappedFile.sliceByteBuffer(); mappedFile.setWrotePosition(0); @@ -438,53 +448,43 @@ public void truncateDirtyLogicFiles(long phyOffset, boolean deleteFile) { int size = byteBuffer.getInt(); long tagsCode = byteBuffer.getLong(); - if (0 == i) { - if (offset >= phyOffset) { + if (offset < 0 || size <= 0 || offset >= phyOffset) { + if (0 == i) { shouldDeleteFile = true; - break; - } else { - int pos = i + CQ_STORE_UNIT_SIZE; - mappedFile.setWrotePosition(pos); - mappedFile.setCommittedPosition(pos); - mappedFile.setFlushedPosition(pos); - this.setMaxPhysicOffset(offset + size); - // This maybe not take effect, when not every consume queue has extend file. - if (isExtAddr(tagsCode)) { - maxExtAddr = tagsCode; - } } - } else { - - if (offset >= 0 && size > 0) { - - if (offset >= phyOffset) { - return; - } + break; + } - int pos = i + CQ_STORE_UNIT_SIZE; - mappedFile.setWrotePosition(pos); - mappedFile.setCommittedPosition(pos); - mappedFile.setFlushedPosition(pos); - this.setMaxPhysicOffset(offset + size); - if (isExtAddr(tagsCode)) { - maxExtAddr = tagsCode; - } + int pos = i + CQ_STORE_UNIT_SIZE; + mappedFile.setWrotePosition(pos); + mappedFile.setCommittedPosition(pos); + mappedFile.setFlushedPosition(pos); + this.setMaxPhysicOffset(offset + size); + // This maybe not take effect, when not every consume queue has extend file. + long logicOffset = mappedFile.getFileFromOffset() + i; + if (logicOffset >= this.minLogicOffset && isExtAddr(tagsCode)) { + maxExtAddr = tagsCode; + hasRetainedExt = true; + } - if (pos == logicFileSize) { - return; - } - } else { - return; - } + if (pos == logicFileSize) { + break; } } if (shouldDeleteFile) { if (deleteFile) { + String mappedFilePath = mappedFile.getFileName(); this.mappedFileQueue.deleteLastMappedFile(); + if (new File(mappedFilePath).exists()) { + cqFileDeletionFailed = true; + log.warn("Consume queue file still exists after deletion: {}", mappedFilePath); + } } else { this.mappedFileQueue.deleteExpiredFile(Collections.singletonList(this.mappedFileQueue.getLastMappedFile())); } + } else { + break; } } else { @@ -492,9 +492,50 @@ public void truncateDirtyLogicFiles(long phyOffset, boolean deleteFile) { } } - if (isExtReadEnable()) { - this.consumeQueueExt.truncateByMaxAddress(maxExtAddr); + if (deleteFile && isExtReadEnable()) { + if (cqFileDeletionFailed) { + log.warn("Skip truncating consume queue ext because a consume queue file was not deleted"); + return; + } + if (hasRetainedExt && this.consumeQueueExt.get(maxExtAddr) == null) { + hasRetainedExt = false; + } + if (!hasRetainedExt) { + maxExtAddr = findLastRetainedExtAddress(); + hasRetainedExt = isExtAddr(maxExtAddr); + } + if (hasRetainedExt) { + this.consumeQueueExt.truncateByMaxAddress(maxExtAddr); + } else { + if (!this.consumeQueueExt.truncateAll()) { + log.warn("Failed to truncate all consume queue ext data, topic={}, queueId={}", + this.topic, this.queueId); + } + } + } + } + + private long findLastRetainedExtAddress() { + List mappedFiles = this.mappedFileQueue.getMappedFiles(); + for (int fileIndex = mappedFiles.size() - 1; fileIndex >= 0; fileIndex--) { + MappedFile mappedFile = mappedFiles.get(fileIndex); + ByteBuffer byteBuffer = mappedFile.sliceByteBuffer(); + for (int position = mappedFile.getWrotePosition() - CQ_STORE_UNIT_SIZE; + position >= 0; position -= CQ_STORE_UNIT_SIZE) { + long logicOffset = mappedFile.getFileFromOffset() + position; + if (logicOffset < this.minLogicOffset) { + return 1; + } + long offset = byteBuffer.getLong(position); + int size = byteBuffer.getInt(position + Long.BYTES); + long tagsCode = byteBuffer.getLong(position + MSG_TAG_OFFSET_INDEX); + if (offset >= 0 && size > 0 && isExtAddr(tagsCode) + && this.consumeQueueExt.get(tagsCode) != null) { + return tagsCode; + } + } } + return 1; } @Override diff --git a/store/src/main/java/org/apache/rocketmq/store/ConsumeQueueExt.java b/store/src/main/java/org/apache/rocketmq/store/ConsumeQueueExt.java index 641f672bba6..8d4af9d48ae 100644 --- a/store/src/main/java/org/apache/rocketmq/store/ConsumeQueueExt.java +++ b/store/src/main/java/org/apache/rocketmq/store/ConsumeQueueExt.java @@ -47,6 +47,7 @@ public class ConsumeQueueExt { private final String storePath; private final int mappedFileSize; private ByteBuffer tempContainer; + private volatile boolean truncateAllPending; public static final int END_BLANK_DATA_LENGTH = 4; @@ -228,6 +229,11 @@ public boolean get(final long address, final CqExtUnit cqExtUnit) { * @return success: < 0: fail: >=0 */ public long put(final CqExtUnit cqExtUnit) { + if (this.truncateAllPending) { + log.warn("Skip saving consume queue ext while truncating all data is pending, {}", cqExtUnit); + return 1; + } + final int retryTimes = 3; try { int size = cqExtUnit.calcUnitSize(); @@ -405,6 +411,34 @@ public void truncateByMaxAddress(final long maxAddress) { this.mappedFileQueue.truncateDirtyFiles(realOffset + cqExtUnit.getSize()); } + /** + * Delete all consume queue extension data when no consume queue entry retains an extension address. + */ + public synchronized boolean truncateAll() { + log.info("Truncate all consume queue ext data."); + this.truncateAllPending = true; + List deletedFiles = new ArrayList<>(); + for (MappedFile mappedFile : new ArrayList<>(this.mappedFileQueue.getMappedFiles())) { + boolean destroyed = mappedFile.destroy(1000 * 3); + boolean fileExists = new File(mappedFile.getFileName()).exists(); + if (destroyed && !fileExists) { + deletedFiles.add(mappedFile); + } else { + log.warn("Consume queue ext file remains after truncating all data, file={}, destroyed={}", + mappedFile.getFileName(), destroyed); + } + } + this.mappedFileQueue.deleteExpiredFile(deletedFiles); + if (!this.mappedFileQueue.getMappedFiles().isEmpty()) { + return false; + } + + this.mappedFileQueue.setFlushedWhere(0); + this.mappedFileQueue.setCommittedWhere(0); + this.truncateAllPending = false; + return true; + } + /** * flush buffer to file. */ diff --git a/store/src/test/java/org/apache/rocketmq/store/ConsumeQueueTest.java b/store/src/test/java/org/apache/rocketmq/store/ConsumeQueueTest.java index e8e3797d021..93b3d96ffd5 100644 --- a/store/src/test/java/org/apache/rocketmq/store/ConsumeQueueTest.java +++ b/store/src/test/java/org/apache/rocketmq/store/ConsumeQueueTest.java @@ -19,6 +19,7 @@ import java.io.File; import java.io.IOException; +import java.lang.reflect.Field; import java.lang.reflect.Method; import java.net.InetAddress; import java.net.InetSocketAddress; @@ -38,6 +39,8 @@ import org.apache.rocketmq.common.message.MessageDecoder; import org.apache.rocketmq.common.message.MessageExtBrokerInner; import org.apache.rocketmq.store.config.MessageStoreConfig; +import org.apache.rocketmq.store.config.StorePathConfigHelper; +import org.apache.rocketmq.store.logfile.MappedFile; import org.apache.rocketmq.store.queue.ConsumeQueueInterface; import org.apache.rocketmq.store.queue.CqUnit; import org.apache.rocketmq.store.queue.ReferredIterator; @@ -775,4 +778,495 @@ public void testCorrectMinOffsetAfterAllFilesDeleted() throws IOException { FileUtils.deleteQuietly(tmpDir); } } + + @Test + public void testTruncateRetainsPreviousFullFileAfterDeletingDirtyTailFile() throws IOException { + File tmpDir = Files.createTempDirectory("truncate-cq-tail-files").toFile(); + MessageStoreConfig storeConfig = new MessageStoreConfig(); + storeConfig.setStorePathRootDir(tmpDir.getAbsolutePath()); + storeConfig.setMappedFileSizeConsumeQueue(2 * ConsumeQueue.CQ_STORE_UNIT_SIZE); + storeConfig.setEnableConsumeQueueExt(false); + DefaultMessageStore messageStore = Mockito.mock(DefaultMessageStore.class); + Mockito.when(messageStore.getMessageStoreConfig()).thenReturn(storeConfig); + Mockito.when(messageStore.getRunningFlags()).thenReturn(new RunningFlags()); + Mockito.when(messageStore.getStoreCheckpoint()).thenReturn(Mockito.mock(StoreCheckpoint.class)); + + ConsumeQueue consumeQueue = new ConsumeQueue("truncateTopic", 0, tmpDir.getAbsolutePath(), + storeConfig.getMappedFileSizeConsumeQueue(), messageStore); + try { + long[] physicalOffsets = {0, 10, 1000, 1010}; + for (int i = 0; i < physicalOffsets.length; i++) { + DispatchRequest request = new DispatchRequest("truncateTopic", 0, physicalOffsets[i], 10, + 0, 0, i, null, null, 0, 0, null); + consumeQueue.putMessagePositionInfoWrapper(request); + } + Assert.assertEquals(4, consumeQueue.getMaxOffsetInQueue()); + + consumeQueue.truncateDirtyLogicFiles(500); + + Assert.assertEquals(2, consumeQueue.getMaxOffsetInQueue()); + } finally { + consumeQueue.destroy(); + FileUtils.deleteQuietly(tmpDir); + } + } + + @Test + public void testTruncateDirtyLogicFilesTruncatesConsumeQueueExtAndSurvivesReload() throws IOException { + File tmpDir = Files.createTempDirectory("truncate-cq-ext").toFile(); + MessageStoreConfig storeConfig = new MessageStoreConfig(); + storeConfig.setStorePathRootDir(tmpDir.getAbsolutePath()); + storeConfig.setMappedFileSizeConsumeQueue(4 * ConsumeQueue.CQ_STORE_UNIT_SIZE); + storeConfig.setMappedFileSizeConsumeQueueExt(10 * ConsumeQueueExt.CqExtUnit.MIN_EXT_UNIT_SIZE); + storeConfig.setEnableConsumeQueueExt(true); + DefaultMessageStore messageStore = Mockito.mock(DefaultMessageStore.class); + Mockito.when(messageStore.getMessageStoreConfig()).thenReturn(storeConfig); + Mockito.when(messageStore.getRunningFlags()).thenReturn(new RunningFlags()); + Mockito.when(messageStore.getStoreCheckpoint()).thenReturn(Mockito.mock(StoreCheckpoint.class)); + + ConsumeQueue consumeQueue = new ConsumeQueue("truncateExtTopic", 0, tmpDir.getAbsolutePath(), + storeConfig.getMappedFileSizeConsumeQueue(), messageStore); + ConsumeQueue reloadedConsumeQueue = null; + try { + for (int i = 0; i < 4; i++) { + DispatchRequest request = new DispatchRequest("truncateExtTopic", 0, 100L * i, 10, + 100 + i, 1000 + i, i, null, null, 0, 0, null); + consumeQueue.putMessagePositionInfoWrapper(request); + } + long truncatedExtAddress = getRawTagsCode(consumeQueue, 2); + + consumeQueue.truncateDirtyLogicFiles(200); + + for (int i = 2; i < 4; i++) { + DispatchRequest replacement = new DispatchRequest("truncateExtTopic", 0, 100L * i, 10, + 200 + i, 2000 + i, i, null, null, 0, 0, null); + consumeQueue.putMessagePositionInfoWrapper(replacement); + } + long replacementExtAddress = getRawTagsCode(consumeQueue, 2); + consumeQueue.flush(0); + + reloadedConsumeQueue = new ConsumeQueue("truncateExtTopic", 0, tmpDir.getAbsolutePath(), + storeConfig.getMappedFileSizeConsumeQueue(), messageStore); + Assert.assertTrue(reloadedConsumeQueue.load()); + reloadedConsumeQueue.recover(); + + Assert.assertEquals(202, reloadedConsumeQueue.getExt(truncatedExtAddress).getTagsCode()); + Assert.assertEquals(truncatedExtAddress, replacementExtAddress); + } finally { + consumeQueue.destroy(); + if (reloadedConsumeQueue != null) { + reloadedConsumeQueue.destroy(); + } + FileUtils.deleteQuietly(tmpDir); + } + } + + @Test + public void testTruncateAllConsumeQueueExtSurvivesReloadAndReusesFirstAddress() throws IOException { + File tmpDir = Files.createTempDirectory("truncate-all-cq-ext").toFile(); + MessageStoreConfig storeConfig = new MessageStoreConfig(); + storeConfig.setStorePathRootDir(tmpDir.getAbsolutePath()); + storeConfig.setMappedFileSizeConsumeQueue(4 * ConsumeQueue.CQ_STORE_UNIT_SIZE); + storeConfig.setMappedFileSizeConsumeQueueExt(10 * ConsumeQueueExt.CqExtUnit.MIN_EXT_UNIT_SIZE); + storeConfig.setEnableConsumeQueueExt(true); + DefaultMessageStore messageStore = Mockito.mock(DefaultMessageStore.class); + Mockito.when(messageStore.getMessageStoreConfig()).thenReturn(storeConfig); + Mockito.when(messageStore.getRunningFlags()).thenReturn(new RunningFlags()); + Mockito.when(messageStore.getStoreCheckpoint()).thenReturn(Mockito.mock(StoreCheckpoint.class)); + + ConsumeQueue consumeQueue = new ConsumeQueue("truncateAllExtTopic", 0, tmpDir.getAbsolutePath(), + storeConfig.getMappedFileSizeConsumeQueue(), messageStore); + ConsumeQueue emptyReloadedConsumeQueue = null; + ConsumeQueue replacementReloadedConsumeQueue = null; + try { + for (int i = 0; i < 2; i++) { + DispatchRequest request = new DispatchRequest("truncateAllExtTopic", 0, 100L * i, 10, + 100 + i, 1000 + i, i, null, null, 0, 0, null); + consumeQueue.putMessagePositionInfoWrapper(request); + } + long firstExtAddress = getRawTagsCode(consumeQueue, 0); + consumeQueue.flush(0); + + consumeQueue.truncateDirtyLogicFiles(0); + consumeQueue.flush(0); + + emptyReloadedConsumeQueue = new ConsumeQueue("truncateAllExtTopic", 0, tmpDir.getAbsolutePath(), + storeConfig.getMappedFileSizeConsumeQueue(), messageStore); + Assert.assertTrue(emptyReloadedConsumeQueue.load()); + emptyReloadedConsumeQueue.recover(); + Assert.assertEquals(0, emptyReloadedConsumeQueue.getMaxOffsetInQueue()); + Assert.assertNull(emptyReloadedConsumeQueue.getExt(firstExtAddress)); + + DispatchRequest replacement = new DispatchRequest("truncateAllExtTopic", 0, 0, 10, + 200, 2000, 0, null, null, 0, 0, null); + emptyReloadedConsumeQueue.putMessagePositionInfoWrapper(replacement); + long replacementExtAddress = getRawTagsCode(emptyReloadedConsumeQueue, 0); + Assert.assertEquals(firstExtAddress, replacementExtAddress); + emptyReloadedConsumeQueue.flush(0); + + replacementReloadedConsumeQueue = new ConsumeQueue("truncateAllExtTopic", 0, tmpDir.getAbsolutePath(), + storeConfig.getMappedFileSizeConsumeQueue(), messageStore); + Assert.assertTrue(replacementReloadedConsumeQueue.load()); + replacementReloadedConsumeQueue.recover(); + Assert.assertEquals(200, replacementReloadedConsumeQueue.getExt(firstExtAddress).getTagsCode()); + } finally { + consumeQueue.destroy(); + if (emptyReloadedConsumeQueue != null) { + emptyReloadedConsumeQueue.destroy(); + } + if (replacementReloadedConsumeQueue != null) { + replacementReloadedConsumeQueue.destroy(); + } + FileUtils.deleteQuietly(tmpDir); + } + } + + @Test + public void testTruncateDeletesEmptyTailAndRetainsPreviousFileExt() throws Exception { + File tmpDir = Files.createTempDirectory("truncate-empty-cq-tail").toFile(); + MessageStoreConfig storeConfig = new MessageStoreConfig(); + storeConfig.setStorePathRootDir(tmpDir.getAbsolutePath()); + storeConfig.setMappedFileSizeConsumeQueue(2 * ConsumeQueue.CQ_STORE_UNIT_SIZE); + storeConfig.setMappedFileSizeConsumeQueueExt(10 * ConsumeQueueExt.CqExtUnit.MIN_EXT_UNIT_SIZE); + storeConfig.setEnableConsumeQueueExt(true); + DefaultMessageStore messageStore = Mockito.mock(DefaultMessageStore.class); + Mockito.when(messageStore.getMessageStoreConfig()).thenReturn(storeConfig); + Mockito.when(messageStore.getRunningFlags()).thenReturn(new RunningFlags()); + Mockito.when(messageStore.getStoreCheckpoint()).thenReturn(Mockito.mock(StoreCheckpoint.class)); + + ConsumeQueue consumeQueue = new ConsumeQueue("truncateEmptyTailTopic", 0, tmpDir.getAbsolutePath(), + storeConfig.getMappedFileSizeConsumeQueue(), messageStore); + ConsumeQueue reloadedConsumeQueue = null; + try { + for (int i = 0; i < 2; i++) { + DispatchRequest request = new DispatchRequest("truncateEmptyTailTopic", 0, 100L * i, 10, + 100 + i, 1000 + i, i, null, null, 0, 0, null); + consumeQueue.putMessagePositionInfoWrapper(request); + } + long lastRetainedExtAddress = getRawTagsCode(consumeQueue, 1); + MappedFileQueue mappedFileQueue = getMappedFileQueue(consumeQueue); + Assert.assertNotNull(mappedFileQueue.getLastMappedFile(0)); + Assert.assertEquals(2, mappedFileQueue.getMappedFiles().size()); + Assert.assertEquals(0, mappedFileQueue.getLastMappedFile().getWrotePosition()); + + consumeQueue.truncateDirtyLogicFiles(500); + + Assert.assertEquals(1, mappedFileQueue.getMappedFiles().size()); + Assert.assertEquals(2, consumeQueue.getMaxOffsetInQueue()); + Assert.assertEquals(101, consumeQueue.getExt(lastRetainedExtAddress).getTagsCode()); + consumeQueue.flush(0); + + reloadedConsumeQueue = new ConsumeQueue("truncateEmptyTailTopic", 0, tmpDir.getAbsolutePath(), + storeConfig.getMappedFileSizeConsumeQueue(), messageStore); + Assert.assertTrue(reloadedConsumeQueue.load()); + reloadedConsumeQueue.recover(); + Assert.assertEquals(2, reloadedConsumeQueue.getMaxOffsetInQueue()); + Assert.assertEquals(101, reloadedConsumeQueue.getExt(lastRetainedExtAddress).getTagsCode()); + } finally { + consumeQueue.destroy(); + if (reloadedConsumeQueue != null) { + reloadedConsumeQueue.destroy(); + } + FileUtils.deleteQuietly(tmpDir); + } + } + + @Test + public void testTruncateWithoutDeletingFilesRetainsExtForReload() throws Exception { + File tmpDir = Files.createTempDirectory("truncate-cq-without-delete").toFile(); + MessageStoreConfig storeConfig = new MessageStoreConfig(); + storeConfig.setStorePathRootDir(tmpDir.getAbsolutePath()); + storeConfig.setMappedFileSizeConsumeQueue(2 * ConsumeQueue.CQ_STORE_UNIT_SIZE); + storeConfig.setMappedFileSizeConsumeQueueExt(10 * ConsumeQueueExt.CqExtUnit.MIN_EXT_UNIT_SIZE); + storeConfig.setEnableConsumeQueueExt(true); + DefaultMessageStore messageStore = Mockito.mock(DefaultMessageStore.class); + Mockito.when(messageStore.getMessageStoreConfig()).thenReturn(storeConfig); + Mockito.when(messageStore.getRunningFlags()).thenReturn(new RunningFlags()); + Mockito.when(messageStore.getStoreCheckpoint()).thenReturn(Mockito.mock(StoreCheckpoint.class)); + + ConsumeQueue consumeQueue = new ConsumeQueue("truncateWithoutDeleteTopic", 0, tmpDir.getAbsolutePath(), + storeConfig.getMappedFileSizeConsumeQueue(), messageStore); + ConsumeQueue reloadedConsumeQueue = null; + MappedFile removedTailFile = null; + try { + for (int i = 0; i < 4; i++) { + DispatchRequest request = new DispatchRequest("truncateWithoutDeleteTopic", 0, 100L * i, 10, + 100 + i, 1000 + i, i, null, null, 0, 0, null); + consumeQueue.putMessagePositionInfoWrapper(request); + } + long dirtyExtAddress = getRawTagsCode(consumeQueue, 2); + consumeQueue.flush(0); + MappedFileQueue mappedFileQueue = getMappedFileQueue(consumeQueue); + removedTailFile = mappedFileQueue.getLastMappedFile(); + + consumeQueue.truncateDirtyLogicFiles(200, false); + + Assert.assertEquals(1, mappedFileQueue.getMappedFiles().size()); + Assert.assertEquals(2, consumeQueue.getMaxOffsetInQueue()); + Assert.assertEquals(102, consumeQueue.getExt(dirtyExtAddress).getTagsCode()); + consumeQueue.flush(0); + + reloadedConsumeQueue = new ConsumeQueue("truncateWithoutDeleteTopic", 0, tmpDir.getAbsolutePath(), + storeConfig.getMappedFileSizeConsumeQueue(), messageStore); + Assert.assertTrue(reloadedConsumeQueue.load()); + reloadedConsumeQueue.recover(); + Assert.assertEquals(4, reloadedConsumeQueue.getMaxOffsetInQueue()); + Assert.assertEquals(102, reloadedConsumeQueue.getExt(dirtyExtAddress).getTagsCode()); + } finally { + consumeQueue.destroy(); + if (reloadedConsumeQueue != null) { + reloadedConsumeQueue.destroy(); + } + if (removedTailFile != null) { + removedTailFile.destroy(1000); + } + FileUtils.deleteQuietly(tmpDir); + } + } + + @Test + public void testTruncateIgnoresExpiredExtBeforeMinLogicOffset() throws IOException { + File tmpDir = Files.createTempDirectory("truncate-expired-cq-ext").toFile(); + MessageStoreConfig storeConfig = new MessageStoreConfig(); + storeConfig.setStorePathRootDir(tmpDir.getAbsolutePath()); + storeConfig.setMappedFileSizeConsumeQueue(4 * ConsumeQueue.CQ_STORE_UNIT_SIZE); + storeConfig.setMappedFileSizeConsumeQueueExt(10 * ConsumeQueueExt.CqExtUnit.MIN_EXT_UNIT_SIZE); + storeConfig.setEnableConsumeQueueExt(true); + DefaultMessageStore messageStore = Mockito.mock(DefaultMessageStore.class); + Mockito.when(messageStore.getMessageStoreConfig()).thenReturn(storeConfig); + Mockito.when(messageStore.getRunningFlags()).thenReturn(new RunningFlags()); + Mockito.when(messageStore.getStoreCheckpoint()).thenReturn(Mockito.mock(StoreCheckpoint.class)); + + ConsumeQueue consumeQueue = new ConsumeQueue("truncateExpiredExtTopic", 0, tmpDir.getAbsolutePath(), + storeConfig.getMappedFileSizeConsumeQueue(), messageStore); + try { + DispatchRequest expiredRequest = new DispatchRequest("truncateExpiredExtTopic", 0, 0, 10, + 100, 1000, 0, null, null, 0, 0, null); + consumeQueue.putMessagePositionInfoWrapper(expiredRequest); + long expiredExtAddress = getRawTagsCode(consumeQueue, 0); + + storeConfig.setEnableConsumeQueueExt(false); + for (int i = 1; i < 3; i++) { + DispatchRequest request = new DispatchRequest("truncateExpiredExtTopic", 0, 100L * i, 10, + 100 + i, 1000 + i, i, null, null, 0, 0, null); + consumeQueue.putMessagePositionInfoWrapper(request); + } + consumeQueue.setMinLogicOffset(ConsumeQueue.CQ_STORE_UNIT_SIZE); + + consumeQueue.truncateDirtyLogicFiles(200); + + Assert.assertEquals(2, consumeQueue.getMaxOffsetInQueue()); + Assert.assertEquals(101, getRawTagsCode(consumeQueue, 1)); + Assert.assertNull(consumeQueue.getExt(expiredExtAddress)); + } finally { + consumeQueue.destroy(); + FileUtils.deleteQuietly(tmpDir); + } + } + + @Test + public void testRecoverClearsExtWhenConsumeQueueFilesAreMissing() throws Exception { + File tmpDir = Files.createTempDirectory("recover-orphan-cq-ext").toFile(); + MessageStoreConfig storeConfig = new MessageStoreConfig(); + storeConfig.setStorePathRootDir(tmpDir.getAbsolutePath()); + storeConfig.setMappedFileSizeConsumeQueue(4 * ConsumeQueue.CQ_STORE_UNIT_SIZE); + storeConfig.setMappedFileSizeConsumeQueueExt(10 * ConsumeQueueExt.CqExtUnit.MIN_EXT_UNIT_SIZE); + storeConfig.setEnableConsumeQueueExt(true); + DefaultMessageStore messageStore = Mockito.mock(DefaultMessageStore.class); + Mockito.when(messageStore.getMessageStoreConfig()).thenReturn(storeConfig); + Mockito.when(messageStore.getRunningFlags()).thenReturn(new RunningFlags()); + Mockito.when(messageStore.getStoreCheckpoint()).thenReturn(Mockito.mock(StoreCheckpoint.class)); + + ConsumeQueue consumeQueue = new ConsumeQueue("recoverOrphanExtTopic", 0, tmpDir.getAbsolutePath(), + storeConfig.getMappedFileSizeConsumeQueue(), messageStore); + ConsumeQueue reloadedConsumeQueue = null; + try { + for (int i = 0; i < 2; i++) { + DispatchRequest request = new DispatchRequest("recoverOrphanExtTopic", 0, 100L * i, 10, + 100 + i, 1000 + i, i, null, null, 0, 0, null); + consumeQueue.putMessagePositionInfoWrapper(request); + } + long orphanedExtAddress = getRawTagsCode(consumeQueue, 0); + consumeQueue.flush(0); + + MappedFileQueue mappedFileQueue = getMappedFileQueue(consumeQueue); + mappedFileQueue.destroy(); + Assert.assertTrue(mappedFileQueue.getMappedFiles().isEmpty()); + + reloadedConsumeQueue = new ConsumeQueue("recoverOrphanExtTopic", 0, tmpDir.getAbsolutePath(), + storeConfig.getMappedFileSizeConsumeQueue(), messageStore); + Assert.assertTrue(reloadedConsumeQueue.load()); + reloadedConsumeQueue.recover(); + + Assert.assertEquals(0, reloadedConsumeQueue.getMaxOffsetInQueue()); + Assert.assertNull(reloadedConsumeQueue.getExt(orphanedExtAddress)); + } finally { + consumeQueue.destroy(); + if (reloadedConsumeQueue != null) { + reloadedConsumeQueue.destroy(); + } + FileUtils.deleteQuietly(tmpDir); + } + } + + @Test + public void testTruncateKeepsExtWhenConsumeQueueTailDeletionFails() throws Exception { + File tmpDir = Files.createTempDirectory("truncate-cq-tail-delete-failure").toFile(); + MessageStoreConfig storeConfig = new MessageStoreConfig(); + storeConfig.setStorePathRootDir(tmpDir.getAbsolutePath()); + storeConfig.setMappedFileSizeConsumeQueue(2 * ConsumeQueue.CQ_STORE_UNIT_SIZE); + storeConfig.setMappedFileSizeConsumeQueueExt(10 * ConsumeQueueExt.CqExtUnit.MIN_EXT_UNIT_SIZE); + storeConfig.setEnableConsumeQueueExt(true); + DefaultMessageStore messageStore = Mockito.mock(DefaultMessageStore.class); + Mockito.when(messageStore.getMessageStoreConfig()).thenReturn(storeConfig); + Mockito.when(messageStore.getRunningFlags()).thenReturn(new RunningFlags()); + Mockito.when(messageStore.getStoreCheckpoint()).thenReturn(Mockito.mock(StoreCheckpoint.class)); + + ConsumeQueue consumeQueue = new ConsumeQueue("truncateDeleteFailureTopic", 0, tmpDir.getAbsolutePath(), + storeConfig.getMappedFileSizeConsumeQueue(), messageStore); + ConsumeQueue reloadedConsumeQueue = null; + MappedFile retainedTailFile = null; + SelectMappedBufferResult heldBuffer = null; + try { + for (int i = 0; i < 4; i++) { + DispatchRequest request = new DispatchRequest("truncateDeleteFailureTopic", 0, 100L * i, 10, + 100 + i, 1000 + i, i, null, null, 0, 0, null); + consumeQueue.putMessagePositionInfoWrapper(request); + } + long dirtyExtAddress = getRawTagsCode(consumeQueue, 2); + consumeQueue.flush(0); + + MappedFileQueue mappedFileQueue = getMappedFileQueue(consumeQueue); + retainedTailFile = mappedFileQueue.getLastMappedFile(); + File retainedTailPath = new File(retainedTailFile.getFileName()); + heldBuffer = retainedTailFile.selectMappedBuffer(0, ConsumeQueue.CQ_STORE_UNIT_SIZE); + Assert.assertNotNull(heldBuffer); + + consumeQueue.truncateDirtyLogicFiles(200); + + Assert.assertTrue(retainedTailPath.exists()); + Assert.assertEquals(1, mappedFileQueue.getMappedFiles().size()); + Assert.assertEquals(2, consumeQueue.getMaxOffsetInQueue()); + Assert.assertEquals(102, consumeQueue.getExt(dirtyExtAddress).getTagsCode()); + + heldBuffer.release(); + heldBuffer = null; + + reloadedConsumeQueue = new ConsumeQueue("truncateDeleteFailureTopic", 0, tmpDir.getAbsolutePath(), + storeConfig.getMappedFileSizeConsumeQueue(), messageStore); + Assert.assertTrue(reloadedConsumeQueue.load()); + reloadedConsumeQueue.recover(); + + Assert.assertEquals(4, reloadedConsumeQueue.getMaxOffsetInQueue()); + Assert.assertEquals(102, reloadedConsumeQueue.getExt(dirtyExtAddress).getTagsCode()); + } finally { + if (heldBuffer != null) { + heldBuffer.release(); + } + consumeQueue.destroy(); + if (reloadedConsumeQueue != null) { + reloadedConsumeQueue.destroy(); + } + if (retainedTailFile != null) { + retainedTailFile.destroy(1000); + } + FileUtils.deleteQuietly(tmpDir); + } + } + + @Test + public void testTruncateAllConsumeQueueExtRetriesAfterMappedFileRelease() throws Exception { + File tmpDir = Files.createTempDirectory("truncate-all-cq-ext-retry").toFile(); + String topic = "truncateAllExtRetryTopic"; + String extStorePath = StorePathConfigHelper.getStorePathConsumeQueueExt(tmpDir.getAbsolutePath()); + int mappedFileSize = 2 * ConsumeQueueExt.CqExtUnit.MIN_EXT_UNIT_SIZE; + ConsumeQueueExt consumeQueueExt = new ConsumeQueueExt(topic, 0, extStorePath, mappedFileSize, 0); + ConsumeQueueExt reloadedConsumeQueueExt = null; + SelectMappedBufferResult heldBuffer = null; + try { + long firstExtAddress = consumeQueueExt.put( + new ConsumeQueueExt.CqExtUnit(100L, 1000L, null)); + long secondExtAddress = consumeQueueExt.put( + new ConsumeQueueExt.CqExtUnit(101L, 1001L, null)); + Assert.assertTrue(ConsumeQueueExt.isExtAddr(firstExtAddress)); + Assert.assertTrue(ConsumeQueueExt.isExtAddr(secondExtAddress)); + Assert.assertNotEquals(firstExtAddress, secondExtAddress); + consumeQueueExt.flush(0); + + MappedFileQueue mappedFileQueue = getMappedFileQueue(consumeQueueExt); + Assert.assertEquals(2, mappedFileQueue.getMappedFiles().size()); + MappedFile firstMappedFile = mappedFileQueue.getMappedFiles().get(0); + MappedFile retainedMappedFile = mappedFileQueue.getMappedFiles().get(1); + File firstMappedFilePath = new File(firstMappedFile.getFileName()); + File retainedMappedFilePath = new File(retainedMappedFile.getFileName()); + long flushedWhereBeforeTruncate = mappedFileQueue.getFlushedWhere(); + Assert.assertTrue(flushedWhereBeforeTruncate > 0); + heldBuffer = retainedMappedFile.selectMappedBuffer(0, ConsumeQueueExt.CqExtUnit.MIN_EXT_UNIT_SIZE); + Assert.assertNotNull(heldBuffer); + + Assert.assertFalse(consumeQueueExt.truncateAll()); + Assert.assertFalse(firstMappedFilePath.exists()); + Assert.assertTrue(retainedMappedFilePath.exists()); + Assert.assertEquals(1, mappedFileQueue.getMappedFiles().size()); + Assert.assertSame(retainedMappedFile, mappedFileQueue.getMappedFiles().get(0)); + Assert.assertEquals(flushedWhereBeforeTruncate, mappedFileQueue.getFlushedWhere()); + Assert.assertEquals(1, consumeQueueExt.put( + new ConsumeQueueExt.CqExtUnit(200L, 2000L, null))); + Assert.assertEquals(flushedWhereBeforeTruncate, mappedFileQueue.getFlushedWhere()); + Assert.assertEquals(1, mappedFileQueue.getMappedFiles().size()); + + heldBuffer.release(); + heldBuffer = null; + Assert.assertTrue(retainedMappedFilePath.exists()); + + Assert.assertTrue(consumeQueueExt.truncateAll()); + Assert.assertFalse(retainedMappedFilePath.exists()); + Assert.assertTrue(mappedFileQueue.getMappedFiles().isEmpty()); + Assert.assertEquals(0, mappedFileQueue.getFlushedWhere()); + Assert.assertEquals(0, mappedFileQueue.getCommittedWhere()); + + long replacementExtAddress = consumeQueueExt.put( + new ConsumeQueueExt.CqExtUnit(200L, 2000L, null)); + Assert.assertEquals(firstExtAddress, replacementExtAddress); + consumeQueueExt.flush(0); + + reloadedConsumeQueueExt = new ConsumeQueueExt(topic, 0, extStorePath, mappedFileSize, 0); + Assert.assertTrue(reloadedConsumeQueueExt.load()); + reloadedConsumeQueueExt.recover(); + ConsumeQueueExt.CqExtUnit replacement = reloadedConsumeQueueExt.get(replacementExtAddress); + Assert.assertNotNull(replacement); + Assert.assertEquals(200, replacement.getTagsCode()); + } finally { + if (heldBuffer != null) { + heldBuffer.release(); + } + consumeQueueExt.destroy(); + if (reloadedConsumeQueueExt != null) { + reloadedConsumeQueueExt.destroy(); + } + FileUtils.deleteQuietly(tmpDir); + } + } + + private MappedFileQueue getMappedFileQueue(ConsumeQueue consumeQueue) throws ReflectiveOperationException { + Field mappedFileQueueField = ConsumeQueue.class.getDeclaredField("mappedFileQueue"); + mappedFileQueueField.setAccessible(true); + return (MappedFileQueue) mappedFileQueueField.get(consumeQueue); + } + + private MappedFileQueue getMappedFileQueue(ConsumeQueueExt consumeQueueExt) throws ReflectiveOperationException { + Field mappedFileQueueField = ConsumeQueueExt.class.getDeclaredField("mappedFileQueue"); + mappedFileQueueField.setAccessible(true); + return (MappedFileQueue) mappedFileQueueField.get(consumeQueueExt); + } + + private long getRawTagsCode(ConsumeQueue consumeQueue, long queueOffset) { + SelectMappedBufferResult result = consumeQueue.getIndexBuffer(queueOffset); + Assert.assertNotNull(result); + try { + return result.getByteBuffer().getLong(12); + } finally { + result.release(); + } + } }