[ISSUE #9441] Add file length check in IndexService (#9442)

Author: lizhimins

tieredstore/src/main/java/org/apache/rocketmq/tieredstore/index/IndexStoreFile.java

@@ -314,7 +314,8 @@ protected CompletableFuture<List<IndexItem>> queryAsyncFromUnsealedFile(
     protected CompletableFuture<List<IndexItem>> queryAsyncFromSegmentFile(
         String key, int maxCount, long beginTime, long endTime) {
 
-        if (this.fileSegment == null || !UPLOAD.equals(this.fileStatus.get())) {
+        if (this.fileSegment == null || !UPLOAD.equals(this.fileStatus.get()) ||
+            this.fileSegment.getCommitPosition() <= this.getSlotPosition(0)) {
             return CompletableFuture.completedFuture(Collections.emptyList());
         }
 

tieredstore/src/main/java/org/apache/rocketmq/tieredstore/provider/FileSegment.java

@@ -325,22 +325,25 @@ public ByteBuffer read(long position, int length) {
 
     public CompletableFuture<ByteBuffer> readAsync(long position, int length) {
         CompletableFuture<ByteBuffer> future = new CompletableFuture<>();
+
         if (position < 0 || position >= commitPosition) {
-            future.completeExceptionally(new TieredStoreException(
-                TieredStoreErrorCode.ILLEGAL_PARAM, "FileSegment read position is illegal position"));
+            future.completeExceptionally(new TieredStoreException(TieredStoreErrorCode.ILLEGAL_PARAM,
+                String.format("FileSegment read position illegal, filePath=%s, fileType=%s, position=%d, length=%d, commit=%d",
+                    filePath, fileType, position, length, commitPosition)));
             return future;
         }
 
         if (length <= 0) {
-            future.completeExceptionally(new TieredStoreException(
-                TieredStoreErrorCode.ILLEGAL_PARAM, "FileSegment read length illegal"));
+            future.completeExceptionally(new TieredStoreException(TieredStoreErrorCode.ILLEGAL_PARAM,
+                String.format("FileSegment read length illegal, filePath=%s, fileType=%s, position=%d, length=%d, commit=%d",
+                    filePath, fileType, position, length, commitPosition)));
             return future;
         }
 
         int readableBytes = (int) (commitPosition - position);
         if (readableBytes < length) {
             length = readableBytes;
-            log.debug("FileSegment#readAsync, expect request position is greater than commit position, " +
+            log.debug("FileSegment expect request position is greater than commit position, " +
                     "file: {}, request position: {}, commit position: {}, change length from {} to {}",
                 getPath(), position, commitPosition, length, readableBytes);
         }