This is an automated email from the ASF dual-hosted git repository.
Samrat002 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new 916e57972a3 [FLINK-40953][s3] Store the native S3 recoverable writer
tail under the target directory
916e57972a3 is described below
commit 916e57972a3867041f01b143b54e0729915e3f87
Author: Gabor Somogyi <[email protected]>
AuthorDate: Tue Oct 20 18:35:38 2026 +0200
[FLINK-40953][s3] Store the native S3 recoverable writer tail under the
target directory
* [FLINK-40953][s3] Store the native S3 recoverable writer tail under the
target directory
Generated-by: Claude Code (Sonnet 5.5)
* [FLINK-40953][s3] Test restoring native S3 tails written by the 2.3
release
The 2.3 release stores the tail of an in-progress file under the target
key. Cover that layout when recovering and cleaning up, and add a test
that restores from such a checkpoint, checkpoints again into the new
location, restores from the old checkpoint a second time and then from
the new one.
Generated-by: Claude Code (Sonnet 5.5)
---
.../NativeS3RecoverableFsDataOutputStream.java | 5 +-
.../fs/s3native/writer/FakeNativeS3Operations.java | 145 ++++++++++++++
.../NativeS3RecoverableFsDataOutputStreamTest.java | 100 ----------
.../writer/NativeS3RecoverableSerializerTest.java | 38 ++++
.../NativeS3RecoverableWriterRecoveryITCase.java | 5 +-
.../writer/NativeS3RecoverableWriterTest.java | 210 +++++++++++++++++++++
6 files changed, 400 insertions(+), 103 deletions(-)
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 79084c23eab..6fc6ff83043 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
@@ -277,7 +277,10 @@ class NativeS3RecoverableFsDataOutputStream extends
RecoverableFsDataOutputStrea
if (currentPartSize > 0) {
currentOutputStream.flush();
- incompletePartKey = ".incomplete/" + uploadId + "/" +
UUID.randomUUID();
+ // Keep the side object in the target's directory, so that
anything scoped to the
+ // target's key prefix (for example IAM or session policies)
covers it as well.
+ final String parent = key.substring(0, key.lastIndexOf('/') +
1);
+ incompletePartKey = parent + ".incomplete/" + uploadId + "/" +
UUID.randomUUID();
s3AccessHelper.putObject(incompletePartKey, currentTempFile);
incompletePartLength = currentPartSize;
}
diff --git
a/flink-filesystems/flink-s3-fs-native/src/test/java/org/apache/flink/fs/s3native/writer/FakeNativeS3Operations.java
b/flink-filesystems/flink-s3-fs-native/src/test/java/org/apache/flink/fs/s3native/writer/FakeNativeS3Operations.java
new file mode 100644
index 00000000000..c8bad44d0ca
--- /dev/null
+++
b/flink-filesystems/flink-s3-fs-native/src/test/java/org/apache/flink/fs/s3native/writer/FakeNativeS3Operations.java
@@ -0,0 +1,145 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.fs.s3native.writer;
+
+import java.io.ByteArrayOutputStream;
+import java.io.File;
+import java.io.IOException;
+import java.nio.file.Files;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicInteger;
+
+/**
+ * In-memory {@link NativeS3ObjectOperations} fake for the recoverable writer
tests. It keeps
+ * multipart uploads, completed objects and standalone objects in memory and
has hooks to inject
+ * part-upload and abort failures, so {@link NativeS3RecoverableWriter} and
{@link
+ * NativeS3RecoverableFsDataOutputStream} can be tested without an S3 endpoint.
+ */
+final class FakeNativeS3Operations extends NativeS3ObjectOperations {
+
+ final Map<String, byte[]> committedObjects = new HashMap<>();
+ final Map<String, Map<Integer, byte[]>> openMultipartUploads = new
HashMap<>();
+
+ /** Standalone objects written with {@link #putObject}, such as the
persisted tail objects. */
+ final Map<String, byte[]> objects = new HashMap<>();
+
+ boolean failUploadPart = false;
+ boolean deletePartFileAfterUpload = false;
+ int abortAttempts = 0;
+ int uploadPartAttempts = 0;
+
+ private final AtomicInteger uploadIdSeq = new AtomicInteger();
+
+ FakeNativeS3Operations() {
+ super(/* s3Client */ null, /* transferManager */ null, "test-bucket",
false);
+ }
+
+ @Override
+ public String startMultiPartUpload(String key) {
+ String uploadId = "U" + uploadIdSeq.incrementAndGet();
+ openMultipartUploads.put(uploadId, new HashMap<>());
+ return uploadId;
+ }
+
+ @Override
+ public UploadPartResult uploadPart(
+ String key, String uploadId, int partNumber, File file, long
length)
+ throws IOException {
+ uploadPartAttempts++;
+ if (failUploadPart) {
+ throw new IOException("injected uploadPart failure for uploadId: "
+ uploadId);
+ }
+ Map<Integer, byte[]> parts = openMultipartUploads.get(uploadId);
+ if (parts == null) {
+ throw new IOException("unknown uploadId: " + uploadId);
+ }
+ byte[] data = Files.readAllBytes(file.toPath());
+ if (data.length != length) {
+ throw new IOException(
+ "part length mismatch: expected " + length + ", got " +
data.length);
+ }
+ parts.put(partNumber, data);
+ if (deletePartFileAfterUpload) {
+ Files.delete(file.toPath());
+ }
+ return new UploadPartResult(partNumber, "etag-" + uploadId + "-" +
partNumber);
+ }
+
+ @Override
+ public CompleteMultipartUploadResult commitMultiPartUpload(
+ String key, String uploadId, List<UploadPartResult> parts, long
length)
+ throws IOException {
+ Map<Integer, byte[]> uploaded = openMultipartUploads.remove(uploadId);
+ if (uploaded == null) {
+ throw new IOException("unknown uploadId: " + uploadId);
+ }
+ List<Integer> ordered = new ArrayList<>(parts.size());
+ for (UploadPartResult p : parts) {
+ ordered.add(p.getPartNumber());
+ }
+ Collections.sort(ordered);
+ ByteArrayOutputStream merged = new ByteArrayOutputStream();
+ for (int n : ordered) {
+ byte[] partData = uploaded.get(n);
+ if (partData == null) {
+ throw new IOException("missing part " + n + " for uploadId " +
uploadId);
+ }
+ merged.write(partData);
+ }
+ byte[] finalBytes = merged.toByteArray();
+ if (finalBytes.length != length) {
+ throw new IOException(
+ "committed length mismatch: expected " + length + ", got "
+ finalBytes.length);
+ }
+ committedObjects.put(key, finalBytes);
+ return new CompleteMultipartUploadResult(
+ "test-bucket", key, "final-etag-" + uploadId, null);
+ }
+
+ @Override
+ public void abortMultiPartUpload(String key, String uploadId) throws
IOException {
+ abortAttempts++;
+ openMultipartUploads.remove(uploadId);
+ }
+
+ @Override
+ public PutObjectResult putObject(String key, File inputFile) throws
IOException {
+ objects.put(key, Files.readAllBytes(inputFile.toPath()));
+ return new PutObjectResult("etag-" + key);
+ }
+
+ @Override
+ public long getObject(String key, File targetLocation) throws IOException {
+ byte[] data = objects.get(key);
+ if (data == null) {
+ throw new IOException("Failed to get object for key: " + key);
+ }
+ Files.write(targetLocation.toPath(), data);
+ return data.length;
+ }
+
+ @Override
+ public boolean deleteObject(String key) {
+ return objects.remove(key) != null;
+ }
+}
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 c2d63ac9c32..3db9cabcaee 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
@@ -24,18 +24,11 @@ import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.io.TempDir;
-import java.io.ByteArrayOutputStream;
import java.io.File;
import java.io.IOException;
import java.nio.file.Files;
import java.nio.file.Path;
-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.atomic.AtomicInteger;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
@@ -260,97 +253,4 @@ class NativeS3RecoverableFsDataOutputStreamTest {
Arrays.fill(b, (byte) c);
return b;
}
-
- /**
- * In-memory {@link NativeS3ObjectOperations} fake with hooks to inject
part-upload and abort
- * failures, so this test can exercise {@link
NativeS3RecoverableFsDataOutputStream}'s failure
- * handling without an S3 endpoint.
- */
- private static final class FakeNativeS3Operations extends
NativeS3ObjectOperations {
-
- final Map<String, byte[]> committedObjects = new HashMap<>();
- final Map<String, Map<Integer, byte[]>> openMultipartUploads = new
HashMap<>();
-
- boolean failUploadPart = false;
- boolean deletePartFileAfterUpload = false;
- int abortAttempts = 0;
- int uploadPartAttempts = 0;
-
- private final AtomicInteger uploadIdSeq = new AtomicInteger();
-
- FakeNativeS3Operations() {
- super(/* s3Client */ null, /* transferManager */ null,
"test-bucket", false);
- }
-
- @Override
- public String startMultiPartUpload(String key) {
- String uploadId = "U" + uploadIdSeq.incrementAndGet();
- openMultipartUploads.put(uploadId, new HashMap<>());
- return uploadId;
- }
-
- @Override
- public UploadPartResult uploadPart(
- String key, String uploadId, int partNumber, File file, long
length)
- throws IOException {
- uploadPartAttempts++;
- if (failUploadPart) {
- throw new IOException("injected uploadPart failure for
uploadId: " + uploadId);
- }
- Map<Integer, byte[]> parts = openMultipartUploads.get(uploadId);
- if (parts == null) {
- throw new IOException("unknown uploadId: " + uploadId);
- }
- byte[] data = Files.readAllBytes(file.toPath());
- if (data.length != length) {
- throw new IOException(
- "part length mismatch: expected " + length + ", got "
+ data.length);
- }
- parts.put(partNumber, data);
- if (deletePartFileAfterUpload) {
- Files.delete(file.toPath());
- }
- return new UploadPartResult(partNumber, "etag-" + uploadId + "-" +
partNumber);
- }
-
- @Override
- public CompleteMultipartUploadResult commitMultiPartUpload(
- String key, String uploadId, List<UploadPartResult> parts,
long length)
- throws IOException {
- Map<Integer, byte[]> uploaded =
openMultipartUploads.remove(uploadId);
- if (uploaded == null) {
- throw new IOException("unknown uploadId: " + uploadId);
- }
- List<Integer> ordered = new ArrayList<>(parts.size());
- for (UploadPartResult p : parts) {
- ordered.add(p.getPartNumber());
- }
- Collections.sort(ordered);
- ByteArrayOutputStream merged = new ByteArrayOutputStream();
- for (int n : ordered) {
- byte[] partData = uploaded.get(n);
- if (partData == null) {
- throw new IOException("missing part " + n + " for uploadId
" + uploadId);
- }
- merged.write(partData);
- }
- byte[] finalBytes = merged.toByteArray();
- if (finalBytes.length != length) {
- throw new IOException(
- "committed length mismatch: expected "
- + length
- + ", got "
- + finalBytes.length);
- }
- committedObjects.put(key, finalBytes);
- return new CompleteMultipartUploadResult(
- "test-bucket", key, "final-etag-" + uploadId, null);
- }
-
- @Override
- public void abortMultiPartUpload(String key, String uploadId) throws
IOException {
- abortAttempts++;
- openMultipartUploads.remove(uploadId);
- }
- }
}
diff --git
a/flink-filesystems/flink-s3-fs-native/src/test/java/org/apache/flink/fs/s3native/writer/NativeS3RecoverableSerializerTest.java
b/flink-filesystems/flink-s3-fs-native/src/test/java/org/apache/flink/fs/s3native/writer/NativeS3RecoverableSerializerTest.java
index 4c5ea143a55..c9b38954f70 100644
---
a/flink-filesystems/flink-s3-fs-native/src/test/java/org/apache/flink/fs/s3native/writer/NativeS3RecoverableSerializerTest.java
+++
b/flink-filesystems/flink-s3-fs-native/src/test/java/org/apache/flink/fs/s3native/writer/NativeS3RecoverableSerializerTest.java
@@ -20,6 +20,8 @@ package org.apache.flink.fs.s3native.writer;
import org.junit.jupiter.api.Test;
+import java.io.ByteArrayOutputStream;
+import java.io.DataOutputStream;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
@@ -110,4 +112,40 @@ class NativeS3RecoverableSerializerTest {
NativeS3RecoverableSerializer serializer =
NativeS3RecoverableSerializer.INSTANCE;
assertThat(serializer.getVersion()).isGreaterThanOrEqualTo(1);
}
+
+ /**
+ * Checkpoints and savepoints hold the tail key as an opaque string, so
state written earlier
+ * must stay readable whatever the key looks like, and the bytes written
must not change.
+ */
+ @Test
+ void testVersion1BytesWithTailKeyAtBucketRoot() throws IOException {
+ NativeS3RecoverableSerializer serializer =
NativeS3RecoverableSerializer.INSTANCE;
+ String tailKey =
".incomplete/upload-id/0f8fad5b-d9cb-469f-a165-70867728950e";
+
+ ByteArrayOutputStream bytes = new ByteArrayOutputStream();
+ DataOutputStream out = new DataOutputStream(bytes);
+ out.writeUTF("dir/out.txt");
+ out.writeUTF("upload-id");
+ out.writeLong(5242880L);
+ out.writeInt(1);
+ out.writeInt(1);
+ out.writeUTF("etag1");
+ out.writeBoolean(true);
+ out.writeUTF(tailKey);
+ out.writeLong(1024L);
+ out.flush();
+ byte[] version1 = bytes.toByteArray();
+
+ NativeS3Recoverable deserialized = serializer.deserialize(1, version1);
+
+ assertThat(deserialized.getObjectName()).isEqualTo("dir/out.txt");
+ assertThat(deserialized.uploadId()).isEqualTo("upload-id");
+ assertThat(deserialized.numBytesInParts()).isEqualTo(5242880L);
+ assertThat(deserialized.parts()).hasSize(1);
+ assertThat(deserialized.parts().get(0).getPartNumber()).isEqualTo(1);
+ assertThat(deserialized.parts().get(0).getETag()).isEqualTo("etag1");
+ assertThat(deserialized.incompleteObjectName()).isEqualTo(tailKey);
+ assertThat(deserialized.incompleteObjectLength()).isEqualTo(1024L);
+ assertThat(serializer.serialize(deserialized)).isEqualTo(version1);
+ }
}
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 1c71c1ec710..2fbbcccc0a1 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
@@ -52,7 +52,8 @@ import static
org.assertj.core.api.Assertions.assertThatThrownBy;
* +-- part 1: PART bytes, uploaded as a completed multipart upload part
* +-- tail: any bytes written after part 1, not yet part of a completed
multipart part
*
- * side object ("<key>/.incomplete/<uploadId>/<uuid>", see
#incompletePrefix())
+ * side object ("<dir>/.incomplete/<uploadId>/<uuid>", see
#incompletePrefix())
+ * - lives in the target object's directory ("<dir>/" is empty for a
root-level key)
* - written by persist() only when there IS a tail, so that the tail
bytes survive a
* writer restart
* - read back by recover(), which downloads it locally and appends it to
the in-progress
@@ -98,7 +99,7 @@ class NativeS3RecoverableWriterRecoveryITCase {
}
private String incompletePrefix(String uploadId) {
- return ".incomplete/" + uploadId + "/";
+ return key.substring(0, key.lastIndexOf('/') + 1) + ".incomplete/" +
uploadId + "/";
}
@Test
diff --git
a/flink-filesystems/flink-s3-fs-native/src/test/java/org/apache/flink/fs/s3native/writer/NativeS3RecoverableWriterTest.java
b/flink-filesystems/flink-s3-fs-native/src/test/java/org/apache/flink/fs/s3native/writer/NativeS3RecoverableWriterTest.java
new file mode 100644
index 00000000000..eff726c53e5
--- /dev/null
+++
b/flink-filesystems/flink-s3-fs-native/src/test/java/org/apache/flink/fs/s3native/writer/NativeS3RecoverableWriterTest.java
@@ -0,0 +1,210 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.fs.s3native.writer;
+
+import org.apache.flink.core.fs.Path;
+import org.apache.flink.core.fs.RecoverableFsDataOutputStream;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import java.util.Collections;
+import java.util.regex.Pattern;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * Tests how {@link NativeS3RecoverableWriter} persists, restores and cleans
up the tail of an
+ * in-progress file, i.e. the buffered bytes that are too small to be a
multipart upload part.
+ */
+class NativeS3RecoverableWriterTest {
+
+ private static final long MIN_PART_SIZE = 10L;
+ private static final String KEY = "dir/out.txt";
+ private static final String UUID_REGEX =
+ "[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}";
+
+ /** Where the 2.3 release stores the tail: under the target key. */
+ private static final String TAIL_KEY_OF_RELEASE_2_3 = KEY +
"/.incomplete/U1/tail";
+
+ /** Where the tail is stored now: in the target's directory. */
+ private static final String TAIL_KEY_IN_TARGET_DIRECTORY =
"dir/.incomplete/U1/tail";
+
+ /** Where the tail of a target without a directory is stored now: at the
bucket root. */
+ private static final String TAIL_KEY_OF_ROOT_LEVEL_TARGET =
".incomplete/U1/tail";
+
+ @TempDir java.nio.file.Path tmp;
+
+ private FakeNativeS3Operations s3;
+
+ @BeforeEach
+ void setUp() {
+ s3 = new FakeNativeS3Operations();
+ }
+
+ /** Recovery reads the tail by the key stored in the recoverable, whatever
its layout. */
+ @ParameterizedTest
+ @ValueSource(
+ strings = {
+ TAIL_KEY_OF_RELEASE_2_3,
+ TAIL_KEY_IN_TARGET_DIRECTORY,
+ TAIL_KEY_OF_ROOT_LEVEL_TARGET
+ })
+ void recoverResumesFromTailAtStoredKey(String tailKey) throws Exception {
+ final String uploadId = s3.startMultiPartUpload(KEY);
+ s3.objects.put(tailKey, ascii("AAAAA"));
+ final NativeS3Recoverable recoverable =
+ new NativeS3Recoverable(KEY, uploadId,
Collections.emptyList(), 0L, tailKey, 5L);
+
+ final RecoverableFsDataOutputStream resumed =
writer().recover(recoverable);
+ resumed.write(ascii("BBB"), 0, 3);
+ resumed.closeForCommit().commit();
+
+ assertThat(s3.committedObjects.get(KEY)).isEqualTo(ascii("AAAAABBB"));
+ }
+
+ @ParameterizedTest
+ @ValueSource(
+ strings = {
+ TAIL_KEY_OF_RELEASE_2_3,
+ TAIL_KEY_IN_TARGET_DIRECTORY,
+ TAIL_KEY_OF_ROOT_LEVEL_TARGET
+ })
+ void cleanupDeletesOnlyTailAtStoredKey(String tailKey) throws Exception {
+ s3.objects.put(tailKey, ascii("AAAAA"));
+ s3.objects.put("dir/other", ascii("other"));
+ final NativeS3Recoverable recoverable =
+ new NativeS3Recoverable(KEY, "U1", Collections.emptyList(),
0L, tailKey, 5L);
+
+ assertThat(writer().cleanupRecoverableState(recoverable)).isTrue();
+
+ assertThat(s3.objects).containsOnlyKeys("dir/other");
+ }
+
+ @Test
+ void cleanupWithoutTailDeletesNothing() throws Exception {
+ s3.objects.put("dir/other", ascii("other"));
+ final NativeS3Recoverable recoverable =
+ new NativeS3Recoverable(KEY, "U1", Collections.emptyList(),
0L);
+
+ assertThat(writer().cleanupRecoverableState(recoverable)).isFalse();
+
+ assertThat(s3.objects).containsOnlyKeys("dir/other");
+ }
+
+ @Test
+ void persistWithoutBufferedBytesStoresNoTail() throws Exception {
+ final RecoverableFsDataOutputStream stream =
writer().open(targetPath(KEY));
+ // A full part is uploaded right away, which leaves nothing buffered.
+ stream.write(ascii("AAAAAAAAAA"), 0, 10);
+
+ final NativeS3Recoverable recoverable = (NativeS3Recoverable)
stream.persist();
+ stream.close();
+
+ assertThat(recoverable.incompleteObjectName()).isNull();
+ assertThat(s3.objects).isEmpty();
+ }
+
+ @Test
+ void persistStoresTailUnderTargetDirectory() throws Exception {
+ final NativeS3Recoverable recoverable =
persistFiveBytes("a/b/out.txt");
+
+ assertThat(recoverable.incompleteObjectName())
+ .matches(
+ Pattern.quote("a/b/.incomplete/" +
recoverable.uploadId() + "/")
+ + UUID_REGEX);
+ assertTailStored(recoverable, "AAAAA");
+ }
+
+ @Test
+ void persistStoresTailAtBucketRootForRootLevelTarget() throws Exception {
+ final NativeS3Recoverable recoverable = persistFiveBytes("out.txt");
+
+ assertThat(recoverable.incompleteObjectName())
+ .matches(Pattern.quote(".incomplete/" + recoverable.uploadId()
+ "/") + UUID_REGEX);
+ assertTailStored(recoverable, "AAAAA");
+ }
+
+ /**
+ * A job upgraded from the 2.3 release restores from a tail stored under
the target key, then
+ * checkpoints into the new location. The old checkpoint stays restorable
until it is retired,
+ * retiring it leaves the new tail alone, and the new checkpoint restores
as well.
+ */
+ @Test
+ void upgradeFromRelease23Checkpoint() throws Exception {
+ final String uploadId = s3.startMultiPartUpload(KEY);
+ s3.objects.put(TAIL_KEY_OF_RELEASE_2_3, ascii("AAAAA"));
+ final NativeS3Recoverable oldCheckpoint =
+ new NativeS3Recoverable(
+ KEY, uploadId, Collections.emptyList(), 0L,
TAIL_KEY_OF_RELEASE_2_3, 5L);
+
+ final RecoverableFsDataOutputStream resumed =
writer().recover(oldCheckpoint);
+ resumed.write(ascii("BBB"), 0, 3);
+ final NativeS3Recoverable newCheckpoint = (NativeS3Recoverable)
resumed.persist();
+ resumed.close();
+
+ assertThat(newCheckpoint.incompleteObjectName())
+ .startsWith("dir/.incomplete/" + uploadId + "/");
+
+ // A failure before the new checkpoint completes restores from the old
one again.
+ final RecoverableFsDataOutputStream resumedAgain =
writer().recover(oldCheckpoint);
+ assertThat(resumedAgain.getPos()).isEqualTo(5L);
+ resumedAgain.close();
+
+ assertThat(writer().cleanupRecoverableState(oldCheckpoint)).isTrue();
+
assertThat(s3.objects).containsOnlyKeys(newCheckpoint.incompleteObjectName());
+
+ final RecoverableFsDataOutputStream restored =
writer().recover(newCheckpoint);
+ restored.write(ascii("C"), 0, 1);
+ restored.closeForCommit().commit();
+
+ assertThat(s3.committedObjects.get(KEY)).isEqualTo(ascii("AAAAABBBC"));
+ }
+
+ private NativeS3Recoverable persistFiveBytes(String key) throws
IOException {
+ final RecoverableFsDataOutputStream stream =
writer().open(targetPath(key));
+ stream.write(ascii("AAAAA"), 0, 5);
+ final NativeS3Recoverable recoverable = (NativeS3Recoverable)
stream.persist();
+ stream.close();
+ return recoverable;
+ }
+
+ private void assertTailStored(NativeS3Recoverable recoverable, String
expectedContent) {
+
assertThat(recoverable.incompleteObjectLength()).isEqualTo(expectedContent.length());
+ assertThat(s3.objects.get(recoverable.incompleteObjectName()))
+ .isEqualTo(ascii(expectedContent));
+ }
+
+ private NativeS3RecoverableWriter writer() {
+ return NativeS3RecoverableWriter.writer(s3, tmp.toString(),
MIN_PART_SIZE, 1);
+ }
+
+ private static Path targetPath(String key) {
+ return new Path("s3://test-bucket/" + key);
+ }
+
+ private static byte[] ascii(String value) {
+ return value.getBytes(StandardCharsets.US_ASCII);
+ }
+}
