From 51dc8f40f0a4e724535df5040bfe08e49d3f2968 Mon Sep 17 00:00:00 2001 From: Ashutosh Samantaray Date: Mon, 12 Oct 2026 14:45:42 +0530 Subject: [PATCH 1/3] fix(s3stream): preserve pending multipart buffers through copy and release Committed-By-Agent: codex Co-authored-by: codex --- .../stream/s3/operator/MultiPartWriter.java | 91 +++++++++++++----- .../stream/s3/operator/ProxyWriterTest.java | 94 +++++++++++++++++++ 2 files changed, 163 insertions(+), 22 deletions(-) diff --git a/s3stream/src/main/java/com/automq/stream/s3/operator/MultiPartWriter.java b/s3stream/src/main/java/com/automq/stream/s3/operator/MultiPartWriter.java index e6a6b9ddd3..a7e24d1251 100644 --- a/s3stream/src/main/java/com/automq/stream/s3/operator/MultiPartWriter.java +++ b/s3stream/src/main/java/com/automq/stream/s3/operator/MultiPartWriter.java @@ -57,6 +57,7 @@ public class MultiPartWriter implements Writer { private final AtomicLong totalWriteSize = new AtomicLong(0L); private String uploadId; private CompletableFuture closeCf; + private CompletableFuture releaseCf; private ObjectPart objectPart = null; public MultiPartWriter(ObjectStorage.WriteOptions writeOptions, ObjectStorage operator, String path, @@ -198,17 +199,19 @@ public CompletableFuture close() { @Override public CompletableFuture release() { - List> partsToWait = parts; - if (objectPart != null) { - // skip waiting for pending part - partsToWait = partsToWait.subList(0, partsToWait.size() - 1); + if (releaseCf != null) { + return releaseCf; } - // wait for all ongoing uploading parts to finish and release pending part - return CompletableFuture.allOf(partsToWait.toArray(new CompletableFuture[0])).whenComplete((nil, ex) -> { - if (objectPart != null) { - objectPart.release(); - } - }); + ObjectPart pendingPart = objectPart; + List> partsToWait = pendingPart == null + ? parts : parts.subList(0, parts.size() - 1); + CompletableFuture pendingRelease = pendingPart == null + ? CompletableFuture.completedFuture(null) : pendingPart.release(); + // A pending range read may still append to the part buffer. Wait for it before freeing the buffer or + // reporting cleanup complete, while also waiting for parts that were already submitted for upload. + releaseCf = CompletableFuture.allOf( + CompletableFuture.allOf(partsToWait.toArray(new CompletableFuture[0])), pendingRelease); + return releaseCf; } @Override @@ -233,6 +236,7 @@ class ObjectPart { private final ThrottleStrategy throttleStrategy; private CompositeByteBuf partBuf = ByteBufAlloc.compositeByteBuffer(); private CompletableFuture lastRangeReadCf = CompletableFuture.completedFuture(null); + private final List pendingWrites = new LinkedList<>(); private long size; public ObjectPart(ThrottleStrategy throttleStrategy) { @@ -240,20 +244,26 @@ public ObjectPart(ThrottleStrategy throttleStrategy) { parts.add(partCf); } - public void write(ByteBuf data) { + public synchronized void write(ByteBuf data) { size += data.readableBytes(); + PendingWrite pendingWrite = new PendingWrite(data); + pendingWrites.add(pendingWrite); // ensure addComponent happen before following write or copyWrite. this.lastRangeReadCf = lastRangeReadCf.handle((nil, ex) -> { - if (ex != null) { - data.release(); - throw new CompletionException(ex); + synchronized (this) { + pendingWrites.remove(pendingWrite); + ByteBuf part = pendingWrite.take(); + if (ex != null) { + part.release(); + throw new CompletionException(ex); + } + partBuf.addComponent(true, part); + return null; } - partBuf.addComponent(true, data); - return null; }); } - public void copyOnWrite() { + public synchronized void copyOnWrite() { int size = partBuf.readableBytes(); if (size > 0) { ByteBuf buf = ByteBufAlloc.byteBuffer(size, writeOptions.allocType()); @@ -262,6 +272,7 @@ public void copyOnWrite() { this.partBuf.release(); this.partBuf = copy; } + pendingWrites.forEach(PendingWrite::copyOnWrite); } public void readAndWrite(S3ObjectMetadata s3ObjectMetadata, long start, long end) { @@ -273,13 +284,19 @@ public void readAndWrite(S3ObjectMetadata s3ObjectMetadata, long start, long end new ReadOptions().throttleStrategy(throttleStrategy).bucket(s3ObjectMetadata.bucket()), s3ObjectMetadata.key(), start, end) ) - .thenAccept(buf -> partBuf.addComponent(true, buf)); + .thenAccept(buf -> { + synchronized (this) { + partBuf.addComponent(true, buf); + } + }); } public void upload() { this.lastRangeReadCf.whenComplete((nil, ex) -> { if (ex != null) { - partBuf.release(); + synchronized (this) { + partBuf.release(); + } partCf.completeExceptionally(ex); } else { upload0(); @@ -289,7 +306,11 @@ public void upload() { private void upload0() { TimerUtil timerUtil = new TimerUtil(); - FutureUtil.propagate(uploadIdCf.thenCompose(uploadId -> operator.uploadPart(writeOptions, path, uploadId, partNumber, partBuf)), partCf); + ByteBuf data; + synchronized (this) { + data = partBuf; + } + FutureUtil.propagate(uploadIdCf.thenCompose(uploadId -> operator.uploadPart(writeOptions, path, uploadId, partNumber, data)), partCf); partCf.whenComplete((nil, ex) -> { S3ObjectMetrics.recordUploadPartStage(timerUtil.elapsedAs(TimeUnit.NANOSECONDS)); }); @@ -303,8 +324,34 @@ public CompletableFuture getFuture() { return partCf.thenApply(nil -> null); } - public void release() { - partBuf.release(); + public CompletableFuture release() { + return lastRangeReadCf.handle((nil, ex) -> { + synchronized (this) { + partBuf.release(); + } + return null; + }); + } + + private class PendingWrite { + private ByteBuf data; + + private PendingWrite(ByteBuf data) { + this.data = data; + } + + private ByteBuf take() { + ByteBuf result = data; + data = null; + return result; + } + + private void copyOnWrite() { + ByteBuf copy = ByteBufAlloc.byteBuffer(data.readableBytes(), writeOptions.allocType()); + copy.writeBytes(data.duplicate()); + data.release(); + data = copy; + } } } diff --git a/s3stream/src/test/java/com/automq/stream/s3/operator/ProxyWriterTest.java b/s3stream/src/test/java/com/automq/stream/s3/operator/ProxyWriterTest.java index 42fff5634d..c7f5dee230 100644 --- a/s3stream/src/test/java/com/automq/stream/s3/operator/ProxyWriterTest.java +++ b/s3stream/src/test/java/com/automq/stream/s3/operator/ProxyWriterTest.java @@ -302,4 +302,98 @@ public void testReleaseDiscardsPendingOperationsWithoutStartingReads() { verify(operator, times(0)).rangeRead(any(), any(), anyLong(), anyLong()); } + @Test + public void testMultipartReplayDoesNotReadBeforeRangeStart() { + // Given seven small copies followed by a larger copy, multipart replay must flush smaller parts before + // calculating the next range, so every range read has a valid end offset. + int mib = 1024 * 1024; + when(operator.createMultipartUpload(any(), eq("testpath"))) + .thenReturn(CompletableFuture.completedFuture("test_upload_id")); + when(operator.rangeRead(any(), any(), anyLong(), anyLong())).thenAnswer(inv -> { + long start = inv.getArgument(2); + long end = inv.getArgument(3); + assertTrue(end > start); + return CompletableFuture.completedFuture(TestUtils.random(1)); + }); + when(operator.uploadPart(any(), eq("testpath"), eq("test_upload_id"), anyInt(), any())) + .thenAnswer(inv -> CompletableFuture.completedFuture(new AbstractObjectStorage.ObjectStorageCompletedPart( + inv.getArgument(3), "etag", "checksum"))); + when(operator.completeMultipartUpload(any(), eq("testpath"), eq("test_upload_id"), any())) + .thenReturn(CompletableFuture.completedFuture(null)); + + for (int i = 0; i < 7; i++) { + writer.copyWrite(new S3ObjectMetadata(i + 1, 4L * mib, S3ObjectType.STREAM), 0, 4L * mib); + } + writer.copyWrite(new S3ObjectMetadata(8, 8L * mib, S3ObjectType.STREAM), 0, 8L * mib); + + assertFalse(writer.close().isCompletedExceptionally()); + verify(operator, times(0)).write(any(), any(), any()); + } + + @Test + public void testReleaseWaitsForPendingMultipartRead() { + // Given multipart selection with a pending range read, release must wait to free the returned buffer and + // must not upload the part after cleanup is requested. + int mib = 1024 * 1024; + CompletableFuture readCf = new CompletableFuture<>(); + when(operator.createMultipartUpload(any(), eq("testpath"))) + .thenReturn(CompletableFuture.completedFuture("test_upload_id")); + when(operator.uploadPartCopy(any(), any(), eq("testpath"), anyLong(), anyLong(), + eq("test_upload_id"), anyInt())) + .thenAnswer(inv -> CompletableFuture.completedFuture(new AbstractObjectStorage.ObjectStorageCompletedPart( + inv.getArgument(6), "etag", "checksum"))); + when(operator.rangeRead(any(), any(), anyLong(), anyLong())).thenReturn(readCf); + + writer.copyWrite(new S3ObjectMetadata(1, 30L * mib, S3ObjectType.STREAM), 0, 30L * mib); + writer.copyWrite(new S3ObjectMetadata(2, 3L * mib, S3ObjectType.STREAM), 0, 3L * mib); + CompletableFuture releaseCf = writer.release(); + assertFalse(releaseCf.isDone()); + + ByteBuf read = TestUtils.randomPooled(3 * mib); + readCf.complete(read); + + assertTrue(releaseCf.isDone()); + assertEquals(0, read.refCnt()); + verify(operator, times(0)).uploadPart(any(), any(), any(), anyInt(), any()); + } + + @Test + public void testCopyOnWriteSnapshotsWriteWaitingForMultipartRead() { + // Given a write queued behind a multipart range read, copyOnWrite must snapshot its buffer before the + // caller changes the original bytes. + int mib = 1024 * 1024; + CompletableFuture readCf = new CompletableFuture<>(); + byte[] uploadedMarker = new byte[1]; + when(operator.createMultipartUpload(any(), eq("testpath"))) + .thenReturn(CompletableFuture.completedFuture("test_upload_id")); + when(operator.uploadPartCopy(any(), any(), eq("testpath"), anyLong(), anyLong(), + eq("test_upload_id"), anyInt())) + .thenAnswer(inv -> CompletableFuture.completedFuture(new AbstractObjectStorage.ObjectStorageCompletedPart( + inv.getArgument(6), "etag", "checksum"))); + when(operator.rangeRead(any(), any(), anyLong(), anyLong())).thenReturn(readCf); + when(operator.uploadPart(any(), eq("testpath"), eq("test_upload_id"), anyInt(), any())) + .thenAnswer(inv -> { + ByteBuf part = inv.getArgument(4); + uploadedMarker[0] = part.getByte(3 * mib); + part.release(); + return CompletableFuture.completedFuture(new AbstractObjectStorage.ObjectStorageCompletedPart( + inv.getArgument(3), "etag", "checksum")); + }); + when(operator.completeMultipartUpload(any(), eq("testpath"), eq("test_upload_id"), any())) + .thenReturn(CompletableFuture.completedFuture(null)); + + writer.copyWrite(new S3ObjectMetadata(1, 30L * mib, S3ObjectType.STREAM), 0, 30L * mib); + writer.copyWrite(new S3ObjectMetadata(2, 3L * mib, S3ObjectType.STREAM), 0, 3L * mib); + ByteBuf source = TestUtils.randomPooled(8); + byte original = source.getByte(0); + writer.write(source.retainedDuplicate()); + writer.copyOnWrite(); + source.setByte(0, original + 1); + + readCf.complete(TestUtils.randomPooled(3 * mib)); + assertTrue(writer.close().isDone()); + assertEquals(original, uploadedMarker[0]); + source.release(); + } + } From 460543f907b51de0f772707adda984cd3fea1ff1 Mon Sep 17 00:00:00 2001 From: Ashutosh Samantaray Date: Mon, 12 Oct 2026 14:59:01 +0530 Subject: [PATCH 2/3] fix(s3stream): flush large writes before copies and snapshot queued parts Committed-By-Agent: codex Co-authored-by: codex --- .../stream/s3/operator/MultiPartWriter.java | 13 ++++ .../stream/s3/operator/ProxyWriterTest.java | 74 ++++++++++++++++++- 2 files changed, 85 insertions(+), 2 deletions(-) diff --git a/s3stream/src/main/java/com/automq/stream/s3/operator/MultiPartWriter.java b/s3stream/src/main/java/com/automq/stream/s3/operator/MultiPartWriter.java index a7e24d1251..b49f9261e7 100644 --- a/s3stream/src/main/java/com/automq/stream/s3/operator/MultiPartWriter.java +++ b/s3stream/src/main/java/com/automq/stream/s3/operator/MultiPartWriter.java @@ -159,6 +159,14 @@ public void copyWrite0(S3ObjectMetadata sourceObjectMateData, long start, long e } } else { if (objectPart.size() + targetSize > maxMergeWriteSize) { + if (objectPart.size() >= minPartSize) { + // A buffered write can already exceed minPartSize. Upload it before starting a copy rather than + // calculating a negative range end from the remaining bytes needed to fill the part. + objectPart.upload(); + this.objectPart = null; + copyWrite0(sourceObjectMateData, start, end); + return; + } long readAndWriteCopyEnd = start + minPartSize - objectPart.size(); objectPart.readAndWrite(sourceObjectMateData, start, readAndWriteCopyEnd); objectPart.upload(); @@ -292,6 +300,11 @@ public void readAndWrite(S3ObjectMetadata s3ObjectMetadata, long start, long end } public void upload() { + if (!lastRangeReadCf.isDone()) { + // This part is no longer reachable through MultiPartWriter.copyOnWrite(), but queued writes may + // still hold caller-owned buffers until the read finishes. Snapshot them before submitting it. + copyOnWrite(); + } this.lastRangeReadCf.whenComplete((nil, ex) -> { if (ex != null) { synchronized (this) { diff --git a/s3stream/src/test/java/com/automq/stream/s3/operator/ProxyWriterTest.java b/s3stream/src/test/java/com/automq/stream/s3/operator/ProxyWriterTest.java index c7f5dee230..4306e58b11 100644 --- a/s3stream/src/test/java/com/automq/stream/s3/operator/ProxyWriterTest.java +++ b/s3stream/src/test/java/com/automq/stream/s3/operator/ProxyWriterTest.java @@ -157,8 +157,8 @@ public void testCopyWrite_escalationPreservesOrder() { when(operator.completeMultipartUpload(any(), eq("testpath"), eq("test_upload_id"), any())) .thenReturn(CompletableFuture.completedFuture(null)); - // rangeRead hands back a not-yet-completed future per call so we can drive completion order by hand. The - // multipart replay chain issues each read only after the previous one completes. + // rangeRead hands back a not-yet-completed future per call so we can drive completion order by hand. + // Reads are serialized within each multipart part; different parts can start reads concurrently. List> reads = new ArrayList<>(); List readSizes = new ArrayList<>(); List readMarkers = new ArrayList<>(); @@ -330,6 +330,37 @@ public void testMultipartReplayDoesNotReadBeforeRangeStart() { verify(operator, times(0)).write(any(), any(), any()); } + @Test + public void testLargeWriteBeforeCopyFlushesWithoutInvalidRange() { + // Given a 28 MiB write followed by an 8 MiB copy, flush the buffered write before issuing the copy. + // No range read may be constructed with a negative end offset. + int mib = 1024 * 1024; + when(operator.createMultipartUpload(any(), eq("testpath"))) + .thenReturn(CompletableFuture.completedFuture("test_upload_id")); + when(operator.uploadPart(any(), eq("testpath"), eq("test_upload_id"), anyInt(), any())) + .thenAnswer(inv -> { + ByteBuf part = inv.getArgument(4); + part.release(); + return CompletableFuture.completedFuture(new AbstractObjectStorage.ObjectStorageCompletedPart( + inv.getArgument(3), "etag", "checksum")); + }); + when(operator.uploadPartCopy(any(), any(), eq("testpath"), anyLong(), anyLong(), + eq("test_upload_id"), anyInt())) + .thenAnswer(inv -> CompletableFuture.completedFuture(new AbstractObjectStorage.ObjectStorageCompletedPart( + inv.getArgument(6), "etag", "checksum"))); + when(operator.completeMultipartUpload(any(), eq("testpath"), eq("test_upload_id"), any())) + .thenReturn(CompletableFuture.completedFuture(null)); + + writer.write(TestUtils.randomPooled(28 * mib)); + writer.copyWrite(new S3ObjectMetadata(1, 8L * mib, S3ObjectType.STREAM), 0, 8L * mib); + + assertTrue(writer.close().isDone()); + assertFalse(writer.close().isCompletedExceptionally()); + verify(operator, times(0)).rangeRead(any(), any(), anyLong(), anyLong()); + verify(operator).uploadPartCopy(any(), any(), eq("testpath"), eq(0L), eq(8L * mib), + eq("test_upload_id"), anyInt()); + } + @Test public void testReleaseWaitsForPendingMultipartRead() { // Given multipart selection with a pending range read, release must wait to free the returned buffer and @@ -396,4 +427,43 @@ public void testCopyOnWriteSnapshotsWriteWaitingForMultipartRead() { source.release(); } + @Test + public void testCopyOnWriteSnapshotsQueuedWriteAfterPartSubmission() { + // Given a queued write makes a part large enough to upload before its range read completes, the writer + // must detach the caller's buffer before copyOnWrite returns and the caller changes it. + int mib = 1024 * 1024; + CompletableFuture readCf = new CompletableFuture<>(); + byte[] uploadedMarker = new byte[1]; + when(operator.createMultipartUpload(any(), eq("testpath"))) + .thenReturn(CompletableFuture.completedFuture("test_upload_id")); + when(operator.uploadPartCopy(any(), any(), eq("testpath"), anyLong(), anyLong(), + eq("test_upload_id"), anyInt())) + .thenAnswer(inv -> CompletableFuture.completedFuture(new AbstractObjectStorage.ObjectStorageCompletedPart( + inv.getArgument(6), "etag", "checksum"))); + when(operator.rangeRead(any(), any(), anyLong(), anyLong())).thenReturn(readCf); + when(operator.uploadPart(any(), eq("testpath"), eq("test_upload_id"), anyInt(), any())) + .thenAnswer(inv -> { + ByteBuf part = inv.getArgument(4); + uploadedMarker[0] = part.getByte(3 * mib); + part.release(); + return CompletableFuture.completedFuture(new AbstractObjectStorage.ObjectStorageCompletedPart( + inv.getArgument(3), "etag", "checksum")); + }); + when(operator.completeMultipartUpload(any(), eq("testpath"), eq("test_upload_id"), any())) + .thenReturn(CompletableFuture.completedFuture(null)); + + writer.copyWrite(new S3ObjectMetadata(1, 30L * mib, S3ObjectType.STREAM), 0, 30L * mib); + writer.copyWrite(new S3ObjectMetadata(2, 3L * mib, S3ObjectType.STREAM), 0, 3L * mib); + ByteBuf source = TestUtils.randomPooled(31 * mib); + byte original = source.getByte(0); + writer.write(source.retainedDuplicate()); + writer.copyOnWrite(); + source.setByte(0, original + 1); + + readCf.complete(TestUtils.randomPooled(3 * mib)); + assertTrue(writer.close().isDone()); + assertEquals(original, uploadedMarker[0]); + source.release(); + } + } From 7c9004391f49745dc97789ca49dfbb19b4a538ec Mon Sep 17 00:00:00 2001 From: Ashutosh Samantaray Date: Mon, 12 Oct 2026 15:13:55 +0530 Subject: [PATCH 3/3] fix(s3stream): make proxy PutObject close and release safe Committed-By-Agent: codex Co-authored-by: codex --- .../stream/s3/operator/ProxyWriter.java | 9 ++++ .../stream/s3/operator/ProxyWriterTest.java | 52 +++++++++++++++++++ 2 files changed, 61 insertions(+) diff --git a/s3stream/src/main/java/com/automq/stream/s3/operator/ProxyWriter.java b/s3stream/src/main/java/com/automq/stream/s3/operator/ProxyWriter.java index 2d7e13e421..33ec29daae 100644 --- a/s3stream/src/main/java/com/automq/stream/s3/operator/ProxyWriter.java +++ b/s3stream/src/main/java/com/automq/stream/s3/operator/ProxyWriter.java @@ -54,6 +54,7 @@ public class ProxyWriter implements Writer { private final CompletableFuture objectCf = new CompletableFuture<>(); private final TimerUtil timerUtil = new TimerUtil(); private Writer multiPartObjectWriter = null; + private boolean putObjectCloseStarted; private long size; public ProxyWriter(WriteOptions writeOptions, ObjectStorage objectStorage, String path, long minPartSize) { @@ -115,6 +116,10 @@ public CompletableFuture close() { if (multiPartObjectWriter != null) { return multiPartObjectWriter.close(); } + if (putObjectCloseStarted) { + return objectCf; + } + putObjectCloseStarted = true; return closeWithPutObject(); } @@ -123,6 +128,10 @@ public CompletableFuture release() { if (multiPartObjectWriter != null) { return multiPartObjectWriter.release(); } + if (putObjectCloseStarted) { + // The replay or PutObject still owns its buffers. Its completion follows their cleanup. + return objectCf.handle((nil, ex) -> null); + } pendingOperations.forEach(PendingOperation::release); pendingOperations.clear(); return CompletableFuture.completedFuture(null); diff --git a/s3stream/src/test/java/com/automq/stream/s3/operator/ProxyWriterTest.java b/s3stream/src/test/java/com/automq/stream/s3/operator/ProxyWriterTest.java index 4306e58b11..d172cf09c5 100644 --- a/s3stream/src/test/java/com/automq/stream/s3/operator/ProxyWriterTest.java +++ b/s3stream/src/test/java/com/automq/stream/s3/operator/ProxyWriterTest.java @@ -39,6 +39,7 @@ import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertSame; import static org.junit.jupiter.api.Assertions.assertTrue; import static org.mockito.ArgumentMatchers.any; import static org.mockito.ArgumentMatchers.anyInt; @@ -73,6 +74,57 @@ public void testWrite_onePart() { Assertions.assertEquals(16 * 1024 * 1024, captor.getValue().readableBytes()); } + @Test + public void testPutObjectCloseIsIdempotent() { + // Given a PutObject is in progress, repeated close calls must return its future and issue only one write. + CompletableFuture uploadCf = new CompletableFuture<>(); + when(operator.write(any(), eq("testpath"), any())).thenReturn(uploadCf); + writer.write(TestUtils.randomPooled(16)); + + CompletableFuture closeCf = writer.close(); + assertSame(closeCf, writer.close()); + + ArgumentCaptor captor = ArgumentCaptor.forClass(ByteBuf.class); + verify(operator, times(1)).write(any(), eq("testpath"), captor.capture()); + captor.getValue().release(); + uploadCf.complete(new ObjectStorage.WriteResult((short) 0)); + + assertTrue(closeCf.isDone()); + assertSame(closeCf, writer.close()); + verify(operator, times(1)).write(any(), eq("testpath"), any()); + } + + @Test + public void testReleaseWaitsForPutObjectReplayAndUpload() { + // Given a small copy is still being read during close, release must wait for both replay and upload cleanup. + CompletableFuture readCf = new CompletableFuture<>(); + CompletableFuture uploadCf = new CompletableFuture<>(); + when(operator.rangeRead(any(), any(), anyLong(), anyLong())).thenReturn(readCf); + when(operator.write(any(), eq("testpath"), any())).thenReturn(uploadCf); + writer.copyWrite(new S3ObjectMetadata(1, 8, S3ObjectType.STREAM), 0, 8); + ByteBuf followingWrite = TestUtils.randomPooled(8); + writer.write(followingWrite); + + CompletableFuture closeCf = writer.close(); + CompletableFuture releaseCf = writer.release(); + assertFalse(closeCf.isDone()); + assertFalse(releaseCf.isDone()); + + ByteBuf read = TestUtils.randomPooled(8); + readCf.complete(read); + assertFalse(releaseCf.isDone()); + + ArgumentCaptor captor = ArgumentCaptor.forClass(ByteBuf.class); + verify(operator).write(any(), eq("testpath"), captor.capture()); + captor.getValue().release(); + uploadCf.complete(new ObjectStorage.WriteResult((short) 0)); + + assertTrue(closeCf.isDone()); + assertTrue(releaseCf.isDone()); + assertEquals(0, read.refCnt()); + assertEquals(0, followingWrite.refCnt()); + } + @Test public void testSmallCopyWritesExceedingMultipartUploadThreshold() { // Given small copy operations total exactly 32 MiB, stay on the PutObject path until one additional byte