You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@rocketmq.apache.org by li...@apache.org on 2025/04/29 02:31:54 UTC
(rocketmq) branch develop updated: [ISSUE #9371] Delete ConsumeQueue index before CommitLog in tiered storage (#9372)
This is an automated email from the ASF dual-hosted git repository.
lizhimin pushed a commit to branch develop
in repository https://gitbox.apache.org/repos/asf/rocketmq.git
The following commit(s) were added to refs/heads/develop by this push:
new 8c623df7ad [ISSUE #9371] Delete ConsumeQueue index before CommitLog in tiered storage (#9372)
8c623df7ad is described below
commit 8c623df7ad85b8446066f322b73cc33d0d9d3f4c
Author: lizhimins <70...@qq.com>
AuthorDate: Tue Apr 29 10:31:49 2025 +0800
[ISSUE #9371] Delete ConsumeQueue index before CommitLog in tiered storage (#9372)
---
.../org/apache/rocketmq/tieredstore/file/FlatAppendFile.java | 6 +++---
.../org/apache/rocketmq/tieredstore/file/FlatCommitLogFile.java | 2 +-
.../org/apache/rocketmq/tieredstore/file/FlatMessageFile.java | 9 +++------
3 files changed, 7 insertions(+), 10 deletions(-)
diff --git a/tieredstore/src/main/java/org/apache/rocketmq/tieredstore/file/FlatAppendFile.java b/tieredstore/src/main/java/org/apache/rocketmq/tieredstore/file/FlatAppendFile.java
index 891170d703..377341d950 100644
--- a/tieredstore/src/main/java/org/apache/rocketmq/tieredstore/file/FlatAppendFile.java
+++ b/tieredstore/src/main/java/org/apache/rocketmq/tieredstore/file/FlatAppendFile.java
@@ -124,17 +124,17 @@ public class FlatAppendFile {
public long getMinOffset() {
List<FileSegment> list = this.fileSegmentTable;
- return list.isEmpty() ? GET_FILE_SIZE_ERROR : list.get(0).getBaseOffset();
+ return list.isEmpty() ? 0L : list.get(0).getBaseOffset();
}
public long getCommitOffset() {
List<FileSegment> list = this.fileSegmentTable;
- return list.isEmpty() ? GET_FILE_SIZE_ERROR : list.get(list.size() - 1).getCommitOffset();
+ return list.isEmpty() ? 0L : list.get(list.size() - 1).getCommitOffset();
}
public long getAppendOffset() {
List<FileSegment> list = this.fileSegmentTable;
- return list.isEmpty() ? GET_FILE_SIZE_ERROR : list.get(list.size() - 1).getAppendOffset();
+ return list.isEmpty() ? 0L : list.get(list.size() - 1).getAppendOffset();
}
public long getMinTimestamp() {
diff --git a/tieredstore/src/main/java/org/apache/rocketmq/tieredstore/file/FlatCommitLogFile.java b/tieredstore/src/main/java/org/apache/rocketmq/tieredstore/file/FlatCommitLogFile.java
index 16c0520475..bdf0bf375e 100644
--- a/tieredstore/src/main/java/org/apache/rocketmq/tieredstore/file/FlatCommitLogFile.java
+++ b/tieredstore/src/main/java/org/apache/rocketmq/tieredstore/file/FlatCommitLogFile.java
@@ -77,7 +77,7 @@ public class FlatCommitLogFile extends FlatAppendFile {
super.destroyExpiredFile(expireTimestamp);
long afterOffset = this.getMinOffset();
- if (beforeOffset != afterOffset) {
+ if (beforeOffset != afterOffset && afterOffset > 0) {
log.info("CommitLog min cq offset reset, filePath={}, offset={}, expireTimestamp={}, change={}-{}",
filePath, firstOffset.get(), expireTimestamp, beforeOffset, afterOffset);
firstOffset.set(GET_OFFSET_ERROR);
diff --git a/tieredstore/src/main/java/org/apache/rocketmq/tieredstore/file/FlatMessageFile.java b/tieredstore/src/main/java/org/apache/rocketmq/tieredstore/file/FlatMessageFile.java
index 2519f91ebe..b0e4dd6e3b 100644
--- a/tieredstore/src/main/java/org/apache/rocketmq/tieredstore/file/FlatMessageFile.java
+++ b/tieredstore/src/main/java/org/apache/rocketmq/tieredstore/file/FlatMessageFile.java
@@ -25,7 +25,6 @@ import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.Semaphore;
-import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
import org.apache.commons.lang3.StringUtils;
@@ -59,7 +58,6 @@ public class FlatMessageFile implements FlatFileInterface {
protected final MetadataStore metadataStore;
protected final FlatCommitLogFile commitLog;
protected final FlatConsumeQueueFile consumeQueue;
- protected final AtomicLong lastDestroyTime;
protected final ConcurrentMap<String, CompletableFuture<?>> inFlightRequestMap;
@@ -77,7 +75,6 @@ public class FlatMessageFile implements FlatFileInterface {
this.metadataStore = fileFactory.getMetadataStore();
this.commitLog = fileFactory.createFlatFileForCommitLog(filePath);
this.consumeQueue = fileFactory.createFlatFileForConsumeQueue(filePath);
- this.lastDestroyTime = new AtomicLong();
this.inFlightRequestMap = new ConcurrentHashMap<>();
}
@@ -388,8 +385,8 @@ public class FlatMessageFile implements FlatFileInterface {
closed = true;
fileLock.lock();
try {
- commitLog.shutdown();
consumeQueue.shutdown();
+ commitLog.shutdown();
} finally {
fileLock.unlock();
}
@@ -399,8 +396,8 @@ public class FlatMessageFile implements FlatFileInterface {
public void destroyExpiredFile(long timestamp) {
fileLock.lock();
try {
- commitLog.destroyExpiredFile(timestamp);
consumeQueue.destroyExpiredFile(timestamp);
+ commitLog.destroyExpiredFile(timestamp);
} finally {
fileLock.unlock();
}
@@ -410,8 +407,8 @@ public class FlatMessageFile implements FlatFileInterface {
this.shutdown();
fileLock.lock();
try {
- commitLog.destroyExpiredFile(Long.MAX_VALUE);
consumeQueue.destroyExpiredFile(Long.MAX_VALUE);
+ commitLog.destroyExpiredFile(Long.MAX_VALUE);
if (queueMetadata != null) {
metadataStore.deleteQueue(queueMetadata.getQueue());
}