[ISSUE #7974] Add repeatedly read same offset log to find unexpected situations (#7975)

tieredstore/src/main/java/org/apache/rocketmq/tieredstore/common/SelectBufferResult.java

@@ -18,6 +18,7 @@
 package org.apache.rocketmq.tieredstore.common;
 
 import java.nio.ByteBuffer;
+import java.util.concurrent.atomic.AtomicLong;
 
 public class SelectBufferResult {
 
@@ -25,12 +26,14 @@ public class SelectBufferResult {
     private final long startOffset;
     private final int size;
     private final long tagCode;
+    private final AtomicLong accessCount;
 
     public SelectBufferResult(ByteBuffer byteBuffer, long startOffset, int size, long tagCode) {
         this.startOffset = startOffset;
         this.byteBuffer = byteBuffer;
         this.size = size;
         this.tagCode = tagCode;
+        this.accessCount = new AtomicLong();
     }
 
     public ByteBuffer getByteBuffer() {
@@ -48,4 +51,8 @@ public class SelectBufferResult {
     public long getTagCode() {
         return tagCode;
     }
+
+    public AtomicLong getAccessCount() {
+        return accessCount;
+    }
 }

tieredstore/src/main/java/org/apache/rocketmq/tieredstore/core/MessageStoreFetcherImpl.java

@@ -76,7 +76,10 @@ public class MessageStoreFetcherImpl implements MessageStoreFetcher {
 
         return Caffeine.newBuilder()
             .scheduler(Scheduler.systemScheduler())
-            .expireAfterWrite(storeConfig.getReadAheadCacheExpireDuration(), TimeUnit.MILLISECONDS)
+            // Clients may repeatedly request messages at the same offset in tiered storage,
+            // causing the request queue to become full. Using expire after read or write policy
+            // to refresh the cache expiration time.
+            .expireAfterAccess(storeConfig.getReadAheadCacheExpireDuration(), TimeUnit.MILLISECONDS)
             .maximumWeight(memoryMaxSize)
             // Using the buffer size of messages to calculate memory usage
             .weigher((String key, SelectBufferResult buffer) -> buffer.getSize())
@@ -98,7 +101,15 @@ public class MessageStoreFetcherImpl implements MessageStoreFetcher {
         SelectBufferResult buffer = this.fetcherCache.getIfPresent(
             String.format(CACHE_KEY_FORMAT, mq.getTopic(), mq.getQueueId(), offset));
         // return duplicate buffer here
-        return buffer == null ? null : new SelectBufferResult(
+        if (buffer == null) {
+            return null;
+        }
+        long count = buffer.getAccessCount().incrementAndGet();
+        if (count % 1000L == 0L) {
+            log.warn("MessageFetcher fetch same offset message too many times, " +
+                "topic={}, queueId={}, offset={}, count={}", mq.getTopic(), mq.getQueueId(), offset, count);
+        }
+        return new SelectBufferResult(
             buffer.getByteBuffer().asReadOnlyBuffer(), buffer.getStartOffset(), buffer.getSize(), buffer.getTagCode());
     }