opensource/rocketmq
3
0
Fork 0
mirror of https://github.com/apache/rocketmq.git synced 2026-08-30 18:10:44 +08:00
Code Issues Packages Projects Releases Wiki Activity

[ISSUE #9912] Reduce excessive requests for consumer offset timestamps in tiered storage (#9991)

Browse Source 
* [ISSUE #9912] Reduce excessive requests for consumer offset timestamps in tiered storage

Signed-off-by: terrance.lzm <terrance.lzm@alibaba-inc.com>

* [ISSUE #9912] Reduce excessive requests for consumer offset timestamps in tiered storage

---------

Signed-off-by: terrance.lzm <terrance.lzm@alibaba-inc.com>
This commit is contained in:
lizhimins
2026-01-13 09:56:22 +08:00
committed by GitHub
parent 9fecafe086
commit 4eead1364d
3 changed files with 37 additions and 20 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tieredstore/src/main/java/org/apache/rocketmq/tieredstore/TieredMessageStore.java
+1 -7
View File
@@ -362,8 +362,7 @@ public class TieredMessageStore extends AbstractPluginMessageStore {
}
@Override
public CompletableFuture<Long> getMessageStoreTimeStampAsync(String topic, int queueId,
long consumeQueueOffset) {
public CompletableFuture<Long> getMessageStoreTimeStampAsync(String topic, int queueId, long consumeQueueOffset) {
if (fetchFromCurrentStore(topic, queueId, consumeQueueOffset)) {
Stopwatch stopwatch = Stopwatch.createStarted();
return fetcher.getMessageStoreTimeStampAsync(topic, queueId, consumeQueueOffset)
@@ -374,11 +373,6 @@ public class TieredMessageStore extends AbstractPluginMessageStore {
.put(TieredStoreMetricsConstant.LABEL_TOPIC, topic)
.build();
TieredStoreMetricsManager.apiLatency.record(stopwatch.elapsed(TimeUnit.MILLISECONDS), latencyAttributes);
if (time == -1) {
log.debug("GetEarliestMessageTimeAsync failed, try to get message time from next store, topic: {}, queue: {}, queue offset: {}",
topic, queueId, consumeQueueOffset);
return next.getMessageStoreTimeStamp(topic, queueId, consumeQueueOffset);
}
return time;
});
}
tieredstore/src/main/java/org/apache/rocketmq/tieredstore/core/MessageStoreFetcherImpl.java
+33 -12
View File
@@ -25,6 +25,7 @@ import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;
import org.apache.rocketmq.common.BoundaryType;
import org.apache.rocketmq.common.MixAll;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.store.GetMessageResult;
import org.apache.rocketmq.store.GetMessageStatus;
@@ -52,6 +53,7 @@ public class MessageStoreFetcherImpl implements MessageStoreFetcher {
private static final Logger log = LoggerFactory.getLogger(MessageStoreUtil.TIERED_STORE_LOGGER_NAME);
protected static final String CACHE_KEY_FORMAT = "%s@%d@%d";
protected static final String FETCHER_GROUP_NAME = MixAll.CID_RMQ_SYS_PREFIX + "FETCHER_TIMESTAMP";
private final String brokerName;
private final MetadataStore metadataStore;
@@ -389,18 +391,37 @@ public class MessageStoreFetcherImpl implements MessageStoreFetcher {
return CompletableFuture.completedFuture(-1L);
}
return flatFile.getConsumeQueueAsync(queueOffset)
.thenComposeAsync(cqItem -> {
long commitLogOffset = MessageFormatUtil.getCommitLogOffsetFromItem(cqItem);
int size = MessageFormatUtil.getSizeFromItem(cqItem);
return flatFile.getCommitLogAsync(commitLogOffset, size);
}, messageStore.getStoreExecutor().bufferFetchExecutor)
.thenApply(MessageFormatUtil::getStoreTimeStamp)
.exceptionally(e -> {
log.error("MessageStoreFetcherImpl#getMessageStoreTimeStampAsync: " +
"get or decode message failed, topic={}, queue={}, offset={}", topic, queueId, queueOffset, e);
return -1L;
});
// The Metrics thread frequently retrieves the storage timestamp of the latest message;
// as an alternative, return the queue's saved timestamp here.
if (queueOffset + 1L == flatFile.getConsumeQueueCommitOffset()) {
long timestamp = flatFile.getMaxStoreTimestamp();
return CompletableFuture.completedFuture(timestamp == Long.MAX_VALUE ? -1L : timestamp);
}
CompletableFuture<Long> future = new CompletableFuture<>();
try {
this.getMessageAsync(FETCHER_GROUP_NAME, topic, queueId, queueOffset, 1, null)
.whenComplete((result, e) -> {
if (e != null) {
log.error("MessageStoreFetcherImpl#getMessageStoreTimeStampAsync: " +
"Get or decode message failed, topic={}, queue={}, offset={}", topic, queueId, queueOffset, e);
future.completeExceptionally(e);
return;
}
if (result != null && result.getMessageBufferList() != null
&& !result.getMessageBufferList().isEmpty()) {
long timestamp = MessageFormatUtil.getStoreTimeStamp(result.getMessageBufferList().get(0));
log.info("MessageStoreFetcherImpl#getMessageStoreTimeStampAsync: " +
"topic={}, queue={}, offset={}, timestamp={}", topic, queueId, queueOffset, timestamp);
future.complete(timestamp);
} else {
future.complete(-1L);
}
});
} catch (Throwable t) {
future.completeExceptionally(t);
}
return future;
}
@Override
tieredstore/src/test/java/org/apache/rocketmq/tieredstore/TieredMessageStoreTest.java
+3 -1
View File
@@ -268,8 +268,10 @@ public class TieredMessageStoreTest {
configuration.update(properties);
Assert.assertEquals(1, (long) currentStore.getMessageStoreTimeStampAsync(mq.getTopic(), mq.getQueueId(), 0).join());
// If data cannot be fetched from tiered storage,
// there is no need to fallback to local storage.
Mockito.when(fetcher.getMessageStoreTimeStampAsync(anyString(), anyInt(), anyLong())).thenReturn(CompletableFuture.completedFuture(-1L));
Assert.assertEquals(3, (long) currentStore.getMessageStoreTimeStampAsync(mq.getTopic(), mq.getQueueId(), 0).join());
Assert.assertEquals(-1L, (long) currentStore.getMessageStoreTimeStampAsync(mq.getTopic(), mq.getQueueId(), 0).join());
}
@Test