diff --git a/docs/content.zh/docs/deployment/filesystems/s3.md b/docs/content.zh/docs/deployment/filesystems/s3.md index f8f7bad7a8e8e9..317c4459ce22f5 100644 --- a/docs/content.zh/docs/deployment/filesystems/s3.md +++ b/docs/content.zh/docs/deployment/filesystems/s3.md @@ -157,6 +157,12 @@ The legacy configuration key `s3.path.style.access` is still supported as a fall The Native S3 FileSystem is a pure-Java implementation built on the AWS SDK v2 completely removing the dependency on Hadoop. It is registered under the schemes *s3://* and *s3a://*. It provides a drop-in replacement for the Presto and Hadoop implementations, supporting checkpointing, the [FileSink]({{< ref "docs/connectors/datastream/filesystem" >}}) (via `RecoverableWriter`), server-side encryption (SSE-S3, SSE-KMS), cross-account access via IAM role assumption, entropy injection, and bulk copy via S3TransferManager. +#### Cleaning Up Unfinished Uploads + +Flink keeps unfinished S3 uploads that may be needed to restore a job from a checkpoint or savepoint. These uploads can remain after a job stops if it is never restored. + +To clean up unused uploads, configure an S3 lifecycle rule for incomplete multipart uploads. Choose a retention period long enough for uploads to finish and for jobs to recover, including any planned downtime. S3 measures this period from when an upload starts. Cleaning up uploads too soon can prevent recovery from older checkpoints or savepoints. This rule does not remove other temporary files. See the [S3-specific FileSink guidance]({{< ref "docs/connectors/datastream/filesystem" >}}#s3-specific). + #### Setup To use the Native S3 FileSystem, copy the JAR file from the `opt` directory to the `plugins` directory: diff --git a/docs/content/docs/deployment/filesystems/s3.md b/docs/content/docs/deployment/filesystems/s3.md index 36c2b55bad31e3..82e48c173f8232 100644 --- a/docs/content/docs/deployment/filesystems/s3.md +++ b/docs/content/docs/deployment/filesystems/s3.md @@ -157,6 +157,12 @@ The legacy configuration key `s3.path.style.access` is still supported as a fall The Native S3 FileSystem is a pure-Java implementation built on the AWS SDK v2 completely removing the dependency on Hadoop. It is registered under the schemes *s3://* and *s3a://*. It provides a drop-in replacement for the Presto and Hadoop implementations, supporting checkpointing, the [FileSink]({{< ref "docs/connectors/datastream/filesystem" >}}) (via `RecoverableWriter`), server-side encryption (SSE-S3, SSE-KMS), cross-account access via IAM role assumption, entropy injection, and bulk copy via S3TransferManager. +#### Cleaning Up Unfinished Uploads + +Flink keeps unfinished S3 uploads that may be needed to restore a job from a checkpoint or savepoint. These uploads can remain after a job stops if it is never restored. + +To clean up unused uploads, configure an S3 lifecycle rule for incomplete multipart uploads. Choose a retention period long enough for uploads to finish and for jobs to recover, including any planned downtime. S3 measures this period from when an upload starts. Cleaning up uploads too soon can prevent recovery from older checkpoints or savepoints. This rule does not remove other temporary files. See the [S3-specific FileSink guidance]({{< ref "docs/connectors/datastream/filesystem" >}}#s3-specific). + #### Setup To use the Native S3 FileSystem, copy the JAR file from the `opt` directory to the `plugins` directory: diff --git a/flink-filesystems/flink-s3-fs-native/README.md b/flink-filesystems/flink-s3-fs-native/README.md index b65583f989d247..3b86cf1396d8fe 100644 --- a/flink-filesystems/flink-s3-fs-native/README.md +++ b/flink-filesystems/flink-s3-fs-native/README.md @@ -6,6 +6,12 @@ This module provides a native S3 filesystem implementation for Apache Flink usin The Native S3 FileSystem is a direct implementation of Flink's FileSystem interface using AWS SDK v2, without Hadoop dependencies. It provides exactly-once semantics for checkpointing and file sinks through S3 multipart uploads. +### Cleaning Up Unfinished Uploads + +Flink keeps unfinished S3 uploads that may be needed to restore a job from a checkpoint or savepoint. These uploads can remain after a job stops if it is never restored. + +To clean up unused uploads, configure an S3 lifecycle rule for incomplete multipart uploads. Choose a retention period long enough for uploads to finish and for jobs to recover, including any planned downtime. S3 measures this period from when an upload starts. Cleaning up uploads too soon can prevent recovery from older checkpoints or savepoints. This rule does not remove other temporary files. See the [S3-specific FileSink guidance](../../docs/content/docs/connectors/datastream/filesystem.md#s3-specific). + ## Supported URI Schemes This module supports both `s3://` and `s3a://` URI schemes: diff --git a/flink-filesystems/flink-s3-fs-native/src/main/java/org/apache/flink/fs/s3native/writer/NativeS3RecoverableFsDataOutputStream.java b/flink-filesystems/flink-s3-fs-native/src/main/java/org/apache/flink/fs/s3native/writer/NativeS3RecoverableFsDataOutputStream.java index 4afe81c3a9ce8a..3d4f77e8979205 100644 --- a/flink-filesystems/flink-s3-fs-native/src/main/java/org/apache/flink/fs/s3native/writer/NativeS3RecoverableFsDataOutputStream.java +++ b/flink-filesystems/flink-s3-fs-native/src/main/java/org/apache/flink/fs/s3native/writer/NativeS3RecoverableFsDataOutputStream.java @@ -78,6 +78,9 @@ class NativeS3RecoverableFsDataOutputStream extends RecoverableFsDataOutputStrea private volatile boolean closed; + // Recovered uploads and uploads handed out by persist() may belong to retained snapshots. + private boolean uploadMayBeReferenced; + public NativeS3RecoverableFsDataOutputStream( NativeS3ObjectOperations s3AccessHelper, String key, @@ -85,7 +88,16 @@ public NativeS3RecoverableFsDataOutputStream( String localTmpDir, long minPartSize) throws IOException { - this(s3AccessHelper, key, uploadId, localTmpDir, minPartSize, new ArrayList<>(), 0L, null); + this( + s3AccessHelper, + key, + uploadId, + localTmpDir, + minPartSize, + new ArrayList<>(), + 0L, + null, + false); } public NativeS3RecoverableFsDataOutputStream( @@ -98,6 +110,29 @@ public NativeS3RecoverableFsDataOutputStream( long numBytesInParts, File incompleteTailFile) throws IOException { + this( + s3AccessHelper, + key, + uploadId, + localTmpDir, + minPartSize, + existingParts, + numBytesInParts, + incompleteTailFile, + true); + } + + private NativeS3RecoverableFsDataOutputStream( + NativeS3ObjectOperations s3AccessHelper, + String key, + String uploadId, + String localTmpDir, + long minPartSize, + List existingParts, + long numBytesInParts, + File incompleteTailFile, + boolean uploadMayBeReferenced) + throws IOException { this.s3AccessHelper = s3AccessHelper; this.key = key; this.uploadId = uploadId; @@ -108,6 +143,7 @@ public NativeS3RecoverableFsDataOutputStream( this.nextPartNumber = existingParts.size() + 1; this.currentPartSize = 0; this.closed = false; + this.uploadMayBeReferenced = uploadMayBeReferenced; if (incompleteTailFile != null) { resumeFromIncompleteTail(incompleteTailFile); @@ -199,7 +235,8 @@ private void uploadCurrentPart() throws IOException { // Do not delete the temp file if uploadPart fails: propagate the original exception // unmasked and let the cleanup path (close() or the closeForCommit() failure handler) - // delete it and abort the upload. nextPartNumber is only advanced on success. + // delete it and abort the upload if it is not needed for recovery. + // nextPartNumber is only advanced on success. NativeS3ObjectOperations.UploadPartResult result = s3AccessHelper.uploadPart( key, uploadId, nextPartNumber, currentTempFile, currentPartSize); @@ -233,8 +270,8 @@ public Committer closeForCommit() throws IOException { new NativeS3Recoverable( key, uploadId, new ArrayList<>(completedParts), numBytesInParts); } catch (IOException e) { - // The commit failed after the multipart upload had been created and parts may - // already have been uploaded. Abort it so it does not leak as an orphan upload. + // The failed commit may leave uploaded parts behind. Abort the upload to avoid an + // orphan only if it is not needed for recovery. closed = true; try { tryAbortUploadAndReleaseResources(); @@ -267,6 +304,7 @@ public RecoverableWriter.ResumeRecoverable persist() throws IOException { incompletePartLength = currentPartSize; } + uploadMayBeReferenced = true; return new NativeS3Recoverable( key, uploadId, @@ -292,7 +330,9 @@ public void close() throws IOException { } } - /** Aborts the multipart upload and releases local resources on the best effort basis. */ + /** + * Releases local resources and aborts uploads that cannot be referenced by recoverable state. + */ private void tryAbortUploadAndReleaseResources() throws IOException { IOException collected = null; if (currentOutputStream != null) { @@ -309,16 +349,18 @@ private void tryAbortUploadAndReleaseResources() throws IOException { collected = ExceptionUtils.firstOrSuppressed(e, collected); } } - try { - s3AccessHelper.abortMultiPartUpload(key, uploadId); - } catch (IOException e) { - LOG.warn( - "Failed to abort multipart upload (key={}, uploadId={}); it may be left as an " - + "orphan upload in S3. Propagating the failure to the caller.", - key, - uploadId, - e); - collected = ExceptionUtils.firstOrSuppressed(e, collected); + if (!uploadMayBeReferenced) { + try { + s3AccessHelper.abortMultiPartUpload(key, uploadId); + } catch (IOException e) { + LOG.warn( + "Failed to abort multipart upload (key={}, uploadId={}); it may be left as an " + + "orphan upload in S3. Propagating the failure to the caller.", + key, + uploadId, + e); + collected = ExceptionUtils.firstOrSuppressed(e, collected); + } } if (collected != null) { throw collected; diff --git a/flink-filesystems/flink-s3-fs-native/src/test/java/org/apache/flink/fs/s3native/writer/NativeS3RecoverableFsDataOutputStreamTest.java b/flink-filesystems/flink-s3-fs-native/src/test/java/org/apache/flink/fs/s3native/writer/NativeS3RecoverableFsDataOutputStreamTest.java index 99a6824b9af3a5..7f6e9ee1cdd248 100644 --- a/flink-filesystems/flink-s3-fs-native/src/test/java/org/apache/flink/fs/s3native/writer/NativeS3RecoverableFsDataOutputStreamTest.java +++ b/flink-filesystems/flink-s3-fs-native/src/test/java/org/apache/flink/fs/s3native/writer/NativeS3RecoverableFsDataOutputStreamTest.java @@ -19,6 +19,8 @@ package org.apache.flink.fs.s3native.writer; import org.apache.flink.core.fs.RecoverableFsDataOutputStream; +import org.apache.flink.core.fs.RecoverableWriter; +import org.apache.flink.core.testutils.CheckedThread; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; @@ -29,14 +31,22 @@ import java.io.IOException; import java.nio.file.Files; import java.nio.file.Path; +import java.time.Duration; import java.util.ArrayList; import java.util.Arrays; import java.util.Collections; import java.util.HashMap; import java.util.List; import java.util.Map; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; +import static java.nio.charset.StandardCharsets.UTF_8; +import static org.apache.flink.core.testutils.CommonTestUtils.waitUtil; import static org.assertj.core.api.Assertions.assertThat; import static org.assertj.core.api.Assertions.assertThatThrownBy; @@ -68,9 +78,7 @@ void closeForCommitAbortsMultipartUploadWhenPartUploadFails() throws Exception { .isInstanceOf(IOException.class) .hasMessageContaining("injected uploadPart failure"); - assertThat(s3.abortAttempts) - .as("closeForCommit must abort the upload on failure") - .isEqualTo(1); + assertThat(s3.abortAttempts).as("closeForCommit must abort the upload on failure").isOne(); assertThat(s3.openMultipartUploads) .as("the multipart upload must not leak after a failed commit") .doesNotContainKey(uploadId); @@ -95,7 +103,7 @@ void closeForCommitSurfacesAbortFailureWhenBothUploadAndAbortFail() throws Excep .hasMessageContaining( "injected abort failure"))); - assertThat(s3.abortAttempts).isEqualTo(1); + assertThat(s3.abortAttempts).isOne(); } @Test @@ -106,7 +114,7 @@ void closeSurfacesAbortFailureInsteadOfSwallowingIt() throws Exception { .isInstanceOf(IOException.class) .hasMessageContaining("injected abort failure"); - assertThat(s3.abortAttempts).isEqualTo(1); + assertThat(s3.abortAttempts).isOne(); assertThat(countLocalFilesIn(tmp)) .as("local resources are still released even when the abort fails") .isZero(); @@ -117,7 +125,7 @@ void closeSurfacesAbortFailureInsteadOfSwallowingIt() throws Exception { void closeAbortsMultipartUploadOnAbnormalClose() throws Exception { stream.close(); - assertThat(s3.abortAttempts).isEqualTo(1); + assertThat(s3.abortAttempts).isOne(); assertThat(s3.openMultipartUploads).doesNotContainKey(uploadId); assertThat(countLocalFilesIn(tmp)).isZero(); } @@ -131,9 +139,7 @@ void closeSurfacesTempFileDeletionFailure() throws Exception { .isInstanceOf(IOException.class) .hasMessageContaining("injected temp-file delete failure"); - assertThat(s3.abortAttempts) - .as("abort is still attempted despite delete failure") - .isEqualTo(1); + assertThat(s3.abortAttempts).as("abort is still attempted despite delete failure").isOne(); } @Test @@ -261,10 +267,137 @@ protected void deleteTempFile(File file) throws IOException { racingStream.close(); - assertThat(s3.abortAttempts).isEqualTo(1); + assertThat(s3.abortAttempts).isOne(); assertThat(countLocalFilesIn(dir)).isZero(); } + @Test + void closeAfterFirstPersistFailureAbortsUpload() throws Exception { + s3.failPutObject = true; + assertThatThrownBy(stream::persist) + .isInstanceOf(IOException.class) + .hasMessageContaining("injected putObject failure"); + + stream.close(); + + assertThat(s3.abortAttempts).isOne(); + assertThat(s3.openMultipartUploads).doesNotContainKey(uploadId); + assertThat(countLocalFilesIn(tmp)).isZero(); + } + + @Test + void recoverAfterLaterPersistFailure() throws Exception { + final RecoverableWriter.ResumeRecoverable recoverable = stream.persist(); + stream.write(bytes('X', 2)); + s3.failPutObject = true; + assertThatThrownBy(stream::persist) + .isInstanceOf(IOException.class) + .hasMessageContaining("injected putObject failure"); + + stream.close(); + + assertThat(countLocalFilesIn(tmp)).isZero(); + assertThat(s3.abortAttempts).isZero(); + try (RecoverableFsDataOutputStream recovered = writer(s3).recover(recoverable)) { + recovered.write(bytes('C', 2)); + recovered.closeForCommit().commit(); + } + assertThat(s3.committedObjects.get(KEY)).containsExactly("AAAAACC".getBytes(UTF_8)); + } + + @Test + void recoverEmptyStreamAfterDisposingRecoveredStream() throws Exception { + stream.close(); + final FakeNativeS3Operations emptyOperations = new FakeNativeS3Operations(); + final RecoverableWriter.ResumeRecoverable recoverable; + try (RecoverableFsDataOutputStream emptyStream = + newStream(emptyOperations, emptyOperations.startMultiPartUpload(KEY))) { + recoverable = emptyStream.persist(); + } + + writer(emptyOperations).recover(recoverable).close(); + + assertThat(countLocalFilesIn(tmp)).isZero(); + assertThat(emptyOperations.abortAttempts).isZero(); + try (RecoverableFsDataOutputStream recovered = + writer(emptyOperations).recover(recoverable)) { + recovered.write(bytes('C', 2)); + recovered.closeForCommit().commit(); + } + assertThat(emptyOperations.committedObjects.get(KEY)).containsExactly(bytes('C', 2)); + } + + @Test + void closeWhilePersistingPreservesRecoverableState() throws Exception { + stream.close(); + final CountDownLatch uploadingTail = new CountDownLatch(1); + final CountDownLatch finishUpload = new CountDownLatch(1); + final FakeNativeS3Operations blockingOperations = + new FakeNativeS3Operations() { + @Override + public PutObjectResult putObject(String key, File file) throws IOException { + uploadingTail.countDown(); + try { + assertThat(finishUpload.await(10, TimeUnit.SECONDS)).isTrue(); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new IOException(e); + } + return super.putObject(key, file); + } + }; + final NativeS3RecoverableFsDataOutputStream concurrentStream = + newStream(blockingOperations, blockingOperations.startMultiPartUpload(KEY)); + concurrentStream.write(bytes('A', 5)); + final ExecutorService executor = Executors.newSingleThreadExecutor(); + final CheckedThread closingThread = + new CheckedThread("close-persisted-s3-stream") { + @Override + public void go() throws Exception { + concurrentStream.close(); + } + }; + try { + final Future persisted = + executor.submit(concurrentStream::persist); + assertThat(uploadingTail.await(10, TimeUnit.SECONDS)).isTrue(); + closingThread.start(); + waitUtil( + () -> + closingThread.getState() == Thread.State.WAITING + || !closingThread.isAlive(), + Duration.ofSeconds(10), + "close() did not wait for persist()"); + assertThat(closingThread.getState()).isEqualTo(Thread.State.WAITING); + finishUpload.countDown(); + final RecoverableWriter.ResumeRecoverable recoverable = + persisted.get(10, TimeUnit.SECONDS); + closingThread.sync(10_000); + assertThat(countLocalFilesIn(tmp)).isZero(); + + assertThat(blockingOperations.abortAttempts).isZero(); + try (RecoverableFsDataOutputStream recovered = + writer(blockingOperations).recover(recoverable)) { + recovered.write(bytes('C', 2)); + recovered.closeForCommit().commit(); + } + assertThat(blockingOperations.committedObjects.get(KEY)) + .containsExactly("AAAAACC".getBytes(UTF_8)); + } finally { + finishUpload.countDown(); + executor.shutdownNow(); + assertThat(executor.awaitTermination(10, TimeUnit.SECONDS)).isTrue(); + if (closingThread.getState() != Thread.State.NEW) { + closingThread.sync(10_000); + } + concurrentStream.close(); + } + } + + private NativeS3RecoverableWriter writer(FakeNativeS3Operations operations) { + return NativeS3RecoverableWriter.writer(operations, tmp.toString(), MIN_PART_SIZE, 1); + } + private NativeS3RecoverableFsDataOutputStream newStream(FakeNativeS3Operations ops, String uid) throws IOException { return new NativeS3RecoverableFsDataOutputStream( @@ -308,12 +441,13 @@ private static byte[] bytes(char c, int n) { * failures, so this test can exercise {@link NativeS3RecoverableFsDataOutputStream}'s failure * handling without an S3 endpoint. */ - private static final class FakeNativeS3Operations extends NativeS3ObjectOperations { + private static class FakeNativeS3Operations extends NativeS3ObjectOperations { final Map committedObjects = new HashMap<>(); final Map> openMultipartUploads = new HashMap<>(); boolean failUploadPart = false; + boolean failPutObject = false; boolean failAbortMultiPartUpload = false; boolean deletePartFileAfterUpload = false; int abortAttempts = 0; @@ -398,5 +532,24 @@ public void abortMultiPartUpload(String key, String uploadId) throws IOException } openMultipartUploads.remove(uploadId); } + + @Override + public PutObjectResult putObject(String key, File inputFile) throws IOException { + if (failPutObject) { + throw new IOException("injected putObject failure for key: " + key); + } + committedObjects.put(key, Files.readAllBytes(inputFile.toPath())); + return new PutObjectResult("etag-" + key); + } + + @Override + public long getObject(String key, File target) throws IOException { + final byte[] data = committedObjects.get(key); + if (data == null) { + throw new IOException("missing object: " + key); + } + Files.write(target.toPath(), data); + return data.length; + } } } diff --git a/flink-filesystems/flink-s3-fs-native/src/test/java/org/apache/flink/fs/s3native/writer/NativeS3RecoverableWriterRecoveryITCase.java b/flink-filesystems/flink-s3-fs-native/src/test/java/org/apache/flink/fs/s3native/writer/NativeS3RecoverableWriterRecoveryITCase.java index 1c71c1ec710fc0..fa89d9ee03c16c 100644 --- a/flink-filesystems/flink-s3-fs-native/src/test/java/org/apache/flink/fs/s3native/writer/NativeS3RecoverableWriterRecoveryITCase.java +++ b/flink-filesystems/flink-s3-fs-native/src/test/java/org/apache/flink/fs/s3native/writer/NativeS3RecoverableWriterRecoveryITCase.java @@ -26,14 +26,20 @@ import org.apache.flink.fs.s3native.SeaweedFsNativeS3TestContainer; import org.apache.commons.lang3.ArrayUtils; +import org.junit.jupiter.api.AfterEach; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; import org.junit.jupiter.api.extension.RegisterExtension; import org.junit.jupiter.api.io.TempDir; +import software.amazon.awssdk.services.s3.model.MultipartUpload; +import java.io.File; import java.io.IOException; +import java.util.ArrayList; import java.util.Arrays; +import java.util.List; import java.util.UUID; +import java.util.concurrent.atomic.AtomicBoolean; import static org.assertj.core.api.Assertions.assertThat; import static org.assertj.core.api.Assertions.assertThatThrownBy; @@ -77,6 +83,7 @@ class NativeS3RecoverableWriterRecoveryITCase { private String bucket; private String key; private SeaweedFsNativeS3Operations s3; + private final List recoverables = new ArrayList<>(); @BeforeEach void setUp() { @@ -90,7 +97,28 @@ private static SeaweedFsNativeS3TestContainer getContainer() { } private NativeS3RecoverableWriter writer() { - return NativeS3RecoverableWriter.writer(s3, tmp.toString(), MIN_PART_SIZE, 1); + return writer(s3); + } + + private NativeS3RecoverableWriter writer(NativeS3ObjectOperations operations) { + return NativeS3RecoverableWriter.writer(operations, tmp.toString(), MIN_PART_SIZE, 1); + } + + @AfterEach + void cleanUpRemoteState() throws IOException { + for (MultipartUpload upload : + getContainer() + .getClient() + .listMultipartUploads(request -> request.bucket(bucket)) + .uploads()) { + if (key.equals(upload.key())) { + s3.abortMultiPartUpload(key, upload.uploadId()); + } + } + for (NativeS3Recoverable recoverable : recoverables) { + writer().cleanupRecoverableState(recoverable); + } + s3.removeObject(key); } private Path targetPath() { @@ -109,9 +137,14 @@ void recoverWithoutIncompleteTailStillWorks() throws Exception { final RecoverableFsDataOutputStream out = writer1.open(targetPath()); out.write(bytes('A', PART), 0, PART); final NativeS3Recoverable r = (NativeS3Recoverable) out.persist(); + recoverables.add(r); assertThat(r.incompleteObjectName()).as("no tail => no side object").isNull(); assertThat(s3.listKeys(incompletePrefix(r.uploadId()))).isEmpty(); + out.close(); + assertThat(countLocalFilesIn(tmp)).isZero(); + assertUploadAvailable(r); + final NativeS3RecoverableWriter writer2 = writer(); final RecoverableFsDataOutputStream resumed = writer2.recover(r); resumed.write(bytes('C', 10), 0, 10); @@ -130,10 +163,15 @@ void recoverWithNestedKeyStillWorks() throws Exception { out.write(bytes('A', PART), 0, PART); out.write(bytes('E', 5), 0, 5); final NativeS3Recoverable r = (NativeS3Recoverable) out.persist(); + recoverables.add(r); assertThat(r.incompleteObjectName()).as("tail written => side object expected").isNotNull(); assertThat(s3.listKeys(incompletePrefix(r.uploadId()))) .containsExactly(r.incompleteObjectName()); + out.close(); + assertThat(countLocalFilesIn(tmp)).isZero(); + assertUploadAvailable(r); + final NativeS3RecoverableWriter writer2 = writer(); final RecoverableFsDataOutputStream resumed = writer2.recover(r); resumed.write(bytes('C', 10), 0, 10); @@ -143,6 +181,127 @@ void recoverWithNestedKeyStillWorks() throws Exception { s3.readObject(key), concat(bytes('A', PART), bytes('E', 5), bytes('C', 10))); } + @Test + void recoverAfterDisposingRecoveredStream() throws Exception { + final NativeS3Recoverable recoverable; + try (RecoverableFsDataOutputStream out = writer().open(targetPath())) { + out.write(bytes('A', PART)); + out.write(bytes('B', 3)); + recoverable = persistAndTrack(out); + } + + writer().recover(recoverable).close(); + + assertThat(countLocalFilesIn(tmp)).isZero(); + assertUploadAvailable(recoverable); + try (RecoverableFsDataOutputStream recovered = writer().recover(recoverable)) { + recovered.write(bytes('C', 10)); + recovered.closeForCommit().commit(); + } + assertContentEquals( + s3.readObject(key), concat(bytes('A', PART), bytes('B', 3), bytes('C', 10))); + } + + @Test + void unpersistedUploadIsAbortedOnCloseForCommitFailure() throws Exception { + final AtomicBoolean failUpload = new AtomicBoolean(); + try (RecoverableFsDataOutputStream out = + writer(failingUploadOperations(failUpload)).open(targetPath())) { + out.write(bytes('A', PART)); + final MultipartUpload upload = + getContainer() + .getClient() + .listMultipartUploads(request -> request.bucket(bucket)) + .uploads() + .stream() + .filter(candidate -> key.equals(candidate.key())) + .findFirst() + .orElseThrow( + () -> + new AssertionError( + "The upload must exist before cleanup")); + assertUploadAvailable(upload.uploadId()); + out.write(bytes('B', 7)); + + failUpload.set(true); + assertThatThrownBy(out::closeForCommit) + .isInstanceOf(IOException.class) + .hasMessageContaining("injected final-part upload failure"); + + assertThat( + getContainer() + .getClient() + .listMultipartUploads(request -> request.bucket(bucket)) + .uploads()) + .extracting(MultipartUpload::key) + .doesNotContain(key); + assertThat(countLocalFilesIn(tmp)).isZero(); + } + } + + @Test + void recoverAfterCloseForCommitFailure() throws Exception { + final AtomicBoolean failUpload = new AtomicBoolean(); + final NativeS3Recoverable recoverable; + try (RecoverableFsDataOutputStream out = + writer(failingUploadOperations(failUpload)).open(targetPath())) { + out.write(bytes('A', PART)); + out.write(bytes('B', 3)); + recoverable = persistAndTrack(out); + out.write(bytes('X', 7)); + failUpload.set(true); + + assertThatThrownBy(out::closeForCommit) + .isInstanceOf(IOException.class) + .hasMessageContaining("injected final-part upload failure"); + } + + assertThat(countLocalFilesIn(tmp)).isZero(); + assertUploadAvailable(recoverable); + try (RecoverableFsDataOutputStream recovered = writer().recover(recoverable)) { + recovered.write(bytes('C', 10)); + recovered.closeForCommit().commit(); + } + assertContentEquals( + s3.readObject(key), concat(bytes('A', PART), bytes('B', 3), bytes('C', 10))); + } + + private NativeS3ObjectOperations failingUploadOperations(AtomicBoolean failUpload) { + return new NativeS3ObjectOperations(getContainer().getClient(), bucket) { + @Override + public UploadPartResult uploadPart( + String objectKey, String uploadId, int partNumber, File inputFile, long length) + throws IOException { + if (failUpload.get()) { + throw new IOException("injected final-part upload failure"); + } + return super.uploadPart(objectKey, uploadId, partNumber, inputFile, length); + } + }; + } + + private NativeS3Recoverable persistAndTrack(RecoverableFsDataOutputStream out) + throws IOException { + final NativeS3Recoverable recoverable = (NativeS3Recoverable) out.persist(); + recoverables.add(recoverable); + return recoverable; + } + + private void assertUploadAvailable(NativeS3Recoverable recoverable) { + assertUploadAvailable(recoverable.uploadId()); + } + + private void assertUploadAvailable(String uploadId) { + assertThat( + getContainer() + .getClient() + .listParts( + request -> + request.bucket(bucket).key(key).uploadId(uploadId)) + .parts()) + .hasSize(1); + } + @Test void recoverFailsCleanlyWhenSideObjectMissing() throws Exception { final NativeS3Recoverable r = persistWithTail(); @@ -174,7 +333,7 @@ private NativeS3Recoverable persistWithTail() throws IOException { final RecoverableFsDataOutputStream out = writer1.open(targetPath()); out.write(bytes('A', PART), 0, PART); out.write(bytes('E', 5), 0, 5); - return (NativeS3Recoverable) out.persist(); + return persistAndTrack(out); } /**