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

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

Author: lizhimins

tieredstore/src/main/java/org/apache/rocketmq/tieredstore/TieredMessageStore.java

@@ -362,8 +362,7 @@ public long getMessageStoreTimeStamp(String topic, int queueId, long consumeQueu
     }
 
     @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 CompletableFuture<Long> getMessageStoreTimeStampAsync(String topic, int q
                         .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

@@ -25,6 +25,7 @@
 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 CompletableFuture<Long> getMessageStoreTimeStampAsync(String topic, int q
             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