diff --git a/s3stream/src/main/java/com/automq/stream/s3/S3Storage.java b/s3stream/src/main/java/com/automq/stream/s3/S3Storage.java index 008238edff..a3da6bf38a 100644 --- a/s3stream/src/main/java/com/automq/stream/s3/S3Storage.java +++ b/s3stream/src/main/java/com/automq/stream/s3/S3Storage.java @@ -483,7 +483,7 @@ private CompletableFuture 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 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 records) { long expectStartOffset = -1L; for (StreamRecordBatch record : records) {