fix(compaction): prevent double release on compaction shutdown (#2117)

Signed-off-by: Shichao Nie <[email protected]>
AutoMQ · Nov 5, 2024 · c098ee3 · c098ee3
1 parent 40da95b
commit c098ee3
Show file tree

Hide file tree

Showing 2 changed files with 8 additions and 2 deletions.
diff --git a/s3stream/src/main/java/com/automq/stream/s3/StreamDataBlock.java b/s3stream/src/main/java/com/automq/stream/s3/StreamDataBlock.java
@@ -80,6 +80,13 @@ public CompletableFuture<ByteBuf> getDataCf() {
         return this.dataCf;
     }
 
+    public ByteBuf getAndReleaseData() {
+        if (refCount.getAndDecrement() == 0) {
+            throw new IllegalStateException("Data has already been released");
+        }
+        return this.dataCf.join();
+    }
+
     public void releaseRef() {
         refCount.decrementAndGet();
     }

diff --git a/s3stream/src/main/java/com/automq/stream/s3/compact/operator/DataBlockWriter.java b/s3stream/src/main/java/com/automq/stream/s3/compact/operator/DataBlockWriter.java
@@ -126,8 +126,7 @@ public CompletableFuture<Void> close() {
     private CompositeByteBuf groupWaitingBlocks() {
         CompositeByteBuf buf = ByteBufAlloc.compositeByteBuffer();
         for (StreamDataBlock block : waitingUploadBlocks) {
-            buf.addComponent(true, block.getDataCf().join());
-            block.releaseRef();
+            buf.addComponent(true, block.getAndReleaseData());
             completedBlocks.add(block);
             nextDataBlockPosition += block.getBlockSize();
         }