feat(s3stream): add logs on completion of timeout read request (AutoMQ#1227)

Signed-off-by: Shichao Nie <[email protected]>

Author: SCNieh

s3stream/src/main/java/com/automq/stream/s3/S3Storage.java

@@ -483,7 +483,7 @@ private CompletableFuture<ReadDataBlock> read0(FetchContext context,
             }
             return new ReadDataBlock(rst, blockCacheRst.getCacheAccessType());
         }).whenComplete((rst, ex) -> {
-            timeout.cancel();
+            handleTimeout(timeout, streamId, startOffset, finalEndOffset, maxBytes);
             if (ex != null) {
                 LOGGER.error("read from block cache failed, stream={}, {}-{}, maxBytes: {}",
                     streamId, startOffset, finalEndOffset, maxBytes, ex);
@@ -492,6 +492,15 @@ private CompletableFuture<ReadDataBlock> read0(FetchContext context,
         });
     }
 
+    private void handleTimeout(Timeout timeout, long streamId, long startOffset, long finalEndOffset, int maxBytes) {
+        if (timeout.isExpired()) {
+            LOGGER.error("[POTENTIAL_BUG_RECOVERED] read from block cache completed, stream={}, [{},{}), maxBytes: {}",
+                    streamId, startOffset, finalEndOffset, maxBytes);
+        } else {
+            timeout.cancel();
+        }
+    }
+
     private void continuousCheck(List<StreamRecordBatch> records) {
         long expectStartOffset = -1L;
         for (StreamRecordBatch record : records) {