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 76b7ed3d464 [FLINK-40017] Introduce Seaweedfs in native-s3-fs
76b7ed3d464 is described below
commit 76b7ed3d4648d3e2e1bc40a70f2e597699b6d392
Author: Samrat <[email protected]>
AuthorDate: Tue Sep 1 12:45:31 2026 +0530
[FLINK-40017] Introduce Seaweedfs in native-s3-fs
---
.../NativeS3RecoverableFsDataOutputStream.java | 2 +-
.../AbstractNativeS3HAApplicationRunITCase.java | 102 ++++++
.../s3native/AbstractNativeS3HAJobRunITCase.java | 99 ++++++
...HAApplicationRunOnNativeS3FileSystemITCase.java | 39 +++
.../HAJobRunOnNativeS3FileSystemITCase.java | 39 +++
.../fs/s3native/NativeS3FileSystemITCase.java | 165 ++++++++++
.../SeaweedFsNativeS3HAClusterExtension.java | 125 ++++++++
.../s3native/SeaweedFsNativeS3TestContainer.java | 165 ++++++++++
.../SeaweedFsNativeS3TestContainerTest.java | 106 +++++++
.../writer/InMemoryNativeS3Operations.java | 194 ------------
.../NativeS3RecoverableFsDataOutputStreamTest.java | 112 ++++++-
.../NativeS3RecoverableWriterRecoveryITCase.java | 227 ++++++++++++++
.../NativeS3RecoverableWriterRecoveryTest.java | 349 ---------------------
.../writer/SeaweedFsNativeS3Operations.java | 75 +++++
14 files changed, 1251 insertions(+), 548 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 c883354f213..4afe81c3a9c 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
@@ -262,7 +262,7 @@ class NativeS3RecoverableFsDataOutputStream extends
RecoverableFsDataOutputStrea
if (currentPartSize > 0) {
currentOutputStream.flush();
- incompletePartKey = key + "/.incomplete/" + uploadId + "/" +
UUID.randomUUID();
+ incompletePartKey = ".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/AbstractNativeS3HAApplicationRunITCase.java
b/flink-filesystems/flink-s3-fs-native/src/test/java/org/apache/flink/fs/s3native/AbstractNativeS3HAApplicationRunITCase.java
new file mode 100644
index 00000000000..2911e6ce2fc
--- /dev/null
+++
b/flink-filesystems/flink-s3-fs-native/src/test/java/org/apache/flink/fs/s3native/AbstractNativeS3HAApplicationRunITCase.java
@@ -0,0 +1,102 @@
+/*
+ * 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;
+
+import org.apache.flink.api.common.ApplicationState;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.core.fs.FileSystem;
+import
org.apache.flink.runtime.highavailability.AbstractHAApplicationRunITCase;
+import org.apache.flink.runtime.highavailability.ApplicationResultStoreOptions;
+import
org.apache.flink.runtime.highavailability.FileSystemApplicationResultStore;
+import org.apache.flink.runtime.testutils.CommonTestUtils;
+
+import org.junit.jupiter.api.AfterAll;
+import software.amazon.awssdk.services.s3.model.S3Object;
+
+import java.util.List;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * Runs {@link AbstractHAApplicationRunITCase} on backed by {@link
SeaweedFsNativeS3TestContainer}.
+ */
+abstract class AbstractNativeS3HAApplicationRunITCase extends
AbstractHAApplicationRunITCase {
+
+ static final String CLUSTER_ID = "test-cluster";
+ private static final String APPLICATION_RESULT_STORE_FOLDER = "ars";
+
+ /** Provided by the concrete runner that owns the SeaweedFS cluster
extension. */
+ abstract SeaweedFsNativeS3TestContainer getSeaweedFsContainer();
+
+ static Configuration createConfiguration(SeaweedFsNativeS3TestContainer
container) {
+ final Configuration config = new Configuration();
+ container.setS3ConfigOptions(config);
+ config.set(ApplicationResultStoreOptions.DELETE_ON_COMMIT,
Boolean.FALSE);
+ config.set(
+ ApplicationResultStoreOptions.STORAGE_PATH,
+ s3UriWithSubPath(container, CLUSTER_ID,
APPLICATION_RESULT_STORE_FOLDER));
+ return addHaConfiguration(config, s3UriWithSubPath(container,
CLUSTER_ID));
+ }
+
+ private static String s3UriWithSubPath(
+ SeaweedFsNativeS3TestContainer container, String... subfolders) {
+ return container.getS3UriForDefaultBucket() + "/" + String.join("/",
subfolders);
+ }
+
+ @AfterAll
+ static void unsetFileSystem() {
+ FileSystem.initialize(new Configuration(), null);
+ }
+
+ @Override
+ protected void runAfterApplicationTermination() throws Exception {
+ final SeaweedFsNativeS3TestContainer container =
getSeaweedFsContainer();
+ final String prefix = String.join("/", CLUSTER_ID,
APPLICATION_RESULT_STORE_FOLDER);
+
+ CommonTestUtils.waitUntilCondition(
+ () -> {
+ final List<S3Object> objects =
container.listObjects(prefix);
+ return objects.stream()
+ .map(S3Object::key)
+ .anyMatch(
+ FileSystemApplicationResultStore
+
::hasValidApplicationResultStoreEntryExtension)
+ && objects.stream()
+ .map(S3Object::key)
+ .noneMatch(
+ FileSystemApplicationResultStore
+
::hasValidDirtyApplicationResultStoreEntryExtension);
+ },
+ 2000L);
+
+ final List<S3Object> objects = container.listObjects(prefix);
+ assertThat(objects).hasSize(1);
+
+ final String key = objects.get(0).key();
+ assertThat(key)
+ .matches(
+ FileSystemApplicationResultStore
+ ::hasValidApplicationResultStoreEntryExtension)
+ .doesNotMatch(
+ FileSystemApplicationResultStore
+
::hasValidDirtyApplicationResultStoreEntryExtension);
+
+
assertThat(container.getObjectAsString(key)).contains(ApplicationState.FINISHED.name());
+ }
+}
diff --git
a/flink-filesystems/flink-s3-fs-native/src/test/java/org/apache/flink/fs/s3native/AbstractNativeS3HAJobRunITCase.java
b/flink-filesystems/flink-s3-fs-native/src/test/java/org/apache/flink/fs/s3native/AbstractNativeS3HAJobRunITCase.java
new file mode 100644
index 00000000000..dfeab993801
--- /dev/null
+++
b/flink-filesystems/flink-s3-fs-native/src/test/java/org/apache/flink/fs/s3native/AbstractNativeS3HAJobRunITCase.java
@@ -0,0 +1,99 @@
+/*
+ * 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;
+
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.core.fs.FileSystem;
+import org.apache.flink.runtime.clusterframework.ApplicationStatus;
+import org.apache.flink.runtime.highavailability.AbstractHAJobRunITCase;
+import org.apache.flink.runtime.highavailability.FileSystemJobResultStore;
+import org.apache.flink.runtime.highavailability.JobResultStoreOptions;
+import org.apache.flink.runtime.testutils.CommonTestUtils;
+
+import org.junit.jupiter.api.AfterAll;
+import software.amazon.awssdk.services.s3.model.S3Object;
+
+import java.util.List;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * Runs {@link AbstractHAJobRunITCase} with the HA data stored in a
native-S3-backed {@link
+ * SeaweedFsNativeS3TestContainer}.
+ */
+abstract class AbstractNativeS3HAJobRunITCase extends AbstractHAJobRunITCase {
+
+ static final String CLUSTER_ID = "test-cluster";
+ private static final String JOB_RESULT_STORE_FOLDER = "jrs";
+
+ /** Provided by the concrete runner that owns the SeaweedFS cluster
extension. */
+ abstract SeaweedFsNativeS3TestContainer getSeaweedFsContainer();
+
+ static Configuration createConfiguration(SeaweedFsNativeS3TestContainer
container) {
+ final Configuration config = new Configuration();
+ container.setS3ConfigOptions(config);
+ config.set(JobResultStoreOptions.DELETE_ON_COMMIT, Boolean.FALSE);
+ config.set(
+ JobResultStoreOptions.STORAGE_PATH,
+ s3UriWithSubPath(container, CLUSTER_ID,
JOB_RESULT_STORE_FOLDER));
+ return addHaConfiguration(config, s3UriWithSubPath(container,
CLUSTER_ID));
+ }
+
+ private static String s3UriWithSubPath(
+ SeaweedFsNativeS3TestContainer container, String... subfolders) {
+ return container.getS3UriForDefaultBucket() + "/" + String.join("/",
subfolders);
+ }
+
+ @AfterAll
+ static void unsetFileSystem() {
+ FileSystem.initialize(new Configuration(), null);
+ }
+
+ @Override
+ protected void runAfterJobTermination() throws Exception {
+ final SeaweedFsNativeS3TestContainer container =
getSeaweedFsContainer();
+ final String prefix = String.join("/", CLUSTER_ID,
JOB_RESULT_STORE_FOLDER);
+
+ CommonTestUtils.waitUntilCondition(
+ () -> {
+ final List<S3Object> objects =
container.listObjects(prefix);
+ return objects.stream()
+ .map(S3Object::key)
+ .anyMatch(
+ FileSystemJobResultStore
+
::hasValidJobResultStoreEntryExtension)
+ && objects.stream()
+ .map(S3Object::key)
+ .noneMatch(
+ FileSystemJobResultStore
+
::hasValidDirtyJobResultStoreEntryExtension);
+ },
+ 2000L);
+
+ final List<S3Object> objects = container.listObjects(prefix);
+ assertThat(objects).hasSize(1);
+
+ final String key = objects.get(0).key();
+ assertThat(key)
+
.matches(FileSystemJobResultStore::hasValidJobResultStoreEntryExtension)
+
.doesNotMatch(FileSystemJobResultStore::hasValidDirtyJobResultStoreEntryExtension);
+
+
assertThat(container.getObjectAsString(key)).contains(ApplicationStatus.SUCCEEDED.name());
+ }
+}
diff --git
a/flink-filesystems/flink-s3-fs-native/src/test/java/org/apache/flink/fs/s3native/HAApplicationRunOnNativeS3FileSystemITCase.java
b/flink-filesystems/flink-s3-fs-native/src/test/java/org/apache/flink/fs/s3native/HAApplicationRunOnNativeS3FileSystemITCase.java
new file mode 100644
index 00000000000..7e55077067e
--- /dev/null
+++
b/flink-filesystems/flink-s3-fs-native/src/test/java/org/apache/flink/fs/s3native/HAApplicationRunOnNativeS3FileSystemITCase.java
@@ -0,0 +1,39 @@
+/*
+ * 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;
+
+import org.junit.jupiter.api.Order;
+import org.junit.jupiter.api.extension.RegisterExtension;
+
+/** Base ITCase tests for HA Application on Native s3. */
+class HAApplicationRunOnNativeS3FileSystemITCase extends
AbstractNativeS3HAApplicationRunITCase {
+
+ // AbstractHAApplicationRunITCase already registers its own extension at
@Order(1), so this
+ // one must run after it.
+ @RegisterExtension
+ @Order(2)
+ private static final SeaweedFsNativeS3HAClusterExtension CLUSTER_EXTENSION
=
+ new SeaweedFsNativeS3HAClusterExtension(
+
AbstractNativeS3HAApplicationRunITCase::createConfiguration);
+
+ @Override
+ SeaweedFsNativeS3TestContainer getSeaweedFsContainer() {
+ return CLUSTER_EXTENSION.getContainer();
+ }
+}
diff --git
a/flink-filesystems/flink-s3-fs-native/src/test/java/org/apache/flink/fs/s3native/HAJobRunOnNativeS3FileSystemITCase.java
b/flink-filesystems/flink-s3-fs-native/src/test/java/org/apache/flink/fs/s3native/HAJobRunOnNativeS3FileSystemITCase.java
new file mode 100644
index 00000000000..444897876cb
--- /dev/null
+++
b/flink-filesystems/flink-s3-fs-native/src/test/java/org/apache/flink/fs/s3native/HAJobRunOnNativeS3FileSystemITCase.java
@@ -0,0 +1,39 @@
+/*
+ * 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;
+
+import org.junit.jupiter.api.Order;
+import org.junit.jupiter.api.extension.RegisterExtension;
+
+/** HA Job Test Implementation. * */
+class HAJobRunOnNativeS3FileSystemITCase extends
AbstractNativeS3HAJobRunITCase {
+
+ // AbstractHAJobRunITCase already registers its own extension at
@Order(1), so this one must
+ // run after it.
+ @RegisterExtension
+ @Order(2)
+ private static final SeaweedFsNativeS3HAClusterExtension CLUSTER_EXTENSION
=
+ new SeaweedFsNativeS3HAClusterExtension(
+ AbstractNativeS3HAJobRunITCase::createConfiguration);
+
+ @Override
+ SeaweedFsNativeS3TestContainer getSeaweedFsContainer() {
+ return CLUSTER_EXTENSION.getContainer();
+ }
+}
diff --git
a/flink-filesystems/flink-s3-fs-native/src/test/java/org/apache/flink/fs/s3native/NativeS3FileSystemITCase.java
b/flink-filesystems/flink-s3-fs-native/src/test/java/org/apache/flink/fs/s3native/NativeS3FileSystemITCase.java
new file mode 100644
index 00000000000..5b10a9319d2
--- /dev/null
+++
b/flink-filesystems/flink-s3-fs-native/src/test/java/org/apache/flink/fs/s3native/NativeS3FileSystemITCase.java
@@ -0,0 +1,165 @@
+/*
+ * 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;
+
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.core.fs.FSDataInputStream;
+import org.apache.flink.core.fs.FSDataOutputStream;
+import org.apache.flink.core.fs.FileStatus;
+import org.apache.flink.core.fs.FileSystem;
+import org.apache.flink.core.fs.Path;
+import org.apache.flink.core.fs.RecoverableFsDataOutputStream;
+import org.apache.flink.core.fs.RecoverableWriter;
+import org.apache.flink.core.testutils.AllCallbackWrapper;
+import org.apache.flink.core.testutils.TestContainerExtension;
+
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.RegisterExtension;
+
+import java.net.URI;
+import java.nio.charset.StandardCharsets;
+import java.util.UUID;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatCode;
+
+/** Exercises native S3 filesystem operations directly. */
+class NativeS3FileSystemITCase {
+
+ @RegisterExtension
+ private static final
AllCallbackWrapper<TestContainerExtension<SeaweedFsNativeS3TestContainer>>
+ SEAWEEDFS_EXTENSION =
+ new AllCallbackWrapper<>(
+ new
TestContainerExtension<>(SeaweedFsNativeS3TestContainer::new));
+
+ private static FileSystem fs;
+ private static String bucketUri;
+
+ private static SeaweedFsNativeS3TestContainer container() {
+ return SEAWEEDFS_EXTENSION.getCustomExtension().getTestContainer();
+ }
+
+ @BeforeAll
+ static void setUp() throws Exception {
+ final Configuration config = new Configuration();
+ container().setS3ConfigOptions(config);
+
+ final NativeS3FileSystemFactory factory = new
NativeS3FileSystemFactory();
+ factory.configure(config);
+
+ bucketUri = container().getS3UriForDefaultBucket();
+ fs = factory.create(URI.create(bucketUri + "/"));
+ }
+
+ @Test
+ void testWriteReadAndStat() throws Exception {
+ final Path file = path("dir/" + UUID.randomUUID() + ".txt");
+ final byte[] data = "hello seaweedfs".getBytes(StandardCharsets.UTF_8);
+ write(file, data);
+
+ assertThat(fs.exists(file)).isTrue();
+ assertThat(fs.getFileStatus(file).getLen()).isEqualTo(data.length);
+ assertThat(read(file, data.length)).isEqualTo(data);
+ }
+
+ @Test
+ void testListRenameDelete() throws Exception {
+ final String dir = "listdir-" + UUID.randomUUID();
+ final Path a = path(dir + "/a.txt");
+ final Path b = path(dir + "/b.txt");
+ write(a, "a".getBytes(StandardCharsets.UTF_8));
+ write(b, "b".getBytes(StandardCharsets.UTF_8));
+
+ final FileStatus[] listed = fs.listStatus(path(dir));
+ assertThat(listed)
+ .extracting(status -> status.getPath().getName())
+ .containsExactlyInAnyOrder("a.txt", "b.txt");
+
+ final Path renamed = path(dir + "/c.txt");
+ assertThat(fs.rename(a, renamed)).isTrue();
+ assertThat(fs.exists(a)).isFalse();
+ assertThat(fs.exists(renamed)).isTrue();
+
+ assertThat(fs.delete(path(dir), true)).isTrue();
+ assertThat(fs.exists(renamed)).isFalse();
+ }
+
+ @Test
+ void testMkdirsDoesNotThrowOnObjectStore() {
+ // S3 has no real directories, so mkdirs() on an object store is a
no-op that must not
+ // throw, even though nothing is actually created.
+ assertThatCode(() -> fs.mkdirs(path("mkdir-" + UUID.randomUUID())))
+ .doesNotThrowAnyException();
+ }
+
+ @Test
+ void testRecoverableWriterMultipartCommit() throws Exception {
+ final Path file = path("recoverable-" + UUID.randomUUID() + ".bin");
+ // Bigger than the S3 multipart minimum part size so the commit
exercises a real
+ // multipart upload rather than a single-shot put.
+ final byte[] data =
+ payload((int)
NativeS3FileSystemFactory.S3_MULTIPART_MIN_PART_SIZE + (1024 * 1024));
+
+ final RecoverableWriter writer = fs.createRecoverableWriter();
+ final RecoverableFsDataOutputStream out = writer.open(file);
+ out.write(data);
+ out.persist();
+ out.closeForCommit().commit();
+
+ assertThat(fs.getFileStatus(file).getLen()).isEqualTo(data.length);
+ assertThat(read(file, data.length)).isEqualTo(data);
+ }
+
+ private static Path path(String name) {
+ return new Path(bucketUri + "/" + name);
+ }
+
+ private static void write(Path path, byte[] data) throws Exception {
+ try (FSDataOutputStream out = fs.create(path,
FileSystem.WriteMode.OVERWRITE)) {
+ out.write(data);
+ }
+ }
+
+ private static byte[] read(Path path, int length) throws Exception {
+ final byte[] target = new byte[length];
+ try (FSDataInputStream in = fs.open(path)) {
+ int offset = 0;
+ while (offset < length) {
+ final int read = in.read(target, offset, length - offset);
+ if (read <= 0) {
+ // read == 0 is treated as EOF.
+ // To avoid spinning without progress just breakout.
+ break;
+ }
+ offset += read;
+ }
+ assertThat(offset).isEqualTo(length);
+ }
+ return target;
+ }
+
+ private static byte[] payload(int size) {
+ final byte[] data = new byte[size];
+ for (int i = 0; i < size; i++) {
+ data[i] = (byte) (i % 127);
+ }
+ return data;
+ }
+}
diff --git
a/flink-filesystems/flink-s3-fs-native/src/test/java/org/apache/flink/fs/s3native/SeaweedFsNativeS3HAClusterExtension.java
b/flink-filesystems/flink-s3-fs-native/src/test/java/org/apache/flink/fs/s3native/SeaweedFsNativeS3HAClusterExtension.java
new file mode 100644
index 00000000000..dac14d6028b
--- /dev/null
+++
b/flink-filesystems/flink-s3-fs-native/src/test/java/org/apache/flink/fs/s3native/SeaweedFsNativeS3HAClusterExtension.java
@@ -0,0 +1,125 @@
+/*
+ * 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;
+
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.core.fs.FileSystem;
+import org.apache.flink.core.testutils.AllCallbackWrapper;
+import org.apache.flink.core.testutils.TestContainerExtension;
+import org.apache.flink.runtime.testutils.MiniClusterResourceConfiguration;
+import org.apache.flink.test.junit5.MiniClusterExtension;
+
+import org.junit.jupiter.api.extension.AfterAllCallback;
+import org.junit.jupiter.api.extension.AfterEachCallback;
+import org.junit.jupiter.api.extension.BeforeAllCallback;
+import org.junit.jupiter.api.extension.BeforeEachCallback;
+import org.junit.jupiter.api.extension.ExtensionContext;
+import org.junit.jupiter.api.extension.ParameterContext;
+import org.junit.jupiter.api.extension.ParameterResolutionException;
+import org.junit.jupiter.api.extension.ParameterResolver;
+
+import java.util.function.Function;
+
+/**
+ * Bundles a {@link SeaweedFsNativeS3TestContainer} and a {@link
MiniClusterExtension} configured to
+ * use it, so that HA IT cases backed by the native S3 FS don't each have to
wire up and order the
+ * two extensions themselves.
+ *
+ * <p>The {@link MiniClusterExtension} is created lazily in {@link #beforeAll}
(after the container
+ * is up), so JUnit never discovers it as a registered extension on its own.
This class therefore
+ * implements the callback/resolver interfaces itself and delegates to the
inner {@link
+ * MiniClusterExtension}, so parameter injection (e.g. {@code
@InjectMiniCluster}) and the per-test
+ * lifecycle still work.
+ */
+final class SeaweedFsNativeS3HAClusterExtension
+ implements BeforeAllCallback,
+ AfterAllCallback,
+ BeforeEachCallback,
+ AfterEachCallback,
+ ParameterResolver {
+
+ private final
AllCallbackWrapper<TestContainerExtension<SeaweedFsNativeS3TestContainer>>
+ seaweedFsExtension =
+ new AllCallbackWrapper<>(
+ new
TestContainerExtension<>(SeaweedFsNativeS3TestContainer::new));
+
+ private final Function<SeaweedFsNativeS3TestContainer, Configuration>
configurationFactory;
+
+ private MiniClusterExtension miniClusterExtension;
+
+ SeaweedFsNativeS3HAClusterExtension(
+ Function<SeaweedFsNativeS3TestContainer, Configuration>
configurationFactory) {
+ this.configurationFactory = configurationFactory;
+ }
+
+ SeaweedFsNativeS3TestContainer getContainer() {
+ return seaweedFsExtension.getCustomExtension().getTestContainer();
+ }
+
+ @Override
+ public void beforeAll(ExtensionContext context) throws Exception {
+ seaweedFsExtension.beforeAll(context);
+ miniClusterExtension =
+ new MiniClusterExtension(
+ () -> {
+ final Configuration configuration =
+ configurationFactory.apply(getContainer());
+ FileSystem.initialize(configuration, null);
+ return new
MiniClusterResourceConfiguration.Builder()
+ .setConfiguration(configuration)
+ .build();
+ });
+ miniClusterExtension.beforeAll(context);
+ }
+
+ @Override
+ public void afterAll(ExtensionContext context) throws Exception {
+ try {
+ if (miniClusterExtension != null) {
+ miniClusterExtension.afterAll(context);
+ }
+ } finally {
+ seaweedFsExtension.afterAll(context);
+ }
+ }
+
+ @Override
+ public void beforeEach(ExtensionContext context) throws Exception {
+ miniClusterExtension.beforeEach(context);
+ }
+
+ @Override
+ public void afterEach(ExtensionContext context) throws Exception {
+ miniClusterExtension.afterEach(context);
+ }
+
+ @Override
+ public boolean supportsParameter(
+ ParameterContext parameterContext, ExtensionContext
extensionContext)
+ throws ParameterResolutionException {
+ return miniClusterExtension.supportsParameter(parameterContext,
extensionContext);
+ }
+
+ @Override
+ public Object resolveParameter(
+ ParameterContext parameterContext, ExtensionContext
extensionContext)
+ throws ParameterResolutionException {
+ return miniClusterExtension.resolveParameter(parameterContext,
extensionContext);
+ }
+}
diff --git
a/flink-filesystems/flink-s3-fs-native/src/test/java/org/apache/flink/fs/s3native/SeaweedFsNativeS3TestContainer.java
b/flink-filesystems/flink-s3-fs-native/src/test/java/org/apache/flink/fs/s3native/SeaweedFsNativeS3TestContainer.java
new file mode 100644
index 00000000000..11ec3e73819
--- /dev/null
+++
b/flink-filesystems/flink-s3-fs-native/src/test/java/org/apache/flink/fs/s3native/SeaweedFsNativeS3TestContainer.java
@@ -0,0 +1,165 @@
+/*
+ * 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;
+
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.core.fs.FileSystem;
+import org.apache.flink.util.DockerImageVersions;
+import org.apache.flink.util.Preconditions;
+
+import com.github.dockerjava.api.command.InspectContainerResponse;
+import org.testcontainers.containers.GenericContainer;
+import org.testcontainers.containers.wait.strategy.HttpWaitStrategy;
+import org.testcontainers.utility.Base58;
+import software.amazon.awssdk.auth.credentials.AwsBasicCredentials;
+import software.amazon.awssdk.auth.credentials.StaticCredentialsProvider;
+import software.amazon.awssdk.regions.Region;
+import software.amazon.awssdk.services.s3.S3Client;
+import software.amazon.awssdk.services.s3.model.S3Object;
+
+import java.net.URI;
+import java.time.Duration;
+import java.util.List;
+import java.util.Locale;
+
+/** Provides a SeaweedFS S3-compatible test instance for the native S3
filesystem. */
+public class SeaweedFsNativeS3TestContainer
+ extends GenericContainer<SeaweedFsNativeS3TestContainer> {
+
+ private static final int DEFAULT_PORT = 8333;
+ private static final String DEFAULT_STORAGE_DIRECTORY = "/data";
+ private static final String HEALTH_ENDPOINT = "/healthz";
+ private static final String AWS_ACCESS_KEY_ID = "AWS_ACCESS_KEY_ID";
+ private static final String AWS_SECRET_ACCESS_KEY =
"AWS_SECRET_ACCESS_KEY";
+
+ private final String accessKey;
+ private final String secretKey;
+ private final String defaultBucketName;
+
+ private S3Client client;
+
+ public SeaweedFsNativeS3TestContainer() {
+ this(randomString("bucket", 6));
+ }
+
+ public SeaweedFsNativeS3TestContainer(String defaultBucketName) {
+ super(DockerImageVersions.SEAWEEDFS);
+
+ this.accessKey = randomString("accessKey", 10);
+ // secrets must have at least 8 characters
+ this.secretKey = randomString("secretKey", 10);
+ this.defaultBucketName = Preconditions.checkNotNull(defaultBucketName);
+
+ withNetworkAliases(randomString("seaweedfs", 6));
+ addExposedPort(DEFAULT_PORT);
+ withEnv(AWS_ACCESS_KEY_ID, accessKey);
+ withEnv(AWS_SECRET_ACCESS_KEY, secretKey);
+ withCommand(
+ "server", "-s3", "-s3.port=" + DEFAULT_PORT, "-dir=" +
DEFAULT_STORAGE_DIRECTORY);
+ setWaitStrategy(
+ new HttpWaitStrategy()
+ .forPort(DEFAULT_PORT)
+ .forPath(HEALTH_ENDPOINT)
+ .withStartupTimeout(Duration.ofMinutes(2)));
+ // A transient 503 during startup can slip past the SDK's default
retry strategy.
+ withStartupAttempts(3);
+ }
+
+ @Override
+ protected void containerIsStarted(InspectContainerResponse containerInfo) {
+ super.containerIsStarted(containerInfo);
+ getClient().createBucket(b -> b.bucket(defaultBucketName));
+ }
+
+ @Override
+ public void stop() {
+ if (client != null) {
+ client.close();
+ client = null;
+ }
+ super.stop();
+ }
+
+ /** Returns a vanilla SDK-v2 client for verification, independent of the
code under test. */
+ public S3Client getClient() {
+ if (client == null) {
+ client =
+ S3Client.builder()
+ .endpointOverride(URI.create(getHttpEndpoint()))
+ .region(Region.US_EAST_1)
+ .credentialsProvider(
+ StaticCredentialsProvider.create(
+
AwsBasicCredentials.create(accessKey, secretKey)))
+ .forcePathStyle(true)
+ .build();
+ }
+ return client;
+ }
+
+ /**
+ * Sets the config required to reach this instance from the native S3
filesystem. SeaweedFS
+ * supports neither AWS chunked encoding nor trailing checksums, so both
are disabled.
+ */
+ public void setS3ConfigOptions(Configuration config) {
+ config.set(NativeS3FileSystemFactory.ENDPOINT, getHttpEndpoint());
+ config.set(NativeS3FileSystemFactory.REGION, Region.US_EAST_1.id());
+ config.set(NativeS3FileSystemFactory.ACCESS_KEY, accessKey);
+ config.set(NativeS3FileSystemFactory.SECRET_KEY, secretKey);
+ config.set(NativeS3FileSystemFactory.PATH_STYLE_ACCESS, true);
+ config.set(NativeS3FileSystemFactory.CHUNKED_ENCODING_ENABLED, false);
+ config.set(NativeS3FileSystemFactory.CHECKSUM_VALIDATION_ENABLED,
false);
+ }
+
+ public void initializeFileSystem(Configuration config) {
+ Preconditions.checkArgument(
+ config.containsKey(NativeS3FileSystemFactory.ENDPOINT.key()),
+ NativeS3FileSystemFactory.ENDPOINT.key()
+ + " needs to be specified before initializing the
FileSystems.");
+ FileSystem.initialize(config, null);
+ }
+
+ /** Returns the internally used default bucket. */
+ public String getDefaultBucketName() {
+ return defaultBucketName;
+ }
+
+ public String getS3UriForDefaultBucket() {
+ return "s3://" + defaultBucketName;
+ }
+
+ public List<S3Object> listObjects(String prefix) {
+ return getClient()
+ .listObjectsV2(b -> b.bucket(defaultBucketName).prefix(prefix))
+ .contents();
+ }
+
+ public String getObjectAsString(String key) {
+ return getClient()
+ .getObjectAsBytes(b -> b.bucket(defaultBucketName).key(key))
+ .asUtf8String();
+ }
+
+ private String getHttpEndpoint() {
+ return String.format("http://%s:%s";, getHost(),
getMappedPort(DEFAULT_PORT));
+ }
+
+ private static String randomString(String prefix, int length) {
+ return String.format("%s-%s", prefix,
Base58.randomString(length).toLowerCase(Locale.ROOT));
+ }
+}
diff --git
a/flink-filesystems/flink-s3-fs-native/src/test/java/org/apache/flink/fs/s3native/SeaweedFsNativeS3TestContainerTest.java
b/flink-filesystems/flink-s3-fs-native/src/test/java/org/apache/flink/fs/s3native/SeaweedFsNativeS3TestContainerTest.java
new file mode 100644
index 00000000000..58951adf782
--- /dev/null
+++
b/flink-filesystems/flink-s3-fs-native/src/test/java/org/apache/flink/fs/s3native/SeaweedFsNativeS3TestContainerTest.java
@@ -0,0 +1,106 @@
+/*
+ * 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;
+
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.core.testutils.EachCallbackWrapper;
+import org.apache.flink.core.testutils.TestContainerExtension;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.RegisterExtension;
+import software.amazon.awssdk.core.sync.RequestBody;
+import software.amazon.awssdk.services.s3.model.Bucket;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** Basic tests for {@link SeaweedFsNativeS3TestContainer}. */
+class SeaweedFsNativeS3TestContainerTest {
+
+ private static final String DEFAULT_BUCKET_NAME = "test-bucket";
+
+ @RegisterExtension
+ private static final
EachCallbackWrapper<TestContainerExtension<SeaweedFsNativeS3TestContainer>>
+ SEAWEEDFS_EXTENSION =
+ new EachCallbackWrapper<>(
+ new TestContainerExtension<>(
+ () -> new
SeaweedFsNativeS3TestContainer(DEFAULT_BUCKET_NAME)));
+
+ private static SeaweedFsNativeS3TestContainer getTestContainer() {
+ return SEAWEEDFS_EXTENSION.getCustomExtension().getTestContainer();
+ }
+
+ @Test
+ void testBucketCreation() {
+ final String bucketName = "other-bucket";
+ getTestContainer().getClient().createBucket(b -> b.bucket(bucketName));
+
+ assertThat(getTestContainer().getClient().listBuckets().buckets())
+ .map(Bucket::name)
+
.containsExactlyInAnyOrder(getTestContainer().getDefaultBucketName(),
bucketName);
+ }
+
+ @Test
+ void testPutObject() {
+ final String key = "test-object";
+ final String content = "test content";
+ getTestContainer()
+ .getClient()
+ .putObject(
+ b ->
b.bucket(getTestContainer().getDefaultBucketName()).key(key),
+ RequestBody.fromString(content));
+
+
assertThat(getTestContainer().getObjectAsString(key)).isEqualTo(content);
+ }
+
+ @Test
+ void testSetS3ConfigOptions() {
+ final Configuration config = new Configuration();
+ getTestContainer().setS3ConfigOptions(config);
+
+
assertThat(config.containsKey(NativeS3FileSystemFactory.ENDPOINT.key())).isTrue();
+
assertThat(config.containsKey(NativeS3FileSystemFactory.REGION.key())).isTrue();
+
assertThat(config.containsKey(NativeS3FileSystemFactory.ACCESS_KEY.key())).isTrue();
+
assertThat(config.containsKey(NativeS3FileSystemFactory.SECRET_KEY.key())).isTrue();
+
assertThat(config.containsKey(NativeS3FileSystemFactory.PATH_STYLE_ACCESS.key())).isTrue();
+
assertThat(config.containsKey(NativeS3FileSystemFactory.CHUNKED_ENCODING_ENABLED.key()))
+ .isTrue();
+
assertThat(config.containsKey(NativeS3FileSystemFactory.CHECKSUM_VALIDATION_ENABLED.key()))
+ .isTrue();
+ }
+
+ @Test
+ void testGetDefaultBucketName() {
+
assertThat(getTestContainer().getDefaultBucketName()).isEqualTo(DEFAULT_BUCKET_NAME);
+ }
+
+ @Test
+ void testDefaultBucketCreation() {
+ assertThat(getTestContainer().getClient().listBuckets().buckets())
+ .singleElement()
+ .extracting(Bucket::name)
+ .isEqualTo(getTestContainer().getDefaultBucketName());
+ }
+
+ @Test
+ void testEndpointRequiredBeforeInitializingFileSystem() {
+ assertThatThrownBy(() -> getTestContainer().initializeFileSystem(new
Configuration()))
+ .isInstanceOf(IllegalArgumentException.class);
+ }
+}
diff --git
a/flink-filesystems/flink-s3-fs-native/src/test/java/org/apache/flink/fs/s3native/writer/InMemoryNativeS3Operations.java
b/flink-filesystems/flink-s3-fs-native/src/test/java/org/apache/flink/fs/s3native/writer/InMemoryNativeS3Operations.java
deleted file mode 100644
index 28d8edfeeef..00000000000
---
a/flink-filesystems/flink-s3-fs-native/src/test/java/org/apache/flink/fs/s3native/writer/InMemoryNativeS3Operations.java
+++ /dev/null
@@ -1,194 +0,0 @@
-/*
- * 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 implementation for {@link NativeS3ObjectOperations}.
- *
- * <p>Backs every reachable S3 operation with hash maps so writer/committer
logic can be exercised
- * without an S3 endpoint (no MinIO/Testcontainers required). The parent's
{@code S3Client} / {@code
- * S3TransferManager} constructor arguments are passed as {@code null} because
no overridden method
- * dereferences them.
- *
- * <p><b>State exposure:</b> the storage maps are exposed as public final
fields so tests can
- * inspect them, corrupt them, or simulate object loss directly:
- *
- * <ul>
- * <li>{@link #storedObjects} — keys written via {@link #putObject(String,
File)} (e.g. the
- * incomplete-tail side objects persisted by {@link
NativeS3RecoverableFsDataOutputStream}).
- * <li>{@link #committedObjects} — keys finalized via {@link
#commitMultiPartUpload}.
- * <li>{@link #openMultipartUploads} — uploadId → partNumber → bytes for
in-flight MPUs; entries
- * are removed on commit or abort.
- * </ul>
- *
- * <p>{@link #getObject} reads from <em>both</em> {@link #storedObjects} and
{@link
- * #committedObjects} so tests can fetch a committed object the same way real
S3 would serve it.
- *
- * <p><b>Thread safety:</b> not thread-safe. Use a single thread per instance,
matching the
- * single-thread invariant of the production {@link
NativeS3RecoverableFsDataOutputStream}.
- */
-public final class InMemoryNativeS3Operations extends NativeS3ObjectOperations
{
-
- public static final String DEFAULT_BUCKET = "test-bucket";
-
- /** Keys written via {@link #putObject(String, File)}. */
- public final Map<String, byte[]> storedObjects = new HashMap<>();
-
- /** Keys finalized via {@link #commitMultiPartUpload}. */
- public final Map<String, byte[]> committedObjects = new HashMap<>();
-
- /** uploadId → partNumber → uploaded bytes for in-flight MPUs. */
- public final Map<String, Map<Integer, byte[]>> openMultipartUploads = new
HashMap<>();
-
- /** When {@code true}, {@link #uploadPart} throws to simulate a
part-upload failure. */
- public boolean failUploadPart = false;
-
- /**
- * When {@code true}, {@link #uploadPart} deletes the local part file
after reading it,
- * simulating an external cleaner that reaps {@code io.tmp.dirs} while the
part is in flight.
- */
- public boolean deletePartFileAfterUpload = false;
-
- /** When {@code true}, {@link #abortMultiPartUpload} throws to simulate an
abort failure. */
- public boolean failAbortMultiPartUpload = false;
-
- /** Number of times {@link #uploadPart} was invoked, including failed
attempts. */
- public int uploadPartAttempts = 0;
-
- /** Number of times {@link #abortMultiPartUpload} was invoked, including
failed attempts. */
- public int abortAttempts = 0;
-
- private final String bucketName;
- private final AtomicInteger uploadIdSeq = new AtomicInteger();
- private final AtomicInteger putObjectSeq = new AtomicInteger();
-
- public InMemoryNativeS3Operations() {
- this(DEFAULT_BUCKET);
- }
-
- public InMemoryNativeS3Operations(String bucketName) {
- super(/* s3Client */ null, /* transferManager */ null, bucketName, /*
useAsync */ false);
- this.bucketName = bucketName;
- }
-
- @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 PutObjectResult putObject(String key, File file) throws IOException
{
- storedObjects.put(key, Files.readAllBytes(file.toPath()));
- return new PutObjectResult("etag-put-" +
putObjectSeq.incrementAndGet());
- }
-
- @Override
- public long getObject(String key, File targetLocation) throws IOException {
- byte[] data = storedObjects.get(key);
- if (data == null) {
- data = committedObjects.get(key);
- }
- if (data == null) {
- throw new IOException("not found: " + key);
- }
- Files.write(targetLocation.toPath(), data);
- return data.length;
- }
-
- @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(bucketName, key,
"final-etag-" + uploadId, null);
- }
-
- @Override
- public void abortMultiPartUpload(String key, String uploadId) throws
IOException {
- abortAttempts++;
- if (failAbortMultiPartUpload) {
- throw new IOException("injected abort failure for uploadId: " +
uploadId);
- }
- openMultipartUploads.remove(uploadId);
- }
-
- @Override
- public boolean deleteObject(String key) {
- return storedObjects.remove(key) != null ||
committedObjects.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 35163cfe6f2..99a6824b9af 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,11 +24,18 @@ 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;
@@ -41,13 +48,13 @@ class NativeS3RecoverableFsDataOutputStreamTest {
@TempDir Path tmp;
- InMemoryNativeS3Operations s3;
+ FakeNativeS3Operations s3;
String uploadId;
NativeS3RecoverableFsDataOutputStream stream;
@BeforeEach
void setUp() throws IOException {
- s3 = new InMemoryNativeS3Operations();
+ s3 = new FakeNativeS3Operations();
uploadId = s3.startMultiPartUpload(KEY);
stream = newStream(s3, uploadId);
stream.write(bytes('A', 5), 0, 5); // < MIN_PART_SIZE, so it is
uploaded during commit
@@ -258,8 +265,8 @@ class NativeS3RecoverableFsDataOutputStreamTest {
assertThat(countLocalFilesIn(dir)).isZero();
}
- private NativeS3RecoverableFsDataOutputStream newStream(
- InMemoryNativeS3Operations ops, String uid) throws IOException {
+ private NativeS3RecoverableFsDataOutputStream
newStream(FakeNativeS3Operations ops, String uid)
+ throws IOException {
return new NativeS3RecoverableFsDataOutputStream(
ops, KEY, uid, tmp.toString(), MIN_PART_SIZE);
}
@@ -295,4 +302,101 @@ 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 failAbortMultiPartUpload = 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++;
+ if (failAbortMultiPartUpload) {
+ throw new IOException("injected abort failure for uploadId: "
+ uploadId);
+ }
+ openMultipartUploads.remove(uploadId);
+ }
+ }
}
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
new file mode 100644
index 00000000000..1c71c1ec710
--- /dev/null
+++
b/flink-filesystems/flink-s3-fs-native/src/test/java/org/apache/flink/fs/s3native/writer/NativeS3RecoverableWriterRecoveryITCase.java
@@ -0,0 +1,227 @@
+/*
+ * 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.apache.flink.core.testutils.AllCallbackWrapper;
+import org.apache.flink.core.testutils.TestContainerExtension;
+import org.apache.flink.fs.s3native.NativeS3FileSystemFactory;
+import org.apache.flink.fs.s3native.SeaweedFsNativeS3TestContainer;
+
+import org.apache.commons.lang3.ArrayUtils;
+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 java.io.IOException;
+import java.util.Arrays;
+import java.util.UUID;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/**
+ * Integration tests for {@link NativeS3RecoverableWriter#recover} running
against SeaweedFS.
+ *
+ * <p>SeaweedFS enforces the S3 5 MiB minimum part size on multipart-complete,
so every scenario
+ * writes one full {@value #PART}-byte first part (the only non-final part)
followed by a small tail
+ * that becomes the final part.
+ *
+ * <p>Terminology used below:
+ *
+ * <pre>
+ * target object (the file the caller is writing, e.g.
"out-<uuid>.txt")
+ * +-- 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())
+ * - 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
+ * multipart upload before returning a resumed output stream
+ * - has no side object at all when persist() is called exactly on a part
boundary
+ * (see recoverWithoutIncompleteTailStillWorks)
+ * </pre>
+ */
+class NativeS3RecoverableWriterRecoveryITCase {
+
+ private static final int PART = (int)
NativeS3FileSystemFactory.S3_MULTIPART_MIN_PART_SIZE;
+ private static final long MIN_PART_SIZE = PART;
+
+ @RegisterExtension
+ private static final
AllCallbackWrapper<TestContainerExtension<SeaweedFsNativeS3TestContainer>>
+ SEAWEEDFS_EXTENSION =
+ new AllCallbackWrapper<>(
+ new
TestContainerExtension<>(SeaweedFsNativeS3TestContainer::new));
+
+ @TempDir java.nio.file.Path tmp;
+
+ private String bucket;
+ private String key;
+ private SeaweedFsNativeS3Operations s3;
+
+ @BeforeEach
+ void setUp() {
+ bucket = getContainer().getDefaultBucketName();
+ key = "out-" + UUID.randomUUID() + ".txt";
+ s3 = new SeaweedFsNativeS3Operations(getContainer().getClient(),
bucket);
+ }
+
+ private static SeaweedFsNativeS3TestContainer getContainer() {
+ return SEAWEEDFS_EXTENSION.getCustomExtension().getTestContainer();
+ }
+
+ private NativeS3RecoverableWriter writer() {
+ return NativeS3RecoverableWriter.writer(s3, tmp.toString(),
MIN_PART_SIZE, 1);
+ }
+
+ private Path targetPath() {
+ return new Path("s3://" + bucket + "/" + key);
+ }
+
+ private String incompletePrefix(String uploadId) {
+ return ".incomplete/" + uploadId + "/";
+ }
+
+ @Test
+ void recoverWithoutIncompleteTailStillWorks() throws Exception {
+ final NativeS3RecoverableWriter writer1 = writer();
+
+ // Write exactly one full part => currentPartSize=0, no side object on
persist.
+ final RecoverableFsDataOutputStream out = writer1.open(targetPath());
+ out.write(bytes('A', PART), 0, PART);
+ final NativeS3Recoverable r = (NativeS3Recoverable) out.persist();
+ assertThat(r.incompleteObjectName()).as("no tail => no side
object").isNull();
+ assertThat(s3.listKeys(incompletePrefix(r.uploadId()))).isEmpty();
+
+ final NativeS3RecoverableWriter writer2 = writer();
+ final RecoverableFsDataOutputStream resumed = writer2.recover(r);
+ resumed.write(bytes('C', 10), 0, 10);
+ resumed.closeForCommit().commit();
+
+ assertContentEquals(s3.readObject(key), concat(bytes('A', PART),
bytes('C', 10)));
+ }
+
+ @Test
+ void recoverWithNestedKeyStillWorks() throws Exception {
+ // Exercise a target key containing "/" path separators, not just a
flat key.
+ key = "nested/path-" + UUID.randomUUID() + "/out.txt";
+ final NativeS3RecoverableWriter writer1 = writer();
+
+ final RecoverableFsDataOutputStream out = writer1.open(targetPath());
+ out.write(bytes('A', PART), 0, PART);
+ out.write(bytes('E', 5), 0, 5);
+ final NativeS3Recoverable r = (NativeS3Recoverable) out.persist();
+ assertThat(r.incompleteObjectName()).as("tail written => side object
expected").isNotNull();
+ assertThat(s3.listKeys(incompletePrefix(r.uploadId())))
+ .containsExactly(r.incompleteObjectName());
+
+ final NativeS3RecoverableWriter writer2 = writer();
+ final RecoverableFsDataOutputStream resumed = writer2.recover(r);
+ resumed.write(bytes('C', 10), 0, 10);
+ resumed.closeForCommit().commit();
+
+ assertContentEquals(
+ s3.readObject(key), concat(bytes('A', PART), bytes('E', 5),
bytes('C', 10)));
+ }
+
+ @Test
+ void recoverFailsCleanlyWhenSideObjectMissing() throws Exception {
+ final NativeS3Recoverable r = persistWithTail();
+ final String sideObjectKey = r.incompleteObjectName();
+ assertThat(sideObjectKey).isNotNull();
+
+ s3.removeObject(sideObjectKey);
+
+ assertRecoverFailsCleanly(r, "Failed to get object");
+ }
+
+ @Test
+ void recoverFailsCleanlyOnLengthMismatch() throws Exception {
+ final NativeS3Recoverable r = persistWithTail();
+ final String sideObjectKey = r.incompleteObjectName();
+
+ // Simulate the side object having been overwritten/corrupted
out-of-band between
+ // persist() and recover() (e.g. a retried writer racing on the same
side-object key, or
+ // an eventual-consistency edge case on a non-AWS S3 implementation):
the side object's
+ // actual length no longer agrees with the length recorded in the
recoverable's metadata.
+ s3.writeObject(sideObjectKey, bytes('X', 99));
+
+ assertRecoverFailsCleanly(r, "unexpected length");
+ }
+
+ /** Writes one full part plus a small tail, forcing a side object to be
created on persist. */
+ private NativeS3Recoverable persistWithTail() throws IOException {
+ final NativeS3RecoverableWriter writer1 = writer();
+ final RecoverableFsDataOutputStream out = writer1.open(targetPath());
+ out.write(bytes('A', PART), 0, PART);
+ out.write(bytes('E', 5), 0, 5);
+ return (NativeS3Recoverable) out.persist();
+ }
+
+ /**
+ * Asserts that recovering {@code r} fails with an {@link IOException}
containing {@code
+ * expectedMessageFragment}, and that no partially-downloaded local file
is left behind.
+ */
+ private void assertRecoverFailsCleanly(NativeS3Recoverable r, String
expectedMessageFragment)
+ throws IOException {
+ final long localFilesBefore = countLocalFilesIn(tmp);
+ final NativeS3RecoverableWriter writer2 = writer();
+
+ assertThatThrownBy(() -> writer2.recover(r))
+ .isInstanceOf(IOException.class)
+ .hasMessageContaining(expectedMessageFragment);
+
+ assertThat(countLocalFilesIn(tmp))
+ .as("partial download must be cleaned up on failure")
+ .isEqualTo(localFilesBefore);
+ }
+
+ private static void assertContentEquals(byte[] actual, byte[] expected) {
+ assertThat(actual).hasSameSizeAs(expected);
+ assertThat(Arrays.equals(actual, expected))
+ .as("committed object content must match every persisted byte")
+ .isTrue();
+ }
+
+ private static long countLocalFilesIn(java.nio.file.Path dir) throws
IOException {
+ if (!java.nio.file.Files.isDirectory(dir)) {
+ return 0;
+ }
+ try (java.util.stream.Stream<java.nio.file.Path> s =
java.nio.file.Files.list(dir)) {
+ return s.count();
+ }
+ }
+
+ private static byte[] bytes(char c, int n) {
+ byte[] b = new byte[n];
+ Arrays.fill(b, (byte) c);
+ return b;
+ }
+
+ private static byte[] concat(byte[]... chunks) {
+ byte[] out = ArrayUtils.EMPTY_BYTE_ARRAY;
+ for (byte[] c : chunks) {
+ out = ArrayUtils.addAll(out, c);
+ }
+ return out;
+ }
+}
diff --git
a/flink-filesystems/flink-s3-fs-native/src/test/java/org/apache/flink/fs/s3native/writer/NativeS3RecoverableWriterRecoveryTest.java
b/flink-filesystems/flink-s3-fs-native/src/test/java/org/apache/flink/fs/s3native/writer/NativeS3RecoverableWriterRecoveryTest.java
deleted file mode 100644
index b6d8e14840b..00000000000
---
a/flink-filesystems/flink-s3-fs-native/src/test/java/org/apache/flink/fs/s3native/writer/NativeS3RecoverableWriterRecoveryTest.java
+++ /dev/null
@@ -1,349 +0,0 @@
-/*
- * 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.apache.flink.core.fs.RecoverableWriter;
-
-import org.junit.jupiter.api.Test;
-import org.junit.jupiter.api.io.TempDir;
-
-import java.io.ByteArrayOutputStream;
-import java.io.IOException;
-import java.util.Arrays;
-
-import static org.assertj.core.api.Assertions.assertThat;
-import static org.assertj.core.api.Assertions.assertThatThrownBy;
-
-/** Tests for {@link NativeS3RecoverableWriter#recover}. */
-class NativeS3RecoverableWriterRecoveryTest {
-
- private static final String BUCKET =
InMemoryNativeS3Operations.DEFAULT_BUCKET;
- private static final String KEY = "out.txt";
- private static final long MIN_PART_SIZE = 10L;
-
- @TempDir java.nio.file.Path tmp;
-
- @Test
- void persistThenRecoverPreservesTailBytes() throws Exception {
- InMemoryNativeS3Operations s3 = new InMemoryNativeS3Operations();
- NativeS3RecoverableWriter writer1 =
- NativeS3RecoverableWriter.writer(
- s3, tmp.toString(), MIN_PART_SIZE, /* maxConcurrent */
1);
-
- // --- Phase 1: write 15 bytes; minPartSize=10 ⇒ 10B uploaded as part
#1, 5B tail in memory.
- RecoverableFsDataOutputStream out = writer1.open(new Path("s3://" +
BUCKET + "/" + KEY));
- byte[] firstChunk = bytes('A', 10); // becomes part #1
- byte[] tail = bytes('E', 5); // becomes the persisted side object
- out.write(firstChunk, 0, firstChunk.length);
- out.write(tail, 0, tail.length);
-
- // --- Phase 2: checkpoint barrier: persist() and round-trip metadata
through the
- // serializer (this is what Flink does when storing
checkpoint state).
- RecoverableWriter.ResumeRecoverable r = out.persist();
- byte[] checkpointed =
-
NativeS3RecoverableSerializer.INSTANCE.serialize((NativeS3Recoverable) r);
-
- // Sanity: the 5-byte tail is sitting in S3 as a side object right now.
- assertThat(s3.storedObjects).hasSize(1);
- String sideObjectKey = s3.storedObjects.keySet().iterator().next();
- assertThat(sideObjectKey).startsWith(KEY + "/.incomplete/");
- assertThat(s3.storedObjects.get(sideObjectKey)).containsExactly(tail);
-
- // --- Phase 3: simulate task crash + restore from checkpoint. We
deliberately do NOT
- // close `out` because in a real crash the JVM dies;
close() would also abort
- // the MPU and invalidate the recoverable.
- NativeS3Recoverable restored =
- NativeS3RecoverableSerializer.INSTANCE.deserialize(
- NativeS3RecoverableSerializer.INSTANCE.getVersion(),
checkpointed);
- assertThat(restored.incompleteObjectName())
- .as("metadata survives the checkpoint round-trip")
- .isEqualTo(sideObjectKey);
- assertThat(restored.incompleteObjectLength()).isEqualTo(tail.length);
-
- NativeS3RecoverableWriter writer2 =
- NativeS3RecoverableWriter.writer(s3, tmp.toString(),
MIN_PART_SIZE, 1);
- RecoverableFsDataOutputStream resumed = writer2.recover(restored);
-
- // --- Phase 4: continue writing 10 more bytes, then commit.
- byte[] afterResume = bytes('F', 10);
- resumed.write(afterResume, 0, afterResume.length);
- RecoverableFsDataOutputStream.Committer committer =
resumed.closeForCommit();
- committer.commit();
-
- // --- Phase 5: read what landed in S3.
- byte[] finalObject = s3.committedObjects.get(KEY);
- assertThat(finalObject).isNotNull();
-
- // EXPECTED (correct exactly-once): 25 bytes — "AAAAAAAAAA" + "EEEEE"
+ "FFFFFFFFFF"
- ByteArrayOutputStream expected = new ByteArrayOutputStream();
- expected.write(firstChunk);
- expected.write(tail);
- expected.write(afterResume);
-
- // ON UN-FIXED CODE this assertion FAILS:
- // expected length 25 ("AAAAAAAAAAEEEEEFFFFFFFFFF")
- // actual length 20 ("AAAAAAAAAAFFFFFFFFFF")
- // The 5 'E' bytes — the tail durably persisted to S3 at checkpoint
time — would be gone.
- assertThat(finalObject)
- .as("recover() must replay the persisted tail before
continuing")
- .containsExactly(expected.toByteArray());
-
- // The side object is intentionally NOT deleted by recover() so that
re-recovery from the
- // same checkpoint stays correct. Cleanup is owned by
cleanupRecoverableState(), which
- // Flink invokes once the checkpoint is retired.
- assertThat(s3.storedObjects)
- .as("side object outlives recover() to support re-recovery
from same checkpoint")
- .containsKey(sideObjectKey);
- assertThat(writer2.cleanupRecoverableState(restored)).isTrue();
- assertThat(s3.storedObjects)
- .as("cleanupRecoverableState must delete the side object")
- .doesNotContainKey(sideObjectKey);
- }
-
- /**
- * Pre-existing happy path: a recoverable with NO incomplete tail (parts
only) must still work.
- */
- @Test
- void recoverWithoutIncompleteTailStillWorks() throws Exception {
- InMemoryNativeS3Operations s3 = new InMemoryNativeS3Operations();
- NativeS3RecoverableWriter writer1 =
- NativeS3RecoverableWriter.writer(s3, tmp.toString(),
MIN_PART_SIZE, 1);
-
- // Write exactly 20 bytes => two full parts, currentPartSize=0, no
side object on persist.
- RecoverableFsDataOutputStream out = writer1.open(new Path("s3://" +
BUCKET + "/" + KEY));
- out.write(bytes('A', 10), 0, 10);
- out.write(bytes('B', 10), 0, 10);
- RecoverableWriter.ResumeRecoverable r = out.persist();
- assertThat(((NativeS3Recoverable) r).incompleteObjectName())
- .as("no tail => no side object")
- .isNull();
- assertThat(s3.storedObjects).isEmpty();
-
- // Resume and append another 10 bytes, commit.
- NativeS3RecoverableWriter writer2 =
- NativeS3RecoverableWriter.writer(s3, tmp.toString(),
MIN_PART_SIZE, 1);
- RecoverableFsDataOutputStream resumed = writer2.recover(r);
- resumed.write(bytes('C', 10), 0, 10);
- resumed.closeForCommit().commit();
-
- assertThat(s3.committedObjects.get(KEY))
- .containsExactly(concat(bytes('A', 10), bytes('B', 10),
bytes('C', 10)));
- }
-
- /** If the side object is gone from S3, recover() must fail cleanly and
clean up local state. */
- @Test
- void recoverFailsCleanlyWhenSideObjectMissing() throws Exception {
- InMemoryNativeS3Operations s3 = new InMemoryNativeS3Operations();
- NativeS3RecoverableWriter writer1 =
- NativeS3RecoverableWriter.writer(s3, tmp.toString(),
MIN_PART_SIZE, 1);
- RecoverableFsDataOutputStream out = writer1.open(new Path("s3://" +
BUCKET + "/" + KEY));
- out.write(bytes('A', 10), 0, 10);
- out.write(bytes('E', 5), 0, 5);
- NativeS3Recoverable r = (NativeS3Recoverable) out.persist();
- String sideObjectKey = r.incompleteObjectName();
- assertThat(sideObjectKey).isNotNull();
-
- s3.storedObjects.remove(sideObjectKey);
- long localFilesBefore = countLocalFilesIn(tmp);
- NativeS3RecoverableWriter writer2 =
- NativeS3RecoverableWriter.writer(s3, tmp.toString(),
MIN_PART_SIZE, 1);
-
- assertThatThrownBy(() -> writer2.recover(r))
- .isInstanceOf(IOException.class)
- .hasMessageContaining("not found");
-
- assertThat(countLocalFilesIn(tmp))
- .as("partial download must be cleaned up on failure")
- .isEqualTo(localFilesBefore);
- }
-
- /**
- * If the side object is the wrong length, recover() must fail and clean
up the partial file.
- */
- @Test
- void recoverFailsCleanlyOnLengthMismatch() throws Exception {
- InMemoryNativeS3Operations s3 = new InMemoryNativeS3Operations();
- NativeS3RecoverableWriter writer1 =
- NativeS3RecoverableWriter.writer(s3, tmp.toString(),
MIN_PART_SIZE, 1);
- RecoverableFsDataOutputStream out = writer1.open(new Path("s3://" +
BUCKET + "/" + KEY));
- out.write(bytes('A', 10), 0, 10);
- out.write(bytes('E', 5), 0, 5);
- NativeS3Recoverable r = (NativeS3Recoverable) out.persist();
- String sideObjectKey = r.incompleteObjectName();
-
- // Corrupt the side object so its actual length disagrees with the
metadata.
- s3.storedObjects.put(sideObjectKey, bytes('X', 99));
-
- long localFilesBefore = countLocalFilesIn(tmp);
- NativeS3RecoverableWriter writer2 =
- NativeS3RecoverableWriter.writer(s3, tmp.toString(),
MIN_PART_SIZE, 1);
-
- assertThatThrownBy(() -> writer2.recover(r))
- .isInstanceOf(IOException.class)
- .hasMessageContaining("unexpected length");
-
- assertThat(countLocalFilesIn(tmp))
- .as("partial download must be cleaned up on failure")
- .isEqualTo(localFilesBefore);
- }
-
- /**
- * persist → recover → persist → recover chain. Models multiple checkpoint
cycles, each one
- * recovering from the previous. Every persisted byte must end up in the
committed object.
- */
- @Test
- void recoverThenPersistThenRecoverPreservesAllBytes() throws Exception {
- InMemoryNativeS3Operations s3 = new InMemoryNativeS3Operations();
-
- // Cycle 1: write 12 bytes (10 -> part, 2 -> tail), persist.
- NativeS3RecoverableWriter w1 =
- NativeS3RecoverableWriter.writer(s3, tmp.toString(),
MIN_PART_SIZE, 1);
- RecoverableFsDataOutputStream s1 = w1.open(new Path("s3://" + BUCKET +
"/" + KEY));
- s1.write(bytes('A', 10), 0, 10);
- s1.write(bytes('B', 2), 0, 2);
- NativeS3Recoverable r1 = (NativeS3Recoverable) s1.persist();
- String firstSideObject = r1.incompleteObjectName();
- assertThat(firstSideObject).isNotNull();
-
- // Cycle 2: recover from r1, write 3 bytes => total tail now 2+3=5
(still < minPartSize),
- // persist again => new side object containing the combined 5-byte
tail.
- NativeS3RecoverableWriter w2 =
- NativeS3RecoverableWriter.writer(s3, tmp.toString(),
MIN_PART_SIZE, 1);
- RecoverableFsDataOutputStream s2 = w2.recover(r1);
- s2.write(bytes('C', 3), 0, 3);
- NativeS3Recoverable r2 = (NativeS3Recoverable) s2.persist();
- String secondSideObject = r2.incompleteObjectName();
- assertThat(secondSideObject).isNotEqualTo(firstSideObject);
- assertThat(s3.storedObjects.get(secondSideObject))
- .as("second side object must contain old tail + new bytes")
- .containsExactly(concat(bytes('B', 2), bytes('C', 3)));
-
- // Cycle 3: recover from r2, write 7 more bytes => tail becomes 5+7=12
>= minPartSize =>
- // upload.
- // Then commit.
- NativeS3RecoverableWriter w3 =
- NativeS3RecoverableWriter.writer(s3, tmp.toString(),
MIN_PART_SIZE, 1);
- RecoverableFsDataOutputStream s3stream = w3.recover(r2);
- s3stream.write(bytes('D', 7), 0, 7);
- s3stream.closeForCommit().commit();
-
- // Final object must be A*10 + B*2 + C*3 + D*7 = 22 bytes.
- assertThat(s3.committedObjects.get(KEY))
- .containsExactly(
- concat(bytes('A', 10), bytes('B', 2), bytes('C', 3),
bytes('D', 7)));
-
- // Both old side objects survive until cleanupRecoverableState is
called per checkpoint.
- assertThat(s3.storedObjects).containsKeys(firstSideObject,
secondSideObject);
- assertThat(w3.cleanupRecoverableState(r1)).isTrue();
- assertThat(w3.cleanupRecoverableState(r2)).isTrue();
- assertThat(s3.storedObjects).doesNotContainKeys(firstSideObject,
secondSideObject);
- }
-
- @Test
- void dataWrittenAfterLastPersistIsDiscardedOnRecovery() throws Exception {
- InMemoryNativeS3Operations s3 = new InMemoryNativeS3Operations();
- NativeS3RecoverableWriter writer1 =
- NativeS3RecoverableWriter.writer(s3, tmp.toString(),
MIN_PART_SIZE, 1);
-
- RecoverableFsDataOutputStream out = writer1.open(new Path("s3://" +
BUCKET + "/" + KEY));
- out.write(bytes('A', 10), 0, 10); // becomes part #1
- out.write(bytes('E', 5), 0, 5); // the tail captured by persist()
- NativeS3Recoverable r = (NativeS3Recoverable) out.persist();
- out.write(bytes('Z', 3), 0, 3);
-
- // Crash + restore from the checkpoint taken at persist()
- NativeS3RecoverableWriter writer2 =
- NativeS3RecoverableWriter.writer(s3, tmp.toString(),
MIN_PART_SIZE, 1);
- RecoverableFsDataOutputStream resumed = writer2.recover(r);
- resumed.write(bytes('F', 4), 0, 4); // legitimately appended after
recovery
- resumed.closeForCommit().commit();
-
- assertThat(s3.committedObjects.get(KEY))
- .as("post-persist 'Z' bytes are dropped; only persisted +
post-recovery bytes land")
- .containsExactly(concat(bytes('A', 10), bytes('E', 5),
bytes('F', 4)));
- }
-
- @Test
- void secondRecoveryAttemptCanStillReadSidePart() throws Exception {
- InMemoryNativeS3Operations s3 = new InMemoryNativeS3Operations();
- NativeS3RecoverableWriter writer1 =
- NativeS3RecoverableWriter.writer(s3, tmp.toString(),
MIN_PART_SIZE, 1);
-
- RecoverableFsDataOutputStream out = writer1.open(new Path("s3://" +
BUCKET + "/" + KEY));
- out.write(bytes('A', 10), 0, 10); // becomes part #1
- out.write(bytes('E', 5), 0, 5); // becomes the persisted side object
- NativeS3Recoverable r = (NativeS3Recoverable) out.persist();
- String sideObjectKey = r.incompleteObjectName();
- assertThat(sideObjectKey).isNotNull();
-
- // First recovery attempt successfully seeds the tail. We model this
attempt crashing before
- // commit, so we intentionally do NOT close/commit it (close() would
abort the shared MPU).
- NativeS3RecoverableWriter writer2 =
- NativeS3RecoverableWriter.writer(s3, tmp.toString(),
MIN_PART_SIZE, 1);
- RecoverableFsDataOutputStream firstAttempt = writer2.recover(r);
- assertThat(firstAttempt.getPos())
- .as("first attempt restores the 5-byte tail behind the 10
uploaded bytes")
- .isEqualTo(15L);
- assertThat(s3.storedObjects)
- .as("recover() must not delete the side object")
- .containsKey(sideObjectKey);
-
- // Second recovery attempt from the SAME checkpoint must still find
the side object.
- NativeS3RecoverableWriter writer3 =
- NativeS3RecoverableWriter.writer(s3, tmp.toString(),
MIN_PART_SIZE, 1);
- RecoverableFsDataOutputStream secondAttempt = writer3.recover(r);
- secondAttempt.write(bytes('F', 4), 0, 4);
- secondAttempt.closeForCommit().commit();
-
- assertThat(s3.committedObjects.get(KEY))
- .as("second attempt replays the tail and commits every
persisted byte")
- .containsExactly(concat(bytes('A', 10), bytes('E', 5),
bytes('F', 4)));
- }
-
- private static long countLocalFilesIn(java.nio.file.Path dir) throws
IOException {
- if (!java.nio.file.Files.isDirectory(dir)) {
- return 0;
- }
- try (java.util.stream.Stream<java.nio.file.Path> s =
java.nio.file.Files.list(dir)) {
- return s.count();
- }
- }
-
- private static byte[] bytes(char c, int n) {
- byte[] b = new byte[n];
- Arrays.fill(b, (byte) c);
- return b;
- }
-
- private static byte[] concat(byte[]... chunks) {
- int total = 0;
- for (byte[] c : chunks) {
- total += c.length;
- }
- byte[] out = new byte[total];
- int off = 0;
- for (byte[] c : chunks) {
- System.arraycopy(c, 0, out, off, c.length);
- off += c.length;
- }
- return out;
- }
-}
diff --git
a/flink-filesystems/flink-s3-fs-native/src/test/java/org/apache/flink/fs/s3native/writer/SeaweedFsNativeS3Operations.java
b/flink-filesystems/flink-s3-fs-native/src/test/java/org/apache/flink/fs/s3native/writer/SeaweedFsNativeS3Operations.java
new file mode 100644
index 00000000000..14d4d746545
--- /dev/null
+++
b/flink-filesystems/flink-s3-fs-native/src/test/java/org/apache/flink/fs/s3native/writer/SeaweedFsNativeS3Operations.java
@@ -0,0 +1,75 @@
+/*
+ * 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 software.amazon.awssdk.core.sync.RequestBody;
+import software.amazon.awssdk.services.s3.S3Client;
+import software.amazon.awssdk.services.s3.model.NoSuchKeyException;
+import software.amazon.awssdk.services.s3.model.S3Object;
+
+import java.util.List;
+import java.util.stream.Collectors;
+
+/**
+ * {@link NativeS3ObjectOperations} backed by a real (SeaweedFS) S3 endpoint,
plus byte-level
+ * helpers so writer/committer tests can inspect and tamper with the objects
that actually landed in
+ * S3.
+ *
+ * <p>A real endpoint reproduces the S3 behaviors the recovery tests actually
depend on: the 5 MiB
+ * multipart minimum part size, real {@code NoSuchKeyException} semantics on
{@code headObject}, and
+ * genuine network-level {@code GetObject}/{@code PutObject} responses.
+ */
+final class SeaweedFsNativeS3Operations extends NativeS3ObjectOperations {
+
+ private final S3Client client;
+ private final String bucket;
+
+ SeaweedFsNativeS3Operations(S3Client client, String bucket) {
+ super(client, bucket);
+ this.client = client;
+ this.bucket = bucket;
+ }
+
+ List<String> listKeys(String prefix) {
+ return client.listObjectsV2(b ->
b.bucket(bucket).prefix(prefix)).contents().stream()
+ .map(S3Object::key)
+ .collect(Collectors.toList());
+ }
+
+ byte[] readObject(String key) {
+ return client.getObjectAsBytes(b ->
b.bucket(bucket).key(key)).asByteArray();
+ }
+
+ void writeObject(String key, byte[] data) {
+ client.putObject(b -> b.bucket(bucket).key(key),
RequestBody.fromBytes(data));
+ }
+
+ void removeObject(String key) {
+ client.deleteObject(b -> b.bucket(bucket).key(key));
+ }
+
+ boolean objectExists(String key) {
+ try {
+ client.headObject(b -> b.bucket(bucket).key(key));
+ return true;
+ } catch (NoSuchKeyException e) {
+ return false;
+ }
+ }
+}
