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 2023/08/22 08:16:32 UTC
[rocketmq] branch fix-index-file created (now 9127edca63)
This is an automated email from the ASF dual-hosted git repository.
lizhimin pushed a change to branch fix-index-file
in repository https://gitbox.apache.org/repos/asf/rocketmq.git
at 9127edca63 Temporary fix index file bug in tiered storage
This branch includes the following new commits:
new 9127edca63 Temporary fix index file bug in tiered storage
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
[rocketmq] 01/01: Temporary fix index file bug in tiered storage
Posted by li...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lizhimin pushed a commit to branch fix-index-file
in repository https://gitbox.apache.org/repos/asf/rocketmq.git
commit 9127edca6368d97a38b976c87f7544b4b7aca250
Author: lizhimins <70...@qq.com>
AuthorDate: Tue Aug 22 16:03:29 2023 +0800
Temporary fix index file bug in tiered storage
---
.../rocketmq/tieredstore/file/TieredIndexFile.java | 42 +++++++++--
.../tieredstore/file/TieredIndexFileTest.java | 88 +++++-----------------
2 files changed, 56 insertions(+), 74 deletions(-)
diff --git a/tieredstore/src/main/java/org/apache/rocketmq/tieredstore/file/TieredIndexFile.java b/tieredstore/src/main/java/org/apache/rocketmq/tieredstore/file/TieredIndexFile.java
index 50beb01ae4..cdb7d66e71 100644
--- a/tieredstore/src/main/java/org/apache/rocketmq/tieredstore/file/TieredIndexFile.java
+++ b/tieredstore/src/main/java/org/apache/rocketmq/tieredstore/file/TieredIndexFile.java
@@ -16,6 +16,7 @@
*/
package org.apache.rocketmq.tieredstore.file;
+import com.google.common.annotations.VisibleForTesting;
import java.io.File;
import java.io.IOException;
import java.nio.ByteBuffer;
@@ -99,7 +100,7 @@ public class TieredIndexFile {
this::doScheduleTask, 10, 10, TimeUnit.SECONDS);
}
- private void doScheduleTask() {
+ protected void doScheduleTask() {
try {
curFileLock.lock();
try {
@@ -145,6 +146,11 @@ public class TieredIndexFile {
}
}
+ @VisibleForTesting
+ public MappedFile getPreMappedFile() {
+ return preMappedFile;
+ }
+
private void initFile() throws IOException {
curMappedFile = new DefaultMappedFile(curFilePath, fileMaxSize);
initIndexFileHeader(curMappedFile);
@@ -156,19 +162,26 @@ public class TieredIndexFile {
if (isFileSealed(curMappedFile)) {
if (preFileExists) {
- preFile.delete();
+ if (preFile.delete()) {
+ logger.info("Pre IndexFile deleted success", preFilepath);
+ } else {
+ logger.error("Pre IndexFile deleted failed", preFilepath);
+ }
}
boolean rename = curMappedFile.renameTo(preFilepath);
if (rename) {
preMappedFile = curMappedFile;
curMappedFile = new DefaultMappedFile(curFilePath, fileMaxSize);
+ initIndexFileHeader(curMappedFile);
preFileExists = true;
}
}
+
if (preFileExists) {
synchronized (TieredIndexFile.class) {
if (inflightCompactFuture.isDone()) {
- inflightCompactFuture = TieredStoreExecutor.compactIndexFileExecutor.submit(new CompactTask(storeConfig, preMappedFile, flatFile), null);
+ inflightCompactFuture = TieredStoreExecutor.compactIndexFileExecutor.submit(
+ new CompactTask(storeConfig, preMappedFile, flatFile), null);
}
}
}
@@ -261,7 +274,8 @@ public class TieredIndexFile {
}
}
- public CompletableFuture<List<Pair<Long, ByteBuffer>>> queryAsync(String topic, String key, long beginTime, long endTime) {
+ public CompletableFuture<List<Pair<Long, ByteBuffer>>> queryAsync(String topic, String key, long beginTime,
+ long endTime) {
int hashCode = indexKeyHashMethod(buildKey(topic, key));
int slotPosition = hashCode % maxHashSlotNum;
List<TieredFileSegment> fileSegmentList = flatFile.getFileListByTime(beginTime, endTime);
@@ -355,7 +369,7 @@ public class TieredIndexFile {
private final int fileMaxSize;
private MappedFile originFile;
private TieredFlatFile fileQueue;
- private final MappedFile compactFile;
+ private MappedFile compactFile;
public CompactTask(TieredMessageStoreConfig storeConfig, MappedFile originFile,
TieredFlatFile fileQueue) throws IOException {
@@ -381,6 +395,17 @@ public class TieredIndexFile {
} catch (Throwable throwable) {
logger.error("TieredIndexFile#compactTask: compact index file failed:", throwable);
}
+
+ try {
+ if (originFile != null) {
+ originFile.destroy(-1);
+ }
+ if (compactFile != null) {
+ compactFile.destroy(-1);
+ }
+ } catch (Throwable throwable) {
+ logger.error("TieredIndexFile#compactTask: destroy index file failed:", throwable);
+ }
}
public void compact() {
@@ -396,6 +421,8 @@ public class TieredIndexFile {
fileQueue.commit(true);
compactFile.destroy(-1);
originFile.destroy(-1);
+ compactFile = null;
+ originFile = null;
}
private void buildCompactFile() {
@@ -415,6 +442,11 @@ public class TieredIndexFile {
int indexTotalSize = 0;
int indexPosition = slotValue;
while (indexPosition >= 0 && indexPosition < maxIndexNum) {
+ if (compactMappedByteBuffer.remaining() < INDEX_FILE_HASH_ORIGIN_INDEX_SIZE + 4) {
+ logger.error("IndexFile compaction error, byteBuffer remaining not enough, left bytes={}",
+ compactMappedByteBuffer.remaining());
+ break;
+ }
int indexOffset = INDEX_FILE_HEADER_SIZE + maxHashSlotNum * INDEX_FILE_HASH_SLOT_SIZE
+ indexPosition * INDEX_FILE_HASH_ORIGIN_INDEX_SIZE;
int rePutIndexOffset = rePutSlotValue + indexTotalSize;
diff --git a/tieredstore/src/test/java/org/apache/rocketmq/tieredstore/file/TieredIndexFileTest.java b/tieredstore/src/test/java/org/apache/rocketmq/tieredstore/file/TieredIndexFileTest.java
index 7ef49578dd..33dcc30f37 100644
--- a/tieredstore/src/test/java/org/apache/rocketmq/tieredstore/file/TieredIndexFileTest.java
+++ b/tieredstore/src/test/java/org/apache/rocketmq/tieredstore/file/TieredIndexFileTest.java
@@ -16,12 +16,10 @@
*/
package org.apache.rocketmq.tieredstore.file;
-import com.sun.jna.Platform;
import java.io.IOException;
import java.nio.ByteBuffer;
+import java.time.Duration;
import java.util.List;
-import java.util.concurrent.TimeUnit;
-import org.apache.commons.lang3.SystemUtils;
import org.apache.commons.lang3.tuple.Pair;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.tieredstore.TieredStoreTestUtil;
@@ -31,9 +29,7 @@ import org.apache.rocketmq.tieredstore.util.TieredStoreUtil;
import org.awaitility.Awaitility;
import org.junit.After;
import org.junit.Assert;
-import org.junit.Assume;
import org.junit.Before;
-import org.junit.Ignore;
import org.junit.Test;
public class TieredIndexFileTest {
@@ -45,11 +41,12 @@ public class TieredIndexFileTest {
@Before
public void setUp() {
storeConfig = new TieredMessageStoreConfig();
+ storeConfig.setBrokerName("IndexFileBroker");
storeConfig.setStorePathRootDir(storePath);
- storeConfig.setTieredBackendServiceProvider("org.apache.rocketmq.tieredstore.provider.memory.MemoryFileSegment");
- storeConfig.setTieredStoreIndexFileMaxHashSlotNum(2);
- storeConfig.setTieredStoreIndexFileMaxIndexNum(3);
- mq = new MessageQueue("TieredIndexFileTest", storeConfig.getBrokerName(), 1);
+ storeConfig.setTieredBackendServiceProvider("org.apache.rocketmq.tieredstore.provider.posix.PosixFileSegment");
+ storeConfig.setTieredStoreIndexFileMaxHashSlotNum(5);
+ storeConfig.setTieredStoreIndexFileMaxIndexNum(20);
+ mq = new MessageQueue("IndexFileTest", storeConfig.getBrokerName(), 1);
TieredStoreUtil.getMetadataStore(storeConfig);
TieredStoreExecutor.init();
}
@@ -61,77 +58,30 @@ public class TieredIndexFileTest {
TieredStoreExecutor.shutdown();
}
- @Ignore
@Test
public void testAppendAndQuery() throws IOException, ClassNotFoundException, NoSuchMethodException {
- if (Platform.isWindows()) {
- return;
- }
-
- // skip this test on windows
- Assume.assumeFalse(SystemUtils.IS_OS_WINDOWS);
TieredFileAllocator fileQueueFactory = new TieredFileAllocator(storeConfig);
TieredIndexFile indexFile = new TieredIndexFile(fileQueueFactory, storePath);
+
indexFile.append(mq, 0, "key3", 3, 300, 1000);
indexFile.append(mq, 0, "key2", 2, 200, 1100);
indexFile.append(mq, 0, "key1", 1, 100, 1200);
- Awaitility.waitAtMost(5, TimeUnit.SECONDS)
- .until(() -> {
- List<Pair<Long, ByteBuffer>> indexList = indexFile.queryAsync(mq.getTopic(), "key1", 1000, 1200).join();
- if (indexList.size() != 1) {
- return false;
- }
-
- ByteBuffer indexBuffer = indexList.get(0).getValue();
- Assert.assertEquals(TieredIndexFile.INDEX_FILE_HASH_COMPACT_INDEX_SIZE * 2, indexBuffer.remaining());
-
- Assert.assertEquals(1, indexBuffer.getLong(4 + 4 + 4));
- Assert.assertEquals(100, indexBuffer.getInt(4 + 4 + 4 + 8));
- Assert.assertEquals(200, indexBuffer.getInt(4 + 4 + 4 + 8 + 4));
-
- Assert.assertEquals(3, indexBuffer.getLong(TieredIndexFile.INDEX_FILE_HASH_COMPACT_INDEX_SIZE + 4 + 4 + 4));
- Assert.assertEquals(300, indexBuffer.getInt(TieredIndexFile.INDEX_FILE_HASH_COMPACT_INDEX_SIZE + 4 + 4 + 4 + 8));
- Assert.assertEquals(0, indexBuffer.getInt(TieredIndexFile.INDEX_FILE_HASH_COMPACT_INDEX_SIZE + 4 + 4 + 4 + 8 + 4));
- return true;
- });
-
- indexFile.append(mq, 0, "key4", 4, 400, 1300);
- indexFile.append(mq, 0, "key4", 4, 400, 1300);
- indexFile.append(mq, 0, "key4", 4, 400, 1300);
-
- Awaitility.waitAtMost(5, TimeUnit.SECONDS)
- .until(() -> {
- List<Pair<Long, ByteBuffer>> indexList = indexFile.queryAsync(mq.getTopic(), "key4", 1300, 1300).join();
- if (indexList.size() != 1) {
- return false;
- }
-
- ByteBuffer indexBuffer = indexList.get(0).getValue();
- Assert.assertEquals(TieredIndexFile.INDEX_FILE_HASH_COMPACT_INDEX_SIZE * 3, indexBuffer.remaining());
- Assert.assertEquals(4, indexBuffer.getLong(4 + 4 + 4));
- Assert.assertEquals(400, indexBuffer.getInt(4 + 4 + 4 + 8));
- Assert.assertEquals(0, indexBuffer.getInt(4 + 4 + 4 + 8 + 4));
- return true;
- });
-
- List<Pair<Long, ByteBuffer>> indexList = indexFile.queryAsync(mq.getTopic(), "key1", 1300, 1300).join();
+ // do not do schedule task here
+ TieredStoreExecutor.shutdown();
+ List<Pair<Long, ByteBuffer>> indexList =
+ indexFile.queryAsync(mq.getTopic(), "key1", 1000, 1200).join();
Assert.assertEquals(0, indexList.size());
- indexList = indexFile.queryAsync(mq.getTopic(), "key4", 1200, 1300).join();
- Assert.assertEquals(2, indexList.size());
-
- ByteBuffer indexBuffer = indexList.get(0).getValue();
- Assert.assertEquals(TieredIndexFile.INDEX_FILE_HASH_COMPACT_INDEX_SIZE * 3, indexBuffer.remaining());
- Assert.assertEquals(4, indexBuffer.getLong(4 + 4 + 4));
- Assert.assertEquals(400, indexBuffer.getInt(4 + 4 + 4 + 8));
- Assert.assertEquals(0, indexBuffer.getInt(4 + 4 + 4 + 8 + 4));
+ // do compaction once
+ TieredStoreExecutor.init();
+ storeConfig.setTieredStoreIndexFileRollingIdleInterval(0);
+ indexFile.doScheduleTask();
+ Awaitility.await().atMost(Duration.ofSeconds(10))
+ .until(() -> !indexFile.getPreMappedFile().getFile().exists());
- indexBuffer = indexList.get(1).getValue();
- Assert.assertEquals(TieredIndexFile.INDEX_FILE_HASH_COMPACT_INDEX_SIZE, indexBuffer.remaining());
- Assert.assertEquals(2, indexBuffer.getLong(4 + 4 + 4));
- Assert.assertEquals(200, indexBuffer.getInt(4 + 4 + 4 + 8));
- Assert.assertEquals(100, indexBuffer.getInt(4 + 4 + 4 + 8 + 4));
+ indexList = indexFile.queryAsync(mq.getTopic(), "key1", 1000, 1200).join();
+ Assert.assertEquals(1, indexList.size());
}
}