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 1/2] [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(); + } + } } From 82bc88273bb4a7f277735621bbc45a70f4d875e3 Mon Sep 17 00:00:00 2001 From: Rui <1685901819@qq.com> Date: Sat, 15 Aug 2026 21:46:43 +0800 Subject: [PATCH 2/2] [ISSUE #10755] Complete pending ConsumeQueueExt cleanup Signed-off-by: Rui <1685901819@qq.com> --- .../apache/rocketmq/store/ConsumeQueue.java | 41 +++-- .../rocketmq/store/ConsumeQueueExt.java | 19 ++- .../rocketmq/store/ConsumeQueueTest.java | 149 +++++++++++++++++- 3 files changed, 177 insertions(+), 32 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 611cfd4620b..c160d37959e 100644 --- a/store/src/main/java/org/apache/rocketmq/store/ConsumeQueue.java +++ b/store/src/main/java/org/apache/rocketmq/store/ConsumeQueue.java @@ -139,10 +139,7 @@ public void recover() { 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); - } + truncateConsumeQueueExt(1); } } else { @@ -204,8 +201,7 @@ public void recover() { if (isExtReadEnable()) { this.consumeQueueExt.recover(); - log.info("Truncate consume queue extend file by max {}", maxExtAddr); - this.consumeQueueExt.truncateByMaxAddress(maxExtAddr); + truncateConsumeQueueExt(maxExtAddr); } } } @@ -431,7 +427,6 @@ public void truncateDirtyLogicFiles(long phyOffset, boolean deleteFile) { this.setMaxPhysicOffset(phyOffset); long maxExtAddr = 1; - boolean hasRetainedExt = false; boolean cqFileDeletionFailed = false; while (true) { MappedFile mappedFile = this.mappedFileQueue.getLastMappedFile(); @@ -464,7 +459,6 @@ public void truncateDirtyLogicFiles(long phyOffset, boolean deleteFile) { long logicOffset = mappedFile.getFileFromOffset() + i; if (logicOffset >= this.minLogicOffset && isExtAddr(tagsCode)) { maxExtAddr = tagsCode; - hasRetainedExt = true; } if (pos == logicFileSize) { @@ -497,21 +491,22 @@ public void truncateDirtyLogicFiles(long phyOffset, boolean deleteFile) { 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); - } - } + truncateConsumeQueueExt(maxExtAddr); + } + } + + private void truncateConsumeQueueExt(long maxExtAddr) { + if (this.consumeQueueExt.getTotalSize() == 0) { + return; + } + if (!isExtAddr(maxExtAddr) || this.consumeQueueExt.get(maxExtAddr) == null) { + maxExtAddr = findLastRetainedExtAddress(); + } + if (isExtAddr(maxExtAddr)) { + 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); } } 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 8d4af9d48ae..c0600b9bbd1 100644 --- a/store/src/main/java/org/apache/rocketmq/store/ConsumeQueueExt.java +++ b/store/src/main/java/org/apache/rocketmq/store/ConsumeQueueExt.java @@ -39,6 +39,7 @@ */ public class ConsumeQueueExt { private static final Logger log = LoggerFactory.getLogger(LoggerName.STORE_LOGGER_NAME); + private static final long TRUNCATE_ALL_RETRY_INTERVAL_MILLIS = 1000; private final MappedFileQueue mappedFileQueue; private final String topic; @@ -48,6 +49,7 @@ public class ConsumeQueueExt { private final int mappedFileSize; private ByteBuffer tempContainer; private volatile boolean truncateAllPending; + private volatile long nextTruncateAllRetryTimestamp; public static final int END_BLANK_DATA_LENGTH = 4; @@ -229,8 +231,7 @@ 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); + if (this.truncateAllPending && !retryTruncateAllIfDue()) { return 1; } @@ -283,6 +284,16 @@ public long put(final CqExtUnit cqExtUnit) { return 1; } + private synchronized boolean retryTruncateAllIfDue() { + if (!this.truncateAllPending) { + return true; + } + if (System.currentTimeMillis() < this.nextTruncateAllRetryTimestamp) { + return false; + } + return truncateAll(); + } + protected void fullFillToEnd(final MappedFile mappedFile, final int wrotePosition) { ByteBuffer mappedFileBuffer = mappedFile.sliceByteBuffer(); mappedFileBuffer.position(wrotePosition); @@ -417,6 +428,7 @@ public void truncateByMaxAddress(final long maxAddress) { public synchronized boolean truncateAll() { log.info("Truncate all consume queue ext data."); this.truncateAllPending = true; + this.nextTruncateAllRetryTimestamp = Long.MAX_VALUE; List deletedFiles = new ArrayList<>(); for (MappedFile mappedFile : new ArrayList<>(this.mappedFileQueue.getMappedFiles())) { boolean destroyed = mappedFile.destroy(1000 * 3); @@ -430,11 +442,14 @@ public synchronized boolean truncateAll() { } this.mappedFileQueue.deleteExpiredFile(deletedFiles); if (!this.mappedFileQueue.getMappedFiles().isEmpty()) { + this.nextTruncateAllRetryTimestamp = + System.currentTimeMillis() + TRUNCATE_ALL_RETRY_INTERVAL_MILLIS; return false; } this.mappedFileQueue.setFlushedWhere(0); this.mappedFileQueue.setCommittedWhere(0); + this.nextTruncateAllRetryTimestamp = 0; this.truncateAllPending = false; return true; } 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 93b3d96ffd5..292c53adc72 100644 --- a/store/src/test/java/org/apache/rocketmq/store/ConsumeQueueTest.java +++ b/store/src/test/java/org/apache/rocketmq/store/ConsumeQueueTest.java @@ -30,6 +30,7 @@ import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentMap; import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicLong; import org.apache.commons.io.FileUtils; import org.apache.rocketmq.common.BrokerConfig; import org.apache.rocketmq.common.MixAll; @@ -1109,6 +1110,133 @@ public void testRecoverClearsExtWhenConsumeQueueFilesAreMissing() throws Excepti } } + @Test + public void testRecoverCompletesPendingExtCleanupAfterFallbackDispatch() throws Exception { + File tmpDir = Files.createTempDirectory("recover-pending-cq-ext").toFile(); + String topic = "recoverPendingExtTopic"; + MessageStoreConfig storeConfig = new MessageStoreConfig(); + storeConfig.setStorePathRootDir(tmpDir.getAbsolutePath()); + storeConfig.setMappedFileSizeConsumeQueue(4 * ConsumeQueue.CQ_STORE_UNIT_SIZE); + storeConfig.setMappedFileSizeConsumeQueueExt(2 * 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(topic, 0, tmpDir.getAbsolutePath(), + storeConfig.getMappedFileSizeConsumeQueue(), messageStore); + ConsumeQueue reloadedConsumeQueue = null; + SelectMappedBufferResult heldBuffer = null; + try { + for (int i = 0; i < 2; i++) { + DispatchRequest request = new DispatchRequest(topic, 0, 100L * i, 10, + 100 + i, 1000 + i, i, null, null, 0, 0, null); + consumeQueue.putMessagePositionInfoWrapper(request); + } + long firstExtAddress = getRawTagsCode(consumeQueue, 0); + long retainedExtAddress = getRawTagsCode(consumeQueue, 1); + consumeQueue.flush(0); + + ConsumeQueueExt consumeQueueExt = getConsumeQueueExt(consumeQueue); + MappedFileQueue extMappedFileQueue = getMappedFileQueue(consumeQueueExt); + Assert.assertEquals(2, extMappedFileQueue.getMappedFiles().size()); + MappedFile retainedMappedFile = extMappedFileQueue.getLastMappedFile(); + File retainedMappedFilePath = new File(retainedMappedFile.getFileName()); + heldBuffer = retainedMappedFile.selectMappedBuffer( + 0, ConsumeQueueExt.CqExtUnit.MIN_EXT_UNIT_SIZE); + Assert.assertNotNull(heldBuffer); + + consumeQueue.truncateDirtyLogicFiles(0); + + Assert.assertEquals(0, consumeQueue.getMaxOffsetInQueue()); + Assert.assertTrue(retainedMappedFilePath.exists()); + + DispatchRequest fallback = new DispatchRequest(topic, 0, 0, 10, + 200, 2000, 0, null, null, 0, 0, null); + consumeQueue.putMessagePositionInfoWrapper(fallback); + Assert.assertEquals(200, getRawTagsCode(consumeQueue, 0)); + consumeQueue.flush(0); + + heldBuffer.release(); + heldBuffer = null; + Assert.assertTrue(retainedMappedFilePath.exists()); + + reloadedConsumeQueue = new ConsumeQueue(topic, 0, tmpDir.getAbsolutePath(), + storeConfig.getMappedFileSizeConsumeQueue(), messageStore); + Assert.assertTrue(reloadedConsumeQueue.load()); + reloadedConsumeQueue.recover(); + + Assert.assertFalse(retainedMappedFilePath.exists()); + Assert.assertNull(reloadedConsumeQueue.getExt(retainedExtAddress)); + + DispatchRequest replacement = new DispatchRequest(topic, 0, 100, 10, + 300, 3000, 1, null, null, 0, 0, null); + reloadedConsumeQueue.putMessagePositionInfoWrapper(replacement); + long replacementExtAddress = getRawTagsCode(reloadedConsumeQueue, 1); + Assert.assertEquals(firstExtAddress, replacementExtAddress); + Assert.assertEquals(300, reloadedConsumeQueue.getExt(replacementExtAddress).getTagsCode()); + } finally { + if (heldBuffer != null) { + heldBuffer.release(); + } + if (reloadedConsumeQueue != null) { + reloadedConsumeQueue.destroy(); + } + consumeQueue.destroy(); + FileUtils.deleteQuietly(tmpDir); + } + } + + @Test + public void testRecoverRetainsExtAddressBeforeRecoveryWindow() throws Exception { + File tmpDir = Files.createTempDirectory("recover-old-cq-ext").toFile(); + String topic = "recoverOldExtTopic"; + MessageStoreConfig storeConfig = new MessageStoreConfig(); + storeConfig.setStorePathRootDir(tmpDir.getAbsolutePath()); + storeConfig.setMappedFileSizeConsumeQueue(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(topic, 0, tmpDir.getAbsolutePath(), + storeConfig.getMappedFileSizeConsumeQueue(), messageStore); + ConsumeQueue reloadedConsumeQueue = null; + try { + DispatchRequest first = new DispatchRequest(topic, 0, 0, 10, + 100, 1000, 0, null, null, 0, 0, null); + consumeQueue.putMessagePositionInfoWrapper(first); + long firstExtAddress = getRawTagsCode(consumeQueue, 0); + + storeConfig.setEnableConsumeQueueExt(false); + for (int i = 1; i < 4; i++) { + DispatchRequest request = new DispatchRequest(topic, 0, 100L * i, 10, + 100 + i, 1000 + i, i, null, null, 0, 0, null); + consumeQueue.putMessagePositionInfoWrapper(request); + } + consumeQueue.flush(0); + storeConfig.setEnableConsumeQueueExt(true); + + reloadedConsumeQueue = new ConsumeQueue(topic, 0, tmpDir.getAbsolutePath(), + storeConfig.getMappedFileSizeConsumeQueue(), messageStore); + Assert.assertTrue(reloadedConsumeQueue.load()); + reloadedConsumeQueue.recover(); + + ConsumeQueueExt.CqExtUnit retainedUnit = reloadedConsumeQueue.getExt(firstExtAddress); + Assert.assertNotNull(retainedUnit); + Assert.assertEquals(100, retainedUnit.getTagsCode()); + } finally { + if (reloadedConsumeQueue != null) { + reloadedConsumeQueue.destroy(); + } + consumeQueue.destroy(); + FileUtils.deleteQuietly(tmpDir); + } + } + @Test public void testTruncateKeepsExtWhenConsumeQueueTailDeletionFails() throws Exception { File tmpDir = Files.createTempDirectory("truncate-cq-tail-delete-failure").toFile(); @@ -1219,21 +1347,22 @@ public void testTruncateAllConsumeQueueExtRetriesAfterMappedFileRelease() throws heldBuffer = null; Assert.assertTrue(retainedMappedFilePath.exists()); - Assert.assertTrue(consumeQueueExt.truncateAll()); + AtomicLong replacementExtAddress = new AtomicLong(1); + Awaitility.await().atMost(5, SECONDS).pollInterval(100, TimeUnit.MILLISECONDS).until(() -> { + replacementExtAddress.set(consumeQueueExt.put( + new ConsumeQueueExt.CqExtUnit(200L, 2000L, null))); + return ConsumeQueueExt.isExtAddr(replacementExtAddress.get()); + }); 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); + Assert.assertEquals(firstExtAddress, replacementExtAddress.get()); consumeQueueExt.flush(0); reloadedConsumeQueueExt = new ConsumeQueueExt(topic, 0, extStorePath, mappedFileSize, 0); Assert.assertTrue(reloadedConsumeQueueExt.load()); reloadedConsumeQueueExt.recover(); - ConsumeQueueExt.CqExtUnit replacement = reloadedConsumeQueueExt.get(replacementExtAddress); + ConsumeQueueExt.CqExtUnit replacement = reloadedConsumeQueueExt.get(replacementExtAddress.get()); Assert.assertNotNull(replacement); Assert.assertEquals(200, replacement.getTagsCode()); } finally { @@ -1260,6 +1389,12 @@ private MappedFileQueue getMappedFileQueue(ConsumeQueueExt consumeQueueExt) thro return (MappedFileQueue) mappedFileQueueField.get(consumeQueueExt); } + private ConsumeQueueExt getConsumeQueueExt(ConsumeQueue consumeQueue) throws ReflectiveOperationException { + Field consumeQueueExtField = ConsumeQueue.class.getDeclaredField("consumeQueueExt"); + consumeQueueExtField.setAccessible(true); + return (ConsumeQueueExt) consumeQueueExtField.get(consumeQueue); + } + private long getRawTagsCode(ConsumeQueue consumeQueue, long queueOffset) { SelectMappedBufferResult result = consumeQueue.getIndexBuffer(queueOffset); Assert.assertNotNull(result);