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

[ISSUE #9112] Speedup revive scan in Pop Consumption and support server side reset offset (#9113)

Browse Source
This commit is contained in:
lizhimins
2025-01-08 23:00:22 +08:00
committed by GitHub
parent a343021229
commit e0db6545b0
6 changed files with 162 additions and 45 deletions
Download Patch File Download Diff File Expand all files Collapse all files
broker/src/main/java/org/apache/rocketmq/broker/pop/PopConsumerKVStore.java
+7 -4
View File
@@ -49,10 +49,13 @@ public interface PopConsumerKVStore {
void deleteRecords(List<PopConsumerRecord> consumerRecordList);
/**
* Scans and returns a list of expired consumer records before the current time.
* @param currentTime The current revive checkpoint timestamp.
* Scans and returns a list of expired consumer records within the specified time range.
* @param lowerTime The start time (inclusive) of the time range to search, in milliseconds.
* @param upperTime The end time (exclusive) of the time range to search, in milliseconds.
* @param maxCount The maximum number of records to return.
* @return A list of expired consumer records.
* Even if more records match the criteria, only this many will be returned.
* @return A list of expired consumer records within the specified time range.
* If no matching records are found, an empty list is returned.
*/
List<PopConsumerRecord> scanExpiredRecords(long currentTime, int maxCount);
List<PopConsumerRecord> scanExpiredRecords(long lowerTime, long upperTime, int maxCount);
}
broker/src/main/java/org/apache/rocketmq/broker/pop/PopConsumerRocksdbStore.java
+11 -9
View File
@@ -28,9 +28,11 @@ import org.rocksdb.ColumnFamilyDescriptor;
import org.rocksdb.ColumnFamilyHandle;
import org.rocksdb.ColumnFamilyOptions;
import org.rocksdb.CompactRangeOptions;
import org.rocksdb.ReadOptions;
import org.rocksdb.RocksDB;
import org.rocksdb.RocksDBException;
import org.rocksdb.RocksIterator;
import org.rocksdb.Slice;
import org.rocksdb.WriteBatch;
import org.rocksdb.WriteOptions;
import org.slf4j.Logger;
@@ -43,7 +45,7 @@ public class PopConsumerRocksdbStore extends AbstractRocksDBStorage implements P
private WriteOptions writeOptions;
private WriteOptions deleteOptions;
private ColumnFamilyHandle columnFamilyHandle;
protected ColumnFamilyHandle columnFamilyHandle;
public PopConsumerRocksdbStore(String filePath) {
super(filePath);
@@ -60,8 +62,7 @@ public class PopConsumerRocksdbStore extends AbstractRocksDBStorage implements P
this.writeOptions.setNoSlowdown(false);
this.deleteOptions = new WriteOptions();
this.deleteOptions.setSync(false);
this.deleteOptions.setLowPri(true);
this.deleteOptions.setSync(true);
this.deleteOptions.setDisableWAL(false);
this.deleteOptions.setNoSlowdown(false);
@@ -135,18 +136,19 @@ public class PopConsumerRocksdbStore extends AbstractRocksDBStorage implements P
}
@Override
public List<PopConsumerRecord> scanExpiredRecords(long currentTime, int maxCount) {
// https://github.com/facebook/rocksdb/issues/10300
public List<PopConsumerRecord> scanExpiredRecords(long lower, long upper, int maxCount) {
// In RocksDB, we can use SstPartitionerFixedPrefixFactory in cfOptions
// and new ColumnFamilyOptions().useFixedLengthPrefixExtractor() to
// configure prefix indexing to improve the performance of scans.
// However, in the current implementation, this is not the bottleneck.
List<PopConsumerRecord> consumerRecordList = new ArrayList<>();
try (RocksIterator iterator = db.newIterator(this.columnFamilyHandle)) {
iterator.seekToFirst();
try (ReadOptions scanOptions = new ReadOptions()
.setIterateLowerBound(new Slice(ByteBuffer.allocate(Long.BYTES).putLong(lower).array()))
.setIterateUpperBound(new Slice(ByteBuffer.allocate(Long.BYTES).putLong(upper).array()));
RocksIterator iterator = db.newIterator(this.columnFamilyHandle, scanOptions)) {
iterator.seek(ByteBuffer.allocate(Long.BYTES).putLong(lower).array());
while (iterator.isValid() && consumerRecordList.size() < maxCount) {
if (ByteBuffer.wrap(iterator.key()).getLong() > currentTime) {
break;
}
consumerRecordList.add(PopConsumerRecord.decode(iterator.value()));
iterator.next();
}
broker/src/main/java/org/apache/rocketmq/broker/pop/PopConsumerService.java
+40 -20
View File
@@ -75,6 +75,7 @@ public class PopConsumerService extends ServiceThread {
private final AtomicBoolean consumerRunning;
private final BrokerConfig brokerConfig;
private final BrokerController brokerController;
private final AtomicLong currentTime;
private final AtomicLong lastCleanupLockTime;
private final PopConsumerCache popConsumerCache;
private final PopConsumerKVStore popConsumerStore;
@@ -88,6 +89,7 @@ public class PopConsumerService extends ServiceThread {
this.consumerRunning = new AtomicBoolean(false);
this.requestCountTable = new ConcurrentHashMap<>();
this.currentTime = new AtomicLong(TimeUnit.SECONDS.toMillis(3));
this.lastCleanupLockTime = new AtomicLong(System.currentTimeMillis());
this.consumerLockService = new PopConsumerLockService(TimeUnit.MINUTES.toMillis(2));
this.popConsumerStore = new PopConsumerRocksdbStore(Paths.get(
@@ -195,12 +197,27 @@ public class PopConsumerService extends ServiceThread {
return context;
}
public Long getPopOffset(String groupId, String topicId, int queueId) {
Long resetOffset =
this.brokerController.getConsumerOffsetManager().queryThenEraseResetOffset(topicId, groupId, queueId);
if (resetOffset != null) {
this.clearCache(groupId, topicId, queueId);
this.brokerController.getConsumerOrderInfoManager().clearBlock(topicId, groupId, queueId);
this.brokerController.getConsumerOffsetManager()
.commitOffset("ResetPopOffset", groupId, topicId, queueId, resetOffset);
}
return resetOffset;
}
public CompletableFuture<GetMessageResult> getMessageAsync(String clientHost,
String groupId, String topicId, int queueId, long offset, int batchSize, MessageFilter filter) {
log.debug("PopConsumerService getMessageAsync, groupId={}, topicId={}, queueId={}, offset={}, batchSize={}, filter={}",
groupId, topicId, offset, queueId, batchSize, filter != null);
Long resetOffset = this.getPopOffset(groupId, topicId, queueId);
final long currentOffset = resetOffset != null ? resetOffset : offset;
CompletableFuture<GetMessageResult> getMessageFuture =
brokerController.getMessageStore().getMessageAsync(groupId, topicId, queueId, offset, batchSize, filter);
@@ -223,7 +240,7 @@ public class PopConsumerService extends ServiceThread {
log.warn("PopConsumerService getMessageAsync, initial offset because store is no correct, " +
"groupId={}, topicId={}, queueId={}, batchSize={}, offset={}->{}",
groupId, topicId, queueId, batchSize, offset, result.getNextBeginOffset());
groupId, topicId, queueId, batchSize, currentOffset, result.getNextBeginOffset());
return brokerController.getMessageStore().getMessageAsync(
groupId, topicId, queueId, result.getNextBeginOffset(), batchSize, filter);
@@ -482,10 +499,12 @@ public class PopConsumerService extends ServiceThread {
}
}
public long revive(long currentTime, int maxCount) {
public long revive(AtomicLong currentTime, int maxCount) {
Stopwatch stopwatch = Stopwatch.createStarted();
List<PopConsumerRecord> consumerRecords =
this.popConsumerStore.scanExpiredRecords(currentTime, maxCount);
long upperTime = System.currentTimeMillis() - 50L;
List<PopConsumerRecord> consumerRecords = this.popConsumerStore.scanExpiredRecords(
currentTime.get() - TimeUnit.SECONDS.toMillis(3), upperTime, maxCount);
long scanCostTime = stopwatch.elapsed(TimeUnit.MILLISECONDS);
Queue<PopConsumerRecord> failureList = new LinkedBlockingQueue<>();
List<CompletableFuture<?>> futureList = new ArrayList<>(consumerRecords.size());
@@ -497,9 +516,9 @@ public class PopConsumerService extends ServiceThread {
long backoffInterval = 1000L * REWRITE_INTERVALS_IN_SECONDS[
Math.min(REWRITE_INTERVALS_IN_SECONDS.length, record.getAttemptTimes())];
long nextInvisibleTime = record.getInvisibleTime() + backoffInterval;
PopConsumerRecord retryRecord = new PopConsumerRecord(record.getPopTime(), record.getGroupId(),
record.getTopicId(), record.getQueueId(), record.getRetryFlag(), nextInvisibleTime,
record.getOffset(), record.getAttemptId());
PopConsumerRecord retryRecord = new PopConsumerRecord(System.currentTimeMillis(),
record.getGroupId(), record.getTopicId(), record.getQueueId(),
record.getRetryFlag(), nextInvisibleTime, record.getOffset(), record.getAttemptId());
retryRecord.setAttemptTimes(record.getAttemptTimes() + 1);
failureList.add(retryRecord);
log.warn("PopConsumerService revive backoff retry, record={}", retryRecord);
@@ -513,14 +532,20 @@ public class PopConsumerService extends ServiceThread {
CompletableFuture.allOf(futureList.toArray(new CompletableFuture[0])).join();
this.popConsumerStore.writeRecords(new ArrayList<>(failureList));
this.popConsumerStore.deleteRecords(consumerRecords);
currentTime.set(consumerRecords.isEmpty() ?
upperTime : consumerRecords.get(consumerRecords.size() - 1).getVisibilityTimeout());
if (brokerConfig.isEnablePopBufferMerge()) {
log.info("PopConsumerService, key size={}, cache size={}, revive count={}, failure count={}, cost={}ms",
popConsumerCache.getCacheKeySize(), popConsumerCache.getCacheSize(), consumerRecords.size(),
failureList.size(), stopwatch.elapsed(TimeUnit.MILLISECONDS));
log.info("PopConsumerService, key size={}, cache size={}, revive count={}, failure count={}, " +
"behindInMillis={}, scanInMillis={}, costInMillis={}",
popConsumerCache.getCacheKeySize(), popConsumerCache.getCacheSize(),
consumerRecords.size(), failureList.size(), upperTime - currentTime.get(),
scanCostTime, stopwatch.elapsed(TimeUnit.MILLISECONDS));
} else {
log.info("PopConsumerService, revive count={}, failure count={}, cost={}ms",
consumerRecords.size(), failureList.size(), stopwatch.elapsed(TimeUnit.MILLISECONDS));
log.info("PopConsumerService, revive count={}, failure count={}, " +
"behindInMillis={}, scanInMillis={}, costInMillis={}",
consumerRecords.size(), failureList.size(), upperTime - currentTime.get(),
scanCostTime, stopwatch.elapsed(TimeUnit.MILLISECONDS));
}
return consumerRecords.size();
@@ -588,11 +613,6 @@ public class PopConsumerService extends ServiceThread {
PutMessageResult putMessageResult =
brokerController.getEscapeBridge().putMessageToSpecificQueue(msgInner);
if (brokerConfig.isEnablePopLog()) {
log.debug("PopConsumerService revive retry msg, put status={}, ck={}, delay={}ms",
putMessageResult, JSON.toJSONString(record), System.currentTimeMillis() - record.getVisibilityTimeout());
}
if (putMessageResult.getAppendMessageResult() == null ||
putMessageResult.getAppendMessageResult().getStatus() != AppendMessageStatus.PUT_OK) {
log.error("PopConsumerService revive retry msg error, put status={}, ck={}, delay={}ms",
@@ -616,7 +636,7 @@ public class PopConsumerService extends ServiceThread {
while (true) {
try {
List<PopConsumerRecord> consumerRecords = this.popConsumerStore.scanExpiredRecords(
Long.MAX_VALUE, brokerConfig.getPopReviveMaxReturnSizePerRead());
0, Long.MAX_VALUE, brokerConfig.getPopReviveMaxReturnSizePerRead());
if (consumerRecords == null || consumerRecords.isEmpty()) {
break;
}
@@ -695,7 +715,7 @@ public class PopConsumerService extends ServiceThread {
while (!isStopped()) {
try {
// to prevent concurrency issues during read and write operations
long reviveCount = this.revive(System.currentTimeMillis() - 50L,
long reviveCount = this.revive(this.currentTime,
brokerConfig.getPopReviveMaxReturnSizePerRead());
long current = System.currentTimeMillis();
@@ -704,7 +724,7 @@ public class PopConsumerService extends ServiceThread {
this.lastCleanupLockTime.set(current);
}
if (reviveCount == 0) {
if (reviveCount < brokerConfig.getPopReviveMaxReturnSizePerRead()) {
this.waitForRunning(500);
}
} catch (Exception e) {
broker/src/test/java/org/apache/rocketmq/broker/pop/PopConsumerRocksdbStoreTest.java
+94 -3
View File
@@ -16,18 +16,26 @@
*/
package org.apache.rocketmq.broker.pop;
import com.google.common.base.Stopwatch;
import java.io.File;
import java.io.IOException;
import java.lang.reflect.Field;
import java.nio.file.Paths;
import java.util.ArrayList;
import java.util.List;
import java.util.UUID;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;
import java.util.stream.IntStream;
import org.apache.commons.io.FileUtils;
import org.apache.rocketmq.common.config.AbstractRocksDBStorage;
import org.apache.rocketmq.common.constant.LoggerName;
import org.apache.rocketmq.tieredstore.util.MessageStoreUtil;
import org.junit.Assert;
import org.junit.Ignore;
import org.junit.Test;
import org.rocksdb.RocksDB;
import org.rocksdb.RocksIterator;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -85,18 +93,101 @@ public class PopConsumerRocksdbStoreTest {
.collect(Collectors.toList()));
List<PopConsumerRecord> consumerRecords =
consumerStore.scanExpiredRecords(20002, 2);
consumerStore.scanExpiredRecords(0, 20002, 2);
Assert.assertEquals(2, consumerRecords.size());
consumerStore.deleteRecords(consumerRecords);
consumerRecords = consumerStore.scanExpiredRecords(20002, 2);
consumerRecords = consumerStore.scanExpiredRecords(0, 20003, 2);
Assert.assertEquals(1, consumerRecords.size());
consumerStore.deleteRecords(consumerRecords);
consumerRecords = consumerStore.scanExpiredRecords(20004, 3);
consumerRecords = consumerStore.scanExpiredRecords(0, 20005, 3);
Assert.assertEquals(2, consumerRecords.size());
consumerStore.shutdown();
deleteStoreDirectory(filePath);
}
private long getDirectorySizeRecursive(File directory) {
long size = 0;
File[] files = directory.listFiles();
if (files != null) {
for (File file : files) {
if (file.isFile()) {
size += file.length();
} else if (file.isDirectory()) {
size += getDirectorySizeRecursive(file);
}
}
}
return size;
}
@Test
@Ignore
@SuppressWarnings("ConstantValue")
public void tombstoneDeletionTest() throws IllegalAccessException, NoSuchFieldException {
PopConsumerRocksdbStore rocksdbStore = new PopConsumerRocksdbStore(getRandomStorePath());
rocksdbStore.start();
int iterCount = 1000 * 1000;
boolean useSeekFirstDelete = false;
Field dbField = AbstractRocksDBStorage.class.getDeclaredField("db");
dbField.setAccessible(true);
RocksDB rocksDB = (RocksDB) dbField.get(rocksdbStore);
long currentTime = 0L;
Stopwatch stopwatch = Stopwatch.createStarted();
for (int i = 0; i < iterCount; i++) {
List<PopConsumerRecord> records = new ArrayList<>();
for (int j = 0; j < 1000; j++) {
PopConsumerRecord record = getConsumerRecord();
record.setPopTime((long) i * iterCount + j);
record.setGroupId("GroupTest");
record.setTopicId("TopicTest");
record.setQueueId(i % 10);
record.setRetryFlag(0);
record.setInvisibleTime(TimeUnit.SECONDS.toMillis(30));
record.setOffset(i);
records.add(record);
}
rocksdbStore.writeRecords(records);
long start = stopwatch.elapsed(TimeUnit.MILLISECONDS);
List<PopConsumerRecord> deleteList = new ArrayList<>();
if (useSeekFirstDelete) {
try (RocksIterator iterator = rocksDB.newIterator(rocksdbStore.columnFamilyHandle)) {
iterator.seekToFirst();
if (i % 10 == 0) {
long fileSize = getDirectorySizeRecursive(new File(rocksdbStore.getFilePath()));
log.info("DirectorySize={}, Cost={}ms",
MessageStoreUtil.toHumanReadable(fileSize), stopwatch.elapsed(TimeUnit.MILLISECONDS) - start);
}
while (iterator.isValid() && deleteList.size() < 1024) {
deleteList.add(PopConsumerRecord.decode(iterator.value()));
iterator.next();
}
}
} else {
long upper = System.currentTimeMillis();
deleteList = rocksdbStore.scanExpiredRecords(currentTime, upper, 800);
if (!deleteList.isEmpty()) {
currentTime = deleteList.get(deleteList.size() - 1).getVisibilityTimeout();
}
long scanCost = stopwatch.elapsed(TimeUnit.MILLISECONDS) - start;
if (i % 100 == 0) {
long fileSize = getDirectorySizeRecursive(new File(rocksdbStore.getFilePath()));
long seekTime = stopwatch.elapsed(TimeUnit.MILLISECONDS);
try (RocksIterator iterator = rocksDB.newIterator(rocksdbStore.columnFamilyHandle)) {
iterator.seekToFirst();
}
log.info("DirectorySize={}, Cost={}ms, SeekFirstCost={}ms", MessageStoreUtil.toHumanReadable(fileSize),
scanCost, stopwatch.elapsed(TimeUnit.MILLISECONDS) - seekTime);
}
}
rocksdbStore.deleteRecords(deleteList);
}
rocksdbStore.shutdown();
deleteStoreDirectory(rocksdbStore.getFilePath());
}
}
broker/src/test/java/org/apache/rocketmq/broker/pop/PopConsumerServiceTest.java
+7 -6
View File
@@ -25,6 +25,7 @@ import java.util.List;
import java.util.UUID;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;
import java.util.stream.Collectors;
import java.util.stream.IntStream;
import org.apache.commons.io.FileUtils;
@@ -371,17 +372,17 @@ public class PopConsumerServiceTest {
Mockito.doReturn(CompletableFuture.completedFuture(null))
.when(consumerServiceSpy).getMessageAsync(any(PopConsumerRecord.class));
consumerServiceSpy.revive(20 * 1000, 1);
consumerServiceSpy.revive(new AtomicLong(20 * 1000), 1);
Mockito.doReturn(CompletableFuture.completedFuture(
Triple.of(null, "GetMessageResult is null", false)))
.when(consumerServiceSpy).getMessageAsync(any(PopConsumerRecord.class));
consumerServiceSpy.revive(20 * 1000, 1);
consumerServiceSpy.revive(new AtomicLong(20 * 1000), 1);
Mockito.doReturn(CompletableFuture.completedFuture(
Triple.of(Mockito.mock(MessageExt.class), null, false)))
.when(consumerServiceSpy).getMessageAsync(any(PopConsumerRecord.class));
consumerServiceSpy.revive(20 * 1000, 1);
consumerServiceSpy.revive(new AtomicLong(20 * 1000), 1);
consumerService.shutdown();
}
@@ -412,11 +413,11 @@ public class PopConsumerServiceTest {
long visibleTimestamp = popTime + invisibleTime;
// revive fails
Assert.assertEquals(1, consumerServiceSpy.revive(visibleTimestamp, 1));
Assert.assertEquals(1, consumerServiceSpy.revive(new AtomicLong(visibleTimestamp), 1));
// should be invisible now
Assert.assertEquals(0, consumerService.getPopConsumerStore().scanExpiredRecords(visibleTimestamp, 1).size());
Assert.assertEquals(0, consumerService.getPopConsumerStore().scanExpiredRecords(0, visibleTimestamp, 1).size());
// will be visible again in 10 seconds
Assert.assertEquals(1, consumerService.getPopConsumerStore().scanExpiredRecords(visibleTimestamp + 10 * 1000, 1).size());
Assert.assertEquals(1, consumerService.getPopConsumerStore().scanExpiredRecords(visibleTimestamp, System.currentTimeMillis() + visibleTimestamp + 10 * 1000, 1).size());
consumerService.shutdown();
}
common/src/main/java/org/apache/rocketmq/common/config/AbstractRocksDBStorage.java
+3 -3
View File
@@ -365,7 +365,7 @@ public abstract class AbstractRocksDBStorage {
}
if (postLoad()) {
this.loaded = true;
LOGGER.info("RocksDB[{}] starts OK", this.dbPath);
LOGGER.info("RocksDB [{}] starts OK", this.dbPath);
this.closed = false;
return true;
} else {
@@ -437,9 +437,9 @@ public abstract class AbstractRocksDBStorage {
this.options = null;
this.loaded = false;
LOGGER.info("shutdown OK. {}", this.dbPath);
LOGGER.info("RocksDB shutdown OK. {}", this.dbPath);
} catch (Exception e) {
LOGGER.error("shutdown Failed. {}", this.dbPath, e);
LOGGER.error("RocksDB shutdown failed. {}", this.dbPath, e);
return false;
}
return true;