Izeren commented on code in PR #28867: URL: https://github.com/apache/flink/pull/28867#discussion_r3803871549
########## flink-filesystems/flink-s3-fs-native/src/test/java/org/apache/flink/fs/s3native/writer/NativeS3RecoverableWriterRecoveryITCase.java: ########## @@ -0,0 +1,193 @@ +/* + * 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. + */ +class NativeS3RecoverableWriterRecoveryITCase { + + private static final int PART = 5 * 1024 * 1024; + private static final long MIN_PART_SIZE = PART; Review Comment: Do we need to have 2 variables for the same? Is that to use diffent types? ########## flink-filesystems/flink-s3-fs-native/src/test/java/org/apache/flink/fs/s3native/writer/NativeS3RecoverableWriterRecoveryITCase.java: ########## @@ -0,0 +1,193 @@ +/* + * 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 Review Comment: Is that not configurable for lower size testing? ########## flink-filesystems/flink-s3-fs-native/src/test/java/org/apache/flink/fs/s3native/writer/NativeS3RecoverableWriterRecoveryITCase.java: ########## @@ -0,0 +1,193 @@ +/* + * 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. + */ +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"; Review Comment: I think we should also test keys that contain `/` (aka nested paths) ########## flink-filesystems/flink-s3-fs-native/src/test/java/org/apache/flink/fs/s3native/writer/NativeS3RecoverableWriterRecoveryITCase.java: ########## @@ -0,0 +1,193 @@ +/* + * 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. + */ +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() { + int lastSlash = key.lastIndexOf('/'); Review Comment: !nit, variables that are not modified further should be final (here and everywhere else) ########## flink-filesystems/flink-s3-fs-native/src/test/java/org/apache/flink/fs/s3native/writer/SeaweedFsNativeS3Operations.java: ########## @@ -0,0 +1,71 @@ +/* + * 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. + */ +final class SeaweedFsNativeS3Operations extends NativeS3ObjectOperations { Review Comment: This implementation doesn't seem to to be much shorter than retired `inMemory` implementation. Are there additional benefits that we are bringing in here that were not feasible with inMemory? ########## flink-filesystems/flink-s3-fs-native/src/test/java/org/apache/flink/fs/s3native/HAJobRunOnNativeS3FileSystemITCase.java: ########## @@ -0,0 +1,129 @@ +/* + * 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.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.apache.flink.runtime.testutils.MiniClusterResourceConfiguration; +import org.apache.flink.test.junit5.MiniClusterExtension; + +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: Just a thought, it looks like we take two similar ITCase classes and register similar extensions with them. Would it make sense to consider intermediate composition layer that provides same extension over provided class and re-use it without code duplication across both? ########## flink-filesystems/flink-s3-fs-native/src/test/java/org/apache/flink/fs/s3native/HAJobRunOnNativeS3FileSystemITCase.java: ########## @@ -0,0 +1,129 @@ +/* + * 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.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.apache.flink.runtime.testutils.MiniClusterResourceConfiguration; +import org.apache.flink.test.junit5.MiniClusterExtension; + +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 { + + private static final String CLUSTER_ID = "test-cluster"; + private static final String JOB_RESULT_STORE_FOLDER = "jrs"; + + @RegisterExtension + @Order(2) Review Comment: Same question. Can we document why is this starting from 2? ########## flink-filesystems/flink-s3-fs-native/src/test/java/org/apache/flink/fs/s3native/writer/NativeS3RecoverableWriterRecoveryITCase.java: ########## @@ -0,0 +1,193 @@ +/* + * 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. + */ +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() { + int lastSlash = key.lastIndexOf('/'); + String parent = lastSlash < 0 ? "" : key.substring(0, lastSlash + 1); + String name = lastSlash < 0 ? key : key.substring(lastSlash + 1); + return parent + "_" + name + ".incomplete."; + } + + @Test + void recoverWithoutIncompleteTailStillWorks() throws Exception { + NativeS3RecoverableWriter writer1 = writer(); + + // Write exactly one full part => currentPartSize=0, no side object on persist. + RecoverableFsDataOutputStream out = writer1.open(targetPath()); + out.write(bytes('A', PART), 0, PART); + RecoverableWriter.ResumeRecoverable r = out.persist(); + assertThat(((NativeS3Recoverable) r).incompleteObjectName()) + .as("no tail => no side object") + .isNull(); + assertThat(s3.listKeys(incompletePrefix())).isEmpty(); + + NativeS3RecoverableWriter writer2 = writer(); + 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 recoverFailsCleanlyWhenSideObjectMissing() throws Exception { + NativeS3RecoverableWriter writer1 = writer(); + RecoverableFsDataOutputStream out = writer1.open(targetPath()); + out.write(bytes('A', PART), 0, PART); + out.write(bytes('E', 5), 0, 5); + NativeS3Recoverable r = (NativeS3Recoverable) out.persist(); + String sideObjectKey = r.incompleteObjectName(); + assertThat(sideObjectKey).isNotNull(); + + s3.removeObject(sideObjectKey); + long localFilesBefore = countLocalFilesIn(tmp); Review Comment: That assumes that tests can't be parallelised, we should either count using expected prefix or document that these tests have to be single threaded ########## flink-filesystems/flink-s3-fs-native/src/test/java/org/apache/flink/fs/s3native/writer/NativeS3RecoverableWriterRecoveryITCase.java: ########## @@ -0,0 +1,193 @@ +/* + * 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. + */ +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() { + int lastSlash = key.lastIndexOf('/'); + String parent = lastSlash < 0 ? "" : key.substring(0, lastSlash + 1); + String name = lastSlash < 0 ? key : key.substring(lastSlash + 1); + return parent + "_" + name + ".incomplete."; + } + + @Test + void recoverWithoutIncompleteTailStillWorks() throws Exception { + NativeS3RecoverableWriter writer1 = writer(); + + // Write exactly one full part => currentPartSize=0, no side object on persist. + RecoverableFsDataOutputStream out = writer1.open(targetPath()); + out.write(bytes('A', PART), 0, PART); + RecoverableWriter.ResumeRecoverable r = out.persist(); + assertThat(((NativeS3Recoverable) r).incompleteObjectName()) + .as("no tail => no side object") + .isNull(); + assertThat(s3.listKeys(incompletePrefix())).isEmpty(); + + NativeS3RecoverableWriter writer2 = writer(); + 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 recoverFailsCleanlyWhenSideObjectMissing() throws Exception { + NativeS3RecoverableWriter writer1 = writer(); + RecoverableFsDataOutputStream out = writer1.open(targetPath()); + out.write(bytes('A', PART), 0, PART); + out.write(bytes('E', 5), 0, 5); + NativeS3Recoverable r = (NativeS3Recoverable) out.persist(); + String sideObjectKey = r.incompleteObjectName(); + assertThat(sideObjectKey).isNotNull(); + + s3.removeObject(sideObjectKey); + long localFilesBefore = countLocalFilesIn(tmp); + NativeS3RecoverableWriter writer2 = writer(); + + assertThatThrownBy(() -> writer2.recover(r)) + .isInstanceOf(IOException.class) + .hasMessageContaining("Failed to get object"); + + assertThat(countLocalFilesIn(tmp)) + .as("partial download must be cleaned up on failure") + .isEqualTo(localFilesBefore); + } + + @Test + void recoverFailsCleanlyOnLengthMismatch() throws Exception { + NativeS3RecoverableWriter writer1 = writer(); + RecoverableFsDataOutputStream out = writer1.open(targetPath()); + out.write(bytes('A', PART), 0, PART); + 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.writeObject(sideObjectKey, bytes('X', 99)); + + long localFilesBefore = countLocalFilesIn(tmp); + NativeS3RecoverableWriter writer2 = writer(); + + assertThatThrownBy(() -> writer2.recover(r)) Review Comment: These asserts are very similar, we maybe we could extract assert helper or consider some level of test parametrisation ########## flink-filesystems/flink-s3-fs-native/src/test/java/org/apache/flink/fs/s3native/writer/NativeS3RecoverableWriterRecoveryITCase.java: ########## @@ -0,0 +1,193 @@ +/* + * 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. + */ +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() { + int lastSlash = key.lastIndexOf('/'); + String parent = lastSlash < 0 ? "" : key.substring(0, lastSlash + 1); + String name = lastSlash < 0 ? key : key.substring(lastSlash + 1); + return parent + "_" + name + ".incomplete."; + } + + @Test + void recoverWithoutIncompleteTailStillWorks() throws Exception { + NativeS3RecoverableWriter writer1 = writer(); + + // Write exactly one full part => currentPartSize=0, no side object on persist. + RecoverableFsDataOutputStream out = writer1.open(targetPath()); + out.write(bytes('A', PART), 0, PART); + RecoverableWriter.ResumeRecoverable r = out.persist(); + assertThat(((NativeS3Recoverable) r).incompleteObjectName()) + .as("no tail => no side object") + .isNull(); + assertThat(s3.listKeys(incompletePrefix())).isEmpty(); + + NativeS3RecoverableWriter writer2 = writer(); + 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 recoverFailsCleanlyWhenSideObjectMissing() throws Exception { + NativeS3RecoverableWriter writer1 = writer(); + RecoverableFsDataOutputStream out = writer1.open(targetPath()); + out.write(bytes('A', PART), 0, PART); + out.write(bytes('E', 5), 0, 5); + NativeS3Recoverable r = (NativeS3Recoverable) out.persist(); + String sideObjectKey = r.incompleteObjectName(); + assertThat(sideObjectKey).isNotNull(); + + s3.removeObject(sideObjectKey); + long localFilesBefore = countLocalFilesIn(tmp); + NativeS3RecoverableWriter writer2 = writer(); + + assertThatThrownBy(() -> writer2.recover(r)) + .isInstanceOf(IOException.class) + .hasMessageContaining("Failed to get object"); + + assertThat(countLocalFilesIn(tmp)) + .as("partial download must be cleaned up on failure") + .isEqualTo(localFilesBefore); + } + + @Test + void recoverFailsCleanlyOnLengthMismatch() throws Exception { + NativeS3RecoverableWriter writer1 = writer(); + RecoverableFsDataOutputStream out = writer1.open(targetPath()); + out.write(bytes('A', PART), 0, PART); + 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. Review Comment: Could we also explain scenarios when this can actually happen? ########## flink-filesystems/flink-s3-fs-native/src/test/java/org/apache/flink/fs/s3native/writer/NativeS3RecoverableWriterRecoveryITCase.java: ########## @@ -0,0 +1,193 @@ +/* + * 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. + */ +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() { + int lastSlash = key.lastIndexOf('/'); + String parent = lastSlash < 0 ? "" : key.substring(0, lastSlash + 1); + String name = lastSlash < 0 ? key : key.substring(lastSlash + 1); + return parent + "_" + name + ".incomplete."; + } + + @Test + void recoverWithoutIncompleteTailStillWorks() throws Exception { Review Comment: Would be nice to have some schematics of the parts/side objects before these tests start to set up a context around terminology and maybe setup visually the scenarios we are testing ########## flink-filesystems/flink-s3-fs-native/src/test/java/org/apache/flink/fs/s3native/HAApplicationRunOnNativeS3FileSystemITCase.java: ########## @@ -0,0 +1,136 @@ +/* + * 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.core.testutils.AllCallbackWrapper; +import org.apache.flink.core.testutils.TestContainerExtension; +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.apache.flink.runtime.testutils.MiniClusterResourceConfiguration; +import org.apache.flink.test.junit5.MiniClusterExtension; + +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 AbstractHAApplicationRunITCase} with HA data stored in SeaweedFS via the native S3 + * FS. + */ +class HAApplicationRunOnNativeS3FileSystemITCase extends AbstractHAApplicationRunITCase { + + private static final String CLUSTER_ID = "test-cluster"; + private static final String APPLICATION_RESULT_STORE_FOLDER = "ars"; + + @RegisterExtension + @Order(2) Review Comment: !nit, is it deliberate that we are starting with 2? ########## flink-filesystems/flink-s3-fs-native/src/test/java/org/apache/flink/fs/s3native/NativeS3FileSystemITCase.java: ########## @@ -0,0 +1,159 @@ +/* + * 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 testMkdirsDoesNotThrow() { + assertThatCode(() -> fs.mkdirs(path("mkdir-" + UUID.randomUUID()))) + .doesNotThrowAnyException(); + } + + @Test + void testRecoverableWriterMultipartCommit() throws Exception { + final Path file = path("recoverable-" + UUID.randomUUID() + ".bin"); + // > s3.upload.min.part.size (5 MiB) so the commit exercises a real multipart upload. + final byte[] data = payload(6 * 1024 * 1024); Review Comment: !nit, would be cleaner to use MIN_PART_SIZE + X pattern it is easier to understand and maintain (e.g. if min part size changes in the future) ########## flink-filesystems/flink-s3-fs-native/src/test/java/org/apache/flink/fs/s3native/NativeS3FileSystemITCase.java: ########## @@ -0,0 +1,159 @@ +/* + * 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 testMkdirsDoesNotThrow() { Review Comment: What is the contract are we trying to test here? Are we trying to test that mkdirs executes certain contract or that it can gracefully handle edge case. I would expect the test to either: `testMkdirsDoesX` (whatever it is supposed to do) `testMkdirsDoesntThrowWhenDirectoryAlreadyExists` (or any other edge case we want to verify). -- 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]
