fix(s3stream): fix timeout detect

Signed-off-by: Robin Han <hanxvdovehx@gmail.com>
AutoMQ · Dec 5, 2023 · ece0ec2 · ece0ec2
1 parent 14caaef
commit ece0ec2
Show file tree

Hide file tree

Showing 4 changed files with 5 additions and 6 deletions.
diff --git a/pom.xml b/pom.xml
@@ -53,7 +53,7 @@
         <guava.version>32.1.3-jre</guava.version>
         <slf4j.version>2.0.9</slf4j.version>
         <snakeyaml.version>2.2</snakeyaml.version>
-        <s3stream.version>0.6.7-SNAPSHOT</s3stream.version>
+        <s3stream.version>0.6.8-SNAPSHOT</s3stream.version>
 
         <!-- Flat buffers related -->
         <flatbuffers.version>23.5.26</flatbuffers.version>

diff --git a/s3stream/pom.xml b/s3stream/pom.xml
@@ -22,7 +22,7 @@
     <modelVersion>4.0.0</modelVersion>
     <groupId>com.automq.elasticstream</groupId>
     <artifactId>s3stream</artifactId>
-    <version>0.6.7-SNAPSHOT</version>
+    <version>0.6.8-SNAPSHOT</version>
     <properties>
         <mockito-core.version>5.5.0</mockito-core.version>
         <junit-jupiter.version>5.10.0</junit-jupiter.version>

diff --git a/s3stream/src/main/java/com/automq/stream/s3/S3Storage.java b/s3stream/src/main/java/com/automq/stream/s3/S3Storage.java
@@ -360,7 +360,7 @@ private CompletableFuture<ReadDataBlock> read0(long streamId, long startOffset,
         if (!logCacheRecords.isEmpty()) {
             endOffset = logCacheRecords.get(0).getBaseOffset();
         }
-        Timeout timeout = timeoutDetect.newTimeout(t -> LOGGER.error("read from block cache timeout, stream={}, {}, maxBytes: {}", streamId, startOffset, maxBytes), 1, TimeUnit.MINUTES);
+        Timeout timeout = timeoutDetect.newTimeout(t -> LOGGER.warn("read from block cache timeout, stream={}, {}, maxBytes: {}", streamId, startOffset, maxBytes), 1, TimeUnit.MINUTES);
         return blockCache.read(streamId, startOffset, endOffset, maxBytes).thenApply(readDataBlock -> {
             List<StreamRecordBatch> rst = new ArrayList<>(readDataBlock.getRecords());
             int remainingBytesSize = maxBytes - rst.stream().mapToInt(StreamRecordBatch::size).sum();
@@ -378,11 +378,11 @@ private CompletableFuture<ReadDataBlock> read0(long streamId, long startOffset,
             continuousCheck(rst);
             return new ReadDataBlock(rst, readDataBlock.getCacheAccessType());
         }).whenComplete((rst, ex) -> {
+            timeout.cancel();
             if (ex != null) {
                 LOGGER.error("read from block cache failed, stream={}, {}-{}, maxBytes: {}",
                         streamId, startOffset, maxBytes, ex);
                 logCacheRecords.forEach(StreamRecordBatch::release);
-                timeout.cancel();
             }
         });
     }

diff --git a/s3stream/src/main/java/com/automq/stream/s3/operator/DefaultS3Operator.java b/s3stream/src/main/java/com/automq/stream/s3/operator/DefaultS3Operator.java
@@ -187,8 +187,7 @@ public CompletableFuture<ByteBuf> rangeRead(String path, long start, long end, T
         }
 
         Timeout timeout = timeoutDetect.newTimeout((t) -> LOGGER.warn("rangeRead {} {}-{} timeout", path, start, end), 1, TimeUnit.MINUTES);
-        cf.whenComplete((rst, ex) -> timeout.cancel());
-        return cf;
+        return cf.whenComplete((rst, ex) -> timeout.cancel());
     }
 
     private void rangeRead0(String path, long start, long end, CompletableFuture<ByteBuf> cf) {