gaborgsomogyi commented on code in PR #28867: URL: https://github.com/apache/flink/pull/28867#discussion_r3842444647
########## flink-filesystems/flink-s3-fs-native/src/test/java/org/apache/flink/fs/s3native/writer/NativeS3RecoverableWriterRecoveryITCase.java: ########## @@ -0,0 +1,238 @@ +/* + * 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.apache.flink.core.testutils.AllCallbackWrapper; +import org.apache.flink.core.testutils.TestContainerExtension; +import org.apache.flink.fs.s3native.SeaweedFsNativeS3TestContainer; + +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 ("_<name>.incomplete.<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 = 5 * 1024 * 1024; Review Comment: Same question as in other PRs, is this something special right? Not using the already available define is an issue but why don't we just set the config lower? ########## flink-filesystems/flink-s3-fs-native/src/test/java/org/apache/flink/fs/s3native/SeaweedFsNativeS3TestContainerTest.java: ########## @@ -0,0 +1,103 @@ +/* + * 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("s3.endpoint")).isTrue(); + assertThat(config.containsKey("s3.path-style-access")).isTrue(); + assertThat(config.containsKey("s3.access-key")).isTrue(); + assertThat(config.containsKey("s3.secret-key")).isTrue(); + assertThat(config.containsKey("s3.chunked-encoding.enabled")).isTrue(); + assertThat(config.containsKey("s3.checksum-validation.enabled")).isTrue(); Review Comment: Were they false before or what's the point here? Additionally why not using `NativeS3FileSystemFactory` defines like we do in the container class? ########## 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("secret", 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. + */ + 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); + } + + 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; + } + + String getS3UriForDefaultBucket() { + return "s3://" + defaultBucketName; + } + + List<S3Object> listObjects(String prefix) { + return getClient() + .listObjectsV2(b -> b.bucket(defaultBucketName).prefix(prefix)) + .contents(); + } + + String getObjectAsString(String key) { + return getClient() + .getObjectAsBytes(b -> b.bucket(defaultBucketName).key(key)) + .asUtf8String(); + } Review Comment: Why not public? We're in the same pkg but as long as it's not true it won't compile, right? Or are these no planned as API? ########## 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("secret", 10); Review Comment: Why not `secretKey` since we call it all the place like that? ########## flink-filesystems/flink-s3-fs-native/src/test/java/org/apache/flink/fs/s3native/writer/NativeS3RecoverableWriterRecoveryITCase.java: ########## @@ -0,0 +1,238 @@ +/* + * 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.apache.flink.core.testutils.AllCallbackWrapper; +import org.apache.flink.core.testutils.TestContainerExtension; +import org.apache.flink.fs.s3native.SeaweedFsNativeS3TestContainer; + +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 ("_<name>.incomplete.<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 = 5 * 1024 * 1024; + 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() { + final int lastSlash = key.lastIndexOf('/'); + final String parent = lastSlash < 0 ? "" : key.substring(0, lastSlash + 1); + final String name = lastSlash < 0 ? key : key.substring(lastSlash + 1); + return parent + "_" + name + ".incomplete."; + } + + @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 RecoverableWriter.ResumeRecoverable r = out.persist(); + assertThat(((NativeS3Recoverable) r).incompleteObjectName()) + .as("no tail => no side object") + .isNull(); + // incompletePrefix() is derived from this test's own UUID-based key, so this listing is + // scoped to this test instance and safe regardless of other tests' concurrent execution. + assertThat(s3.listKeys(incompletePrefix())).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())).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) { + 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; + } Review Comment: I see similar codes all around in the connector in different forms. Here `Arrays.fill`, later some payload generation, merge, etc... Can we just have a single way to do such re-occurring tasks? ########## 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("secret", 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. + */ + 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); + } + + 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; + } + + String getS3UriForDefaultBucket() { + return "s3://" + defaultBucketName; + } + + List<S3Object> listObjects(String prefix) { + return getClient() + .listObjectsV2(b -> b.bucket(defaultBucketName).prefix(prefix)) + .contents(); + } + + 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) { Review Comment: I can hardly believe we don't have such in Flink already ########## flink-filesystems/flink-s3-fs-native/src/test/java/org/apache/flink/fs/s3native/SeaweedFsNativeS3HAClusterExtension.java: ########## @@ -0,0 +1,82 @@ +/* + * 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.BeforeAllCallback; +import org.junit.jupiter.api.extension.ExtensionContext; + +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. + */ +final class SeaweedFsNativeS3HAClusterExtension implements BeforeAllCallback, AfterAllCallback { + + 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() { Review Comment: Public? I think this works only if same pkg. ########## flink-filesystems/flink-s3-fs-native/src/test/java/org/apache/flink/fs/s3native/SeaweedFsNativeS3TestContainerTest.java: ########## @@ -0,0 +1,103 @@ +/* + * 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("s3.endpoint")).isTrue(); + assertThat(config.containsKey("s3.path-style-access")).isTrue(); + assertThat(config.containsKey("s3.access-key")).isTrue(); + assertThat(config.containsKey("s3.secret-key")).isTrue(); + assertThat(config.containsKey("s3.chunked-encoding.enabled")).isTrue(); + assertThat(config.containsKey("s3.checksum-validation.enabled")).isTrue(); Review Comment: I've just double checked and we set 7 values in `setS3ConfigOptions` but we check here 6, why? ########## flink-filesystems/flink-s3-fs-native/src/test/java/org/apache/flink/fs/s3native/writer/NativeS3RecoverableWriterRecoveryITCase.java: ########## @@ -0,0 +1,238 @@ +/* + * 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.apache.flink.core.testutils.AllCallbackWrapper; +import org.apache.flink.core.testutils.TestContainerExtension; +import org.apache.flink.fs.s3native.SeaweedFsNativeS3TestContainer; + +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 ("_<name>.incomplete.<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 = 5 * 1024 * 1024; + 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() { + final int lastSlash = key.lastIndexOf('/'); + final String parent = lastSlash < 0 ? "" : key.substring(0, lastSlash + 1); + final String name = lastSlash < 0 ? key : key.substring(lastSlash + 1); + return parent + "_" + name + ".incomplete."; + } + + @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 RecoverableWriter.ResumeRecoverable r = out.persist(); + assertThat(((NativeS3Recoverable) r).incompleteObjectName()) + .as("no tail => no side object") + .isNull(); + // incompletePrefix() is derived from this test's own UUID-based key, so this listing is + // scoped to this test instance and safe regardless of other tests' concurrent execution. + assertThat(s3.listKeys(incompletePrefix())).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())).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) { + 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; + } Review Comment: BTW, don't we have something in the classpath already? ``` byte[] bytes = ArrayUtils.addAll(a, b); ``` ########## flink-filesystems/flink-s3-fs-native/src/test/java/org/apache/flink/fs/s3native/NativeS3FileSystemITCase.java: ########## @@ -0,0 +1,163 @@ +/* + * 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) { Review Comment: What would 0 mean? ########## flink-filesystems/flink-s3-fs-native/src/test/java/org/apache/flink/fs/s3native/HAJobRunOnNativeS3FileSystemITCase.java: ########## @@ -0,0 +1,114 @@ +/* + * 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 org.junit.jupiter.api.Order; +import org.junit.jupiter.api.extension.RegisterExtension; +import software.amazon.awssdk.services.s3.model.S3Object; + +import java.util.List; + +import static org.assertj.core.api.Assertions.assertThat; + +/** Runs {@link AbstractHAJobRunITCase} with HA data stored in SeaweedFS via the native S3 FS. */ +class HAJobRunOnNativeS3FileSystemITCase extends AbstractHAJobRunITCase { Review Comment: This is a copy-paste from `HAApplicationRunOnNativeS3FileSystemITCase`. Sincs we add this as net new code I don't agree that with the previous discussion that it's a maybe later thing. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
