opensource/rocketmq
3
0
Fork 0
mirror of https://github.com/apache/rocketmq.git synced 2026-08-28 20:09:14 +08:00
Code Issues Packages Projects Releases Wiki Activity

[ISSUE #10417] Fix stale minOffset after consumeQueue truncation (#10418)

Browse Source
This commit is contained in:
chenxu80
2026-06-23 17:10:16 +08:00
committed by GitHub
parent b5bc1ff5d6
commit 2e6632ff40
2 changed files with 71 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files
store/src/main/java/org/apache/rocketmq/store/ConsumeQueue.java
+9 -8
View File
@@ -552,17 +552,18 @@ public class ConsumeQueue implements ConsumeQueueInterface {
*/
@Override
public void correctMinOffset(long minCommitLogOffset) {
// Check if the consume queue is the state of deprecation.
if (minLogicOffset >= mappedFileQueue.getMaxOffset()) {
log.info("ConsumeQueue[Topic={}, queue-id={}] contains no valid entries", topic, queueId);
MappedFile lastMappedFile = this.mappedFileQueue.getLastMappedFile();
if (null == lastMappedFile) {
log.info("ConsumeQueue[Topic={}, queue-id={}] contains no entry,"
+ " reset minLogicOffset to 0, maxPhysicOffset to -1", topic, queueId);
this.minLogicOffset = 0;
this.maxPhysicOffset = -1;
return;
}
// Check whether the consume queue maps no valid data at all. This check may cost 1 IO operation.
// The rationale is that consume queue always preserves the last file. In case there are many deprecated topics,
// This check would save a lot of efforts.
MappedFile lastMappedFile = this.mappedFileQueue.getLastMappedFile();
if (null == lastMappedFile) {
// Check if the consume queue is the state of deprecation.
if (minLogicOffset > mappedFileQueue.getMaxOffset()) {
log.warn("ConsumeQueue[Topic={}, queue-id={}] contains no valid entries", topic, queueId);
return;
}
store/src/test/java/org/apache/rocketmq/store/ConsumeQueueTest.java
+62
View File
@@ -24,6 +24,7 @@ import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.net.SocketAddress;
import java.net.UnknownHostException;
import java.nio.file.Files;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
@@ -713,4 +714,65 @@ public class ConsumeQueueTest {
consumeQueue.destroy();
FileUtils.deleteDirectory(tmpDir);
}
@Test
public void testCorrectMinOffsetAfterAllFilesDeleted() throws IOException {
String topic = "T1";
int queueId = 0;
MessageStoreConfig storeConfig = new MessageStoreConfig();
File tmpDir = Files.createTempDirectory("testCorrectMinOffsetAfterAllFilesDeleted").toFile();
storeConfig.setStorePathRootDir(tmpDir.getAbsolutePath());
storeConfig.setEnableConsumeQueueExt(false);
DefaultMessageStore messageStore = Mockito.mock(DefaultMessageStore.class);
Mockito.when(messageStore.getMessageStoreConfig()).thenReturn(storeConfig);
RunningFlags runningFlags = new RunningFlags();
Mockito.when(messageStore.getRunningFlags()).thenReturn(runningFlags);
StoreCheckpoint storeCheckpoint = Mockito.mock(StoreCheckpoint.class);
Mockito.when(messageStore.getStoreCheckpoint()).thenReturn(storeCheckpoint);
ConsumeQueue consumeQueue = null;
ConsumeQueue reloadedConsumeQueue = null;
try {
consumeQueue = new ConsumeQueue(topic, queueId, storeConfig.getStorePathRootDir(),
storeConfig.getMappedFileSizeConsumeQueue(), messageStore);
int max = 10;
int messageSize = 100;
for (int i = 0; i < max; ++i) {
DispatchRequest dispatchRequest = new DispatchRequest(topic, queueId, messageSize * i, messageSize, 0, 0, i,
null, null, 0, 0, null);
consumeQueue.putMessagePositionInfoWrapper(dispatchRequest);
}
File consumeQueueDir = new File(storeConfig.getStorePathRootDir(), topic + File.separator + queueId);
Assert.assertTrue(consumeQueueDir.exists());
consumeQueue.setMinLogicOffset(ConsumeQueue.CQ_STORE_UNIT_SIZE);
consumeQueue.setMaxPhysicOffset(max * messageSize);
consumeQueue.destroy();
consumeQueue = null;
FileUtils.deleteQuietly(consumeQueueDir);
Assert.assertFalse(consumeQueueDir.exists());
reloadedConsumeQueue = new ConsumeQueue(topic, queueId, storeConfig.getStorePathRootDir(),
storeConfig.getMappedFileSizeConsumeQueue(), messageStore);
Assert.assertTrue(reloadedConsumeQueue.load());
reloadedConsumeQueue.recover();
reloadedConsumeQueue.setMinLogicOffset(ConsumeQueue.CQ_STORE_UNIT_SIZE);
reloadedConsumeQueue.setMaxPhysicOffset(max * messageSize);
reloadedConsumeQueue.correctMinOffset(0L);
Assert.assertEquals(0L, reloadedConsumeQueue.getMinLogicOffset());
Assert.assertEquals(-1L, reloadedConsumeQueue.getMaxPhysicOffset());
} finally {
if (reloadedConsumeQueue != null) {
reloadedConsumeQueue.destroy();
}
if (consumeQueue != null) {
consumeQueue.destroy();
}
FileUtils.deleteQuietly(tmpDir);
}
}
}