[ISSUE apache#9213] Fix get the earliest time error when data is clean up in tiered storage

Author: lizhimins

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

@@ -318,12 +318,14 @@ public CompletableFuture<Long> getEarliestMessageTimeAsync(String topic, int que
         return fetcher.getEarliestMessageTimeAsync(topic, queueId)
             .thenApply(time -> {
                 Attributes latencyAttributes = TieredStoreMetricsManager.newAttributesBuilder()
-                    .put(TieredStoreMetricsConstant.LABEL_OPERATION, TieredStoreMetricsConstant.OPERATION_API_GET_EARLIEST_MESSAGE_TIME)
+                    .put(TieredStoreMetricsConstant.LABEL_OPERATION,
+                        TieredStoreMetricsConstant.OPERATION_API_GET_EARLIEST_MESSAGE_TIME)
                     .put(TieredStoreMetricsConstant.LABEL_TOPIC, topic)
                     .build();
                 TieredStoreMetricsManager.apiLatency.record(stopwatch.elapsed(TimeUnit.MILLISECONDS), latencyAttributes);
                 if (time < 0) {
-                    log.debug("GetEarliestMessageTimeAsync failed, try to get earliest message time from next store: topic: {}, queue: {}",
+                    log.debug("GetEarliestMessageTimeAsync failed, " +
+                            "try to get earliest message time from next store: topic: {}, queue: {}",
                         topic, queueId);
                     return finalNextEarliestMessageTime != Long.MAX_VALUE ? finalNextEarliestMessageTime : -1;
                 }

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

@@ -375,12 +375,18 @@ public CompletableFuture<GetMessageResult> getMessageAsync(
     @Override
     public CompletableFuture<Long> getEarliestMessageTimeAsync(String topic, int queueId) {
         FlatMessageFile flatFile = flatFileStore.getFlatFile(new MessageQueue(topic, brokerName, queueId));
+
         if (flatFile == null) {
             return CompletableFuture.completedFuture(-1L);
         }
 
-        // read from timestamp to timestamp + length
         int length = MessageFormatUtil.STORE_TIMESTAMP_POSITION + 8;
+        if (flatFile.getConsumeQueueCommitOffset() <= flatFile.getConsumeQueueMinOffset() ||
+            flatFile.getCommitLogCommitOffset() - flatFile.getCommitLogMinOffset() <= length) {
+            return CompletableFuture.completedFuture(-1L);
+        }
+
+        // read from timestamp to timestamp + length
         return flatFile.getCommitLogAsync(flatFile.getCommitLogMinOffset(), length)
             .thenApply(MessageFormatUtil::getStoreTimeStamp);
     }