fix(s3stream): fix dead lock issue (#1810)

Signed-off-by: SSpirits <[email protected]>
AutoMQ · Aug 15, 2024 · 9e4eafb · 9e4eafb
1 parent fd03fb4
commit 9e4eafb
Showing 1 changed file with 3 additions and 2 deletions.
diff --git a/s3stream/src/main/java/com/automq/stream/s3/wal/impl/object/RecordAccumulator.java b/s3stream/src/main/java/com/automq/stream/s3/wal/impl/object/RecordAccumulator.java
@@ -155,7 +155,7 @@ public void start() {
                     .filter(uploadTime -> time.nanoseconds() - uploadTime > DEFAULT_UPLOAD_WARNING_TIMEOUT)
                     .count();
                 if (count > 0) {
-                    log.warn("Found {} pending upload tasks exceed 5s.", count);
+                    log.error("Found {} pending upload tasks exceed 5s.", count);
                 }
             } catch (Throwable ignore) {
             }
@@ -528,7 +528,8 @@ assert record != null;
                 if (throwable instanceof WALFencedException) {
                     List<Record> uploadedRecords = uploadMap.remove(firstOffset);
                     Throwable finalThrowable = throwable;
-                    uploadedRecords.forEach(record -> record.future.completeExceptionally(finalThrowable));
+                    // Release lock and complete future in callback thread.
+                    callbackService.submit(() -> uploadedRecords.forEach(record -> record.future.completeExceptionally(finalThrowable)));
                 } else if (throwable != null) {
                     // Never fail the write task, the under layer storage will retry forever.
                     log.error("[Bug] Failed to write records to S3: {}", firstOffset, throwable);