| /* |
| * 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.hadoop.fs.s3a; |
| |
| import com.amazonaws.services.s3.model.ListObjectsV2Request; |
| import com.amazonaws.services.s3.model.ListObjectsV2Result; |
| |
| import org.apache.hadoop.conf.Configuration; |
| import org.apache.hadoop.fs.FSDataOutputStream; |
| import org.apache.hadoop.fs.FileStatus; |
| import org.apache.hadoop.fs.LocatedFileStatus; |
| import org.apache.hadoop.fs.Path; |
| import org.apache.hadoop.fs.RemoteIterator; |
| import org.apache.hadoop.fs.contract.AbstractFSContract; |
| import org.apache.hadoop.fs.contract.ContractTestUtils; |
| import org.apache.hadoop.fs.contract.s3a.S3AContract; |
| |
| import com.amazonaws.services.s3.model.S3ObjectSummary; |
| import com.google.common.collect.Lists; |
| import org.assertj.core.api.Assertions; |
| import org.junit.Assume; |
| import org.junit.Test; |
| |
| import java.io.FileNotFoundException; |
| import java.io.IOException; |
| import java.util.ArrayList; |
| import java.util.Collection; |
| import java.util.HashSet; |
| import java.util.List; |
| import java.util.stream.Collectors; |
| |
| import static org.apache.hadoop.fs.contract.ContractTestUtils.touch; |
| import static org.apache.hadoop.fs.contract.ContractTestUtils.writeTextFile; |
| import static org.apache.hadoop.fs.s3a.Constants.*; |
| import static org.apache.hadoop.fs.s3a.FailureInjectionPolicy.*; |
| import static org.apache.hadoop.fs.s3a.InconsistentAmazonS3Client.*; |
| import static org.apache.hadoop.test.LambdaTestUtils.intercept; |
| |
| /** |
| * Test S3Guard list consistency feature by injecting delayed listObjects() |
| * visibility via {@link InconsistentAmazonS3Client}. |
| * |
| * Tests here generally: |
| * 1. Use the inconsistency injection mentioned above. |
| * 2. Only run when S3Guard is enabled. |
| */ |
| public class ITestS3GuardListConsistency extends AbstractS3ATestBase { |
| |
| private Invoker invoker; |
| |
| @Override |
| public void setup() throws Exception { |
| super.setup(); |
| invoker = new Invoker(new S3ARetryPolicy(getConfiguration()), |
| Invoker.NO_OP |
| ); |
| Assume.assumeTrue("No metadata store in test filesystem", |
| getFileSystem().hasMetadataStore()); |
| } |
| |
| @Override |
| public void teardown() throws Exception { |
| clearInconsistency(getFileSystem()); |
| super.teardown(); |
| } |
| |
| @Override |
| protected AbstractFSContract createContract(Configuration conf) { |
| conf.setClass(S3_CLIENT_FACTORY_IMPL, InconsistentS3ClientFactory.class, |
| S3ClientFactory.class); |
| // Other configs would break test assumptions |
| conf.set(FAIL_INJECT_INCONSISTENCY_KEY, DEFAULT_DELAY_KEY_SUBSTRING); |
| conf.setFloat(FAIL_INJECT_INCONSISTENCY_PROBABILITY, 1.0f); |
| // this is a long value to guarantee that the inconsistency holds |
| // even over long-haul connections, and in the debugger too/ |
| conf.setLong(FAIL_INJECT_INCONSISTENCY_MSEC, 600_1000L); |
| return new S3AContract(conf); |
| } |
| |
| /** |
| * Helper function for other test cases: does a single rename operation and |
| * validates the aftermath. |
| * @param mkdirs Directories to create |
| * @param srcdirs Source paths for rename operation |
| * @param dstdirs Destination paths for rename operation |
| * @param yesdirs Files that must exist post-rename (e.g. srcdirs children) |
| * @param nodirs Files that must not exist post-rename (e.g. dstdirs children) |
| * @throws Exception |
| */ |
| private void doTestRenameSequence(Path[] mkdirs, Path[] srcdirs, |
| Path[] dstdirs, Path[] yesdirs, Path[] nodirs) throws Exception { |
| S3AFileSystem fs = getFileSystem(); |
| |
| |
| if (mkdirs != null) { |
| for (Path mkdir : mkdirs) { |
| assertTrue(fs.mkdirs(mkdir)); |
| } |
| clearInconsistency(fs); |
| } |
| |
| assertEquals("srcdirs and dstdirs must have equal length", |
| srcdirs.length, dstdirs.length); |
| for (int i = 0; i < srcdirs.length; i++) { |
| assertTrue("Rename returned false: " + srcdirs[i] + " -> " + dstdirs[i], |
| fs.rename(srcdirs[i], dstdirs[i])); |
| } |
| |
| for (Path yesdir : yesdirs) { |
| assertTrue("Path was supposed to exist: " + yesdir, fs.exists(yesdir)); |
| } |
| for (Path nodir : nodirs) { |
| assertFalse("Path is not supposed to exist: " + nodir, fs.exists(nodir)); |
| } |
| } |
| |
| /** |
| * Delete an array of paths; log exceptions. |
| * @param paths paths to delete |
| */ |
| private void deletePathsQuietly(Path...paths) { |
| for (Path dir : paths) { |
| try { |
| getFileSystem().delete(dir, true); |
| } catch (IOException e) { |
| LOG.info("Failed to delete {}: {}", dir, e.toString()); |
| LOG.debug("Delete failure:, e"); |
| } |
| } |
| } |
| |
| /** |
| * Tests that after renaming a directory, the original directory and its |
| * contents are indeed missing and the corresponding new paths are visible. |
| * @throws Exception |
| */ |
| @Test |
| public void testConsistentListAfterRename() throws Exception { |
| Path d1f = path("d1/f"); |
| Path d1f2 = path("d1/f-" + DEFAULT_DELAY_KEY_SUBSTRING); |
| Path[] mkdirs = {d1f, d1f2}; |
| Path d1 = path("d1"); |
| Path[] srcdirs = {d1}; |
| Path d2 = path("d2"); |
| Path[] dstdirs = {d2}; |
| Path d2f2 = path("d2/f-" + DEFAULT_DELAY_KEY_SUBSTRING); |
| Path[] yesdirs = {d2, path("d2/f"), d2f2}; |
| Path[] nodirs = { |
| d1, d1f, d1f2}; |
| try { |
| doTestRenameSequence(mkdirs, srcdirs, dstdirs, yesdirs, nodirs); |
| } finally { |
| clearInconsistency(getFileSystem()); |
| deletePathsQuietly(d1, d2, d1f, d1f2, d2f2); |
| } |
| } |
| |
| /** |
| * Tests a circular sequence of renames to verify that overwriting recently |
| * deleted files and reading recently created files from rename operations |
| * works as expected. |
| * @throws Exception |
| */ |
| @Test |
| public void testRollingRenames() throws Exception { |
| Path[] dir0 = {path("rolling/1")}; |
| Path[] dir1 = {path("rolling/2")}; |
| Path[] dir2 = {path("rolling/3")}; |
| // These sets have to be in reverse order compared to the movement |
| Path[] setA = {dir1[0], dir0[0]}; |
| Path[] setB = {dir2[0], dir1[0]}; |
| Path[] setC = {dir0[0], dir2[0]}; |
| |
| try { |
| for(int i = 0; i < 2; i++) { |
| Path[] firstSet = i == 0 ? setA : null; |
| doTestRenameSequence(firstSet, setA, setB, setB, dir0); |
| doTestRenameSequence(null, setB, setC, setC, dir1); |
| doTestRenameSequence(null, setC, setA, setA, dir2); |
| } |
| |
| S3AFileSystem fs = getFileSystem(); |
| assertFalse("Renaming deleted file should have failed", |
| fs.rename(dir2[0], dir1[0])); |
| assertTrue("Renaming over existing file should have succeeded", |
| fs.rename(dir1[0], dir0[0])); |
| } finally { |
| clearInconsistency(getFileSystem()); |
| deletePathsQuietly(dir0[0], dir1[0], dir2[0]); |
| } |
| } |
| |
| /** |
| * Tests that deleted files immediately stop manifesting in list operations |
| * even when the effect in S3 is delayed. |
| * @throws Exception |
| */ |
| @Test |
| public void testConsistentListAfterDelete() throws Exception { |
| S3AFileSystem fs = getFileSystem(); |
| |
| // Any S3 keys that contain DELAY_KEY_SUBSTRING will be delayed |
| // in listObjects() results via InconsistentS3Client |
| Path inconsistentPath = |
| path("a/b/dir3-" + DEFAULT_DELAY_KEY_SUBSTRING); |
| |
| Path dir1 = path("a/b/dir1"); |
| Path dir2 = path("a/b/dir2"); |
| Path[] testDirs = { |
| dir1, |
| dir2, |
| inconsistentPath}; |
| |
| for (Path path : testDirs) { |
| assertTrue("Can't create directory: " + path, fs.mkdirs(path)); |
| } |
| clearInconsistency(fs); |
| for (Path path : testDirs) { |
| assertTrue("Can't delete path: " + path, fs.delete(path, false)); |
| } |
| |
| FileStatus[] paths = fs.listStatus(path("a/b/")); |
| List<Path> list = new ArrayList<>(); |
| for (FileStatus fileState : paths) { |
| list.add(fileState.getPath()); |
| } |
| Assertions.assertThat(list) |
| .describedAs("Expected deleted files to be excluded") |
| .doesNotContain(dir1) |
| .doesNotContain(dir2) |
| .doesNotContain(inconsistentPath); |
| } |
| |
| /** |
| * Tests that rename immediately after files in the source directory are |
| * deleted results in exactly the correct set of destination files and none |
| * of the source files. |
| * @throws Exception |
| */ |
| @Test |
| public void testConsistentRenameAfterDelete() throws Exception { |
| S3AFileSystem fs = getFileSystem(); |
| |
| // Any S3 keys that contain DELAY_KEY_SUBSTRING will be delayed |
| // in listObjects() results via InconsistentS3Client |
| Path inconsistentPath = |
| path("a/b/dir3-" + DEFAULT_DELAY_KEY_SUBSTRING); |
| |
| Path[] testDirs = {path("a/b/dir1"), |
| path("a/b/dir2"), |
| inconsistentPath}; |
| |
| for (Path path : testDirs) { |
| assertTrue(fs.mkdirs(path)); |
| } |
| clearInconsistency(fs); |
| assertTrue(fs.delete(testDirs[1], false)); |
| assertTrue(fs.delete(testDirs[2], false)); |
| |
| ContractTestUtils.rename(fs, path("a"), path("a3")); |
| ContractTestUtils.assertPathsDoNotExist(fs, |
| "Source paths shouldn't exist post rename operation", |
| testDirs[0], testDirs[1], testDirs[2]); |
| FileStatus[] paths = fs.listStatus(path("a3/b")); |
| List<Path> list = new ArrayList<>(); |
| for (FileStatus fileState : paths) { |
| list.add(fileState.getPath()); |
| } |
| Assertions.assertThat(list) |
| .contains(path("a3/b/dir1")) |
| .doesNotContain(path("a3/b/dir2")) |
| .doesNotContain(path("a3/b/dir3-" + |
| DEFAULT_DELAY_KEY_SUBSTRING)); |
| |
| intercept(FileNotFoundException.class, "", |
| "Recently renamed dir should not be visible", |
| () -> S3AUtils.mapLocatedFiles( |
| fs.listFilesAndEmptyDirectories(path("a"), true), |
| FileStatus::getPath)); |
| } |
| |
| @Test |
| public void testConsistentListStatusAfterPut() throws Exception { |
| |
| S3AFileSystem fs = getFileSystem(); |
| |
| // Any S3 keys that contain DELAY_KEY_SUBSTRING will be delayed |
| // in listObjects() results via InconsistentS3Client |
| Path inconsistentPath = |
| path("a/b/dir3-" + DEFAULT_DELAY_KEY_SUBSTRING); |
| |
| Path[] testDirs = {path("a/b/dir1"), |
| path("a/b/dir2"), |
| inconsistentPath}; |
| |
| for (Path path : testDirs) { |
| assertTrue(fs.mkdirs(path)); |
| } |
| |
| FileStatus[] paths = fs.listStatus(path("a/b/")); |
| List<Path> list = new ArrayList<>(); |
| for (FileStatus fileState : paths) { |
| list.add(fileState.getPath()); |
| } |
| Assertions.assertThat(list) |
| .contains(path("a/b/dir1")) |
| .contains(path("a/b/dir2")) |
| .contains(inconsistentPath); |
| } |
| |
| /** |
| * Similar to {@link #testConsistentListStatusAfterPut()}, this tests that the |
| * FS listLocatedStatus() call will return consistent list. |
| */ |
| @Test |
| public void testConsistentListLocatedStatusAfterPut() throws Exception { |
| final S3AFileSystem fs = getFileSystem(); |
| String rootDir = "doTestConsistentListLocatedStatusAfterPut"; |
| fs.mkdirs(path(rootDir)); |
| |
| final int[] numOfPaths = {0, 1, 5}; |
| for (int normalPathNum : numOfPaths) { |
| for (int delayedPathNum : new int[] {0, 2}) { |
| LOG.info("Testing with normalPathNum={}, delayedPathNum={}", |
| normalPathNum, delayedPathNum); |
| doTestConsistentListLocatedStatusAfterPut(fs, rootDir, normalPathNum, |
| delayedPathNum); |
| } |
| } |
| } |
| |
| /** |
| * Helper method to implement the tests of consistent listLocatedStatus(). |
| * @param fs The S3 file system from contract |
| * @param normalPathNum number paths listed directly from S3 without delaying |
| * @param delayedPathNum number paths listed with delaying |
| * @throws Exception |
| */ |
| private void doTestConsistentListLocatedStatusAfterPut(S3AFileSystem fs, |
| String rootDir, int normalPathNum, int delayedPathNum) throws Exception { |
| final List<Path> testDirs = new ArrayList<>(normalPathNum + delayedPathNum); |
| int index = 0; |
| for (; index < normalPathNum; index++) { |
| testDirs.add(path(rootDir + "/dir-" + |
| index)); |
| } |
| for (; index < normalPathNum + delayedPathNum; index++) { |
| // Any S3 keys that contain DELAY_KEY_SUBSTRING will be delayed |
| // in listObjects() results via InconsistentS3Client |
| testDirs.add(path(rootDir + "/dir-" + index + |
| DEFAULT_DELAY_KEY_SUBSTRING)); |
| } |
| |
| for (Path path : testDirs) { |
| // delete the old test path (if any) so that when we call mkdirs() later, |
| // the to delay directories will be tracked via putObject() request. |
| fs.delete(path, true); |
| assertTrue(fs.mkdirs(path)); |
| } |
| |
| // this should return the union data from S3 and MetadataStore |
| final RemoteIterator<LocatedFileStatus> statusIterator = |
| fs.listLocatedStatus(path(rootDir + "/")); |
| List<Path> list = new ArrayList<>(); |
| for (; statusIterator.hasNext();) { |
| list.add(statusIterator.next().getPath()); |
| } |
| |
| // This should fail without S3Guard, and succeed with it because part of the |
| // children under test path are delaying visibility |
| for (Path path : testDirs) { |
| assertTrue("listLocatedStatus should list " + path, list.contains(path)); |
| } |
| } |
| |
| /** |
| * Tests that the S3AFS listFiles() call will return consistent file list. |
| */ |
| @Test |
| public void testConsistentListFiles() throws Exception { |
| final S3AFileSystem fs = getFileSystem(); |
| |
| final int[] numOfPaths = {0, 2}; |
| for (int dirNum : numOfPaths) { |
| for (int normalFile : numOfPaths) { |
| for (int delayedFile : new int[] {0, 1}) { |
| for (boolean recursive : new boolean[] {true, false}) { |
| doTestListFiles(fs, dirNum, normalFile, delayedFile, recursive); |
| } |
| } |
| } |
| } |
| } |
| |
| /** |
| * Helper method to implement the tests of consistent listFiles(). |
| * |
| * The file structure has dirNum subdirectories, and each directory (including |
| * the test base directory itself) has normalFileNum normal files and |
| * delayedFileNum delayed files. |
| * |
| * @param fs The S3 file system from contract |
| * @param dirNum number of subdirectories |
| * @param normalFileNum number files in each directory without delay to list |
| * @param delayedFileNum number files in each directory with delay to list |
| * @param recursive listFiles recursively if true |
| * @throws Exception if any unexpected error |
| */ |
| private void doTestListFiles(S3AFileSystem fs, int dirNum, int normalFileNum, |
| int delayedFileNum, boolean recursive) throws Exception { |
| describe("Testing dirNum=%d, normalFile=%d, delayedFile=%d, " |
| + "recursive=%s", dirNum, normalFileNum, delayedFileNum, recursive); |
| final Path baseTestDir = path("doTestListFiles-" + dirNum + "-" |
| + normalFileNum + "-" + delayedFileNum + "-" + recursive); |
| // delete the old test path (if any) so that when we call mkdirs() later, |
| // the to delay sub directories will be tracked via putObject() request. |
| fs.delete(baseTestDir, true); |
| |
| // make subdirectories (if any) |
| final List<Path> testDirs = new ArrayList<>(dirNum + 1); |
| assertTrue(fs.mkdirs(baseTestDir)); |
| testDirs.add(baseTestDir); |
| for (int i = 0; i < dirNum; i++) { |
| final Path subdir = path(baseTestDir + "/dir-" + i); |
| assertTrue(fs.mkdirs(subdir)); |
| testDirs.add(subdir); |
| } |
| |
| final Collection<String> fileNames |
| = new ArrayList<>(normalFileNum + delayedFileNum); |
| int index = 0; |
| for (; index < normalFileNum; index++) { |
| fileNames.add("file-" + index); |
| } |
| for (; index < normalFileNum + delayedFileNum; index++) { |
| // Any S3 keys that contain DELAY_KEY_SUBSTRING will be delayed |
| // in listObjects() results via InconsistentS3Client |
| fileNames.add("file-" + index + "-" + DEFAULT_DELAY_KEY_SUBSTRING); |
| } |
| |
| int filesAndEmptyDirectories = 0; |
| |
| // create files under each test directory |
| for (Path dir : testDirs) { |
| for (String fileName : fileNames) { |
| writeTextFile(fs, new Path(dir, fileName), "I, " + fileName, false); |
| filesAndEmptyDirectories++; |
| } |
| } |
| |
| // this should return the union data from S3 and MetadataStore |
| final RemoteIterator<LocatedFileStatus> statusIterator |
| = fs.listFiles(baseTestDir, recursive); |
| final Collection<Path> listedFiles = new HashSet<>(); |
| for (; statusIterator.hasNext();) { |
| final FileStatus status = statusIterator.next(); |
| assertTrue("FileStatus " + status + " is not a file!", status.isFile()); |
| listedFiles.add(status.getPath()); |
| } |
| LOG.info("S3AFileSystem::listFiles('{}', {}) -> {}", |
| baseTestDir, recursive, listedFiles); |
| |
| // This should fail without S3Guard, and succeed with it because part of the |
| // files to list are delaying visibility |
| if (!recursive) { |
| // in this case only the top level files are listed |
| verifyFileIsListed(listedFiles, baseTestDir, fileNames); |
| assertEquals("Unexpected number of files returned by listFiles() call", |
| normalFileNum + delayedFileNum, listedFiles.size()); |
| } else { |
| for (Path dir : testDirs) { |
| verifyFileIsListed(listedFiles, dir, fileNames); |
| } |
| assertEquals("Unexpected number of files returned by listFiles() call", |
| filesAndEmptyDirectories, |
| listedFiles.size()); |
| } |
| } |
| |
| private static void verifyFileIsListed(Collection<Path> listedFiles, |
| Path currentDir, Collection<String> fileNames) { |
| for (String fileName : fileNames) { |
| final Path file = new Path(currentDir, fileName); |
| assertTrue(file + " should have been listed", listedFiles.contains(file)); |
| } |
| } |
| |
| @Test |
| public void testCommitByRenameOperations() throws Throwable { |
| S3AFileSystem fs = getFileSystem(); |
| Path work = path("test-commit-by-rename-" + DEFAULT_DELAY_KEY_SUBSTRING); |
| Path task00 = new Path(work, "task00"); |
| fs.mkdirs(task00); |
| String name = "part-00"; |
| try (FSDataOutputStream out = |
| fs.create(new Path(task00, name), false)) { |
| out.writeChars("hello"); |
| } |
| for (FileStatus stat : fs.listStatus(task00)) { |
| fs.rename(stat.getPath(), work); |
| } |
| List<FileStatus> files = new ArrayList<>(2); |
| for (FileStatus stat : fs.listStatus(work)) { |
| if (stat.isFile()) { |
| files.add(stat); |
| } |
| } |
| assertFalse("renamed file " + name + " not found in " + work, |
| files.isEmpty()); |
| assertEquals("more files found than expected in " + work |
| + " " + ls(work), 1, files.size()); |
| FileStatus status = files.get(0); |
| assertEquals("Wrong filename in " + status, |
| name, status.getPath().getName()); |
| } |
| |
| @Test |
| public void testInconsistentS3ClientDeletes() throws Throwable { |
| describe("Verify that delete adds tombstones which block entries" |
| + " returned in (inconsistent) listings"); |
| // Test only implemented for v2 S3 list API |
| assumeV2ListAPI(); |
| |
| S3AFileSystem fs = getFileSystem(); |
| Path root = path("testInconsistentS3ClientDeletes-" |
| + DEFAULT_DELAY_KEY_SUBSTRING); |
| for (int i = 0; i < 3; i++) { |
| fs.mkdirs(new Path(root, "dir-" + i)); |
| touch(fs, new Path(root, "file-" + i)); |
| for (int j = 0; j < 3; j++) { |
| touch(fs, new Path(new Path(root, "dir-" + i), "file-" + i + "-" + j)); |
| } |
| } |
| clearInconsistency(fs); |
| |
| String key = fs.pathToKey(root) + "/"; |
| |
| LOG.info("Listing objects before executing delete()"); |
| ListObjectsV2Result preDeleteDelimited = listObjectsV2(fs, key, "/"); |
| ListObjectsV2Result preDeleteUndelimited = listObjectsV2(fs, key, null); |
| |
| LOG.info("Deleting the directory {}", root); |
| fs.delete(root, true); |
| LOG.info("Delete completed; listing results which must exclude deleted" |
| + " paths"); |
| |
| ListObjectsV2Result postDeleteDelimited = listObjectsV2(fs, key, "/"); |
| boolean stripTombstones = false; |
| assertObjectSummariesEqual( |
| "InconsistentAmazonS3Client added back objects incorrectly " + |
| "in a non-recursive listing", |
| preDeleteDelimited, postDeleteDelimited, |
| stripTombstones); |
| |
| assertListSizeEqual("InconsistentAmazonS3Client added back prefixes incorrectly " + |
| "in a non-recursive listing", |
| preDeleteDelimited.getCommonPrefixes(), |
| postDeleteDelimited.getCommonPrefixes()); |
| LOG.info("Executing Deep listing"); |
| ListObjectsV2Result postDeleteUndelimited = listObjectsV2(fs, key, null); |
| assertObjectSummariesEqual("InconsistentAmazonS3Client added back objects" |
| + " incorrectly in a recursive listing", |
| preDeleteUndelimited, postDeleteUndelimited, |
| stripTombstones); |
| |
| assertListSizeEqual("InconsistentAmazonS3Client added back prefixes incorrectly " + |
| "in a recursive listing", |
| preDeleteUndelimited.getCommonPrefixes(), |
| postDeleteUndelimited.getCommonPrefixes() |
| ); |
| } |
| |
| private void assertObjectSummariesEqual(final String message, |
| final ListObjectsV2Result expected, |
| final ListObjectsV2Result actual, |
| final boolean stripTombstones) { |
| assertCollectionsEqual( |
| message, |
| stringify(expected.getObjectSummaries(), stripTombstones), |
| stringify(actual.getObjectSummaries(), stripTombstones)); |
| } |
| |
| List<String> stringify(List<S3ObjectSummary> objects, |
| boolean stripTombstones) { |
| return objects.stream() |
| .filter(s -> !stripTombstones || !(s.getKey().endsWith("/"))) |
| .map(s -> s.getKey()) |
| .collect(Collectors.toList()); |
| } |
| |
| /** |
| * Require the v2 S3 list API. |
| */ |
| protected void assumeV2ListAPI() { |
| Assume.assumeTrue(getConfiguration() |
| .getInt(LIST_VERSION, DEFAULT_LIST_VERSION) == 2); |
| } |
| |
| /** |
| * Verify that delayed S3 listings doesn't stop the FS from deleting |
| * a directory tree. This has not always been the case; this test |
| * verifies the fix and prevents regression. |
| */ |
| @Test |
| public void testDeleteUsesS3Guard() throws Throwable { |
| describe("Verify that delete() uses S3Guard to get a consistent" |
| + " listing of its directory structure"); |
| assumeV2ListAPI(); |
| S3AFileSystem fs = getFileSystem(); |
| Path root = path( |
| "testDeleteUsesS3Guard-" + DEFAULT_DELAY_KEY_SUBSTRING); |
| for (int i = 0; i < 3; i++) { |
| Path path = new Path(root, "file-" + i); |
| touch(fs, path); |
| } |
| // we now expect the listing to miss these |
| String key = fs.pathToKey(root) + "/"; |
| |
| // verify that the inconsistent listing does not show these |
| LOG.info("Listing objects before executing delete()"); |
| List<Path> preDeletePaths = objectsToPaths(listObjectsV2(fs, key, null)); |
| Assertions.assertThat(preDeletePaths) |
| .isEmpty(); |
| // do the delete |
| fs.delete(root, true); |
| |
| // now go through every file and verify that it is not there. |
| // if you comment out the delete above and run this test case, |
| // the assertion will fail; this is how the validity of the assertions |
| // were verified. |
| clearInconsistency(fs); |
| List<Path> postDeletePaths = |
| objectsToPaths(listObjectsV2(fs, key, null)); |
| Assertions.assertThat(postDeletePaths) |
| .isEmpty(); |
| } |
| |
| private List<Path> objectsToPaths(ListObjectsV2Result r) { |
| S3AFileSystem fs = getFileSystem(); |
| return r.getObjectSummaries().stream() |
| .map(s -> fs.keyToQualifiedPath(s.getKey())) |
| .collect(Collectors.toList()); |
| } |
| |
| /** |
| * Tests that the file's eTag and versionId are preserved in recursive |
| * listings. |
| */ |
| @Test |
| public void testListingReturnsVersionMetadata() throws Throwable { |
| S3AFileSystem fs = getFileSystem(); |
| |
| // write simple file |
| Path parent = path(getMethodName()); |
| Path file = new Path(parent, "file1"); |
| try (FSDataOutputStream outputStream = fs.create(file)) { |
| outputStream.writeChars("hello"); |
| } |
| |
| // get individual file status |
| FileStatus[] fileStatuses = fs.listStatus(file); |
| assertEquals(1, fileStatuses.length); |
| S3AFileStatus status = (S3AFileStatus) fileStatuses[0]; |
| String eTag = status.getETag(); |
| assertNotNull("Etag in " + eTag, eTag); |
| String versionId = status.getVersionId(); |
| |
| // get status through recursive directory listing |
| RemoteIterator<LocatedFileStatus> filesIterator = fs.listFiles( |
| parent, true); |
| List<LocatedFileStatus> files = Lists.newArrayList(); |
| while (filesIterator.hasNext()) { |
| files.add(filesIterator.next()); |
| } |
| Assertions.assertThat(files) |
| .hasSize(1); |
| |
| // ensure eTag and versionId are preserved in directory listing |
| S3ALocatedFileStatus locatedFileStatus = |
| (S3ALocatedFileStatus) files.get(0); |
| assertEquals("etag of " + locatedFileStatus, |
| eTag, locatedFileStatus.getETag()); |
| assertEquals("versionID of " + locatedFileStatus, |
| versionId, locatedFileStatus.getVersionId()); |
| } |
| |
| /** |
| * Assert that the two collections match using |
| * object equality of the elements within. |
| * @param message text for the assertion |
| * @param expected expected list |
| * @param actual actual list |
| * @param <T> type of list |
| */ |
| private <T> void assertCollectionsEqual(String message, |
| Collection<T> expected, |
| Collection<T> actual) { |
| Assertions.assertThat(actual) |
| .describedAs(message) |
| .containsExactlyInAnyOrderElementsOf(expected); |
| } |
| |
| /** |
| * Assert that the two list sizes match; failure message includes the lists. |
| * @param message text for the assertion |
| * @param expected expected list |
| * @param actual actual list |
| * @param <T> type of list |
| */ |
| private <T> void assertListSizeEqual(String message, |
| List<T> expected, |
| List<T> actual) { |
| String leftContents = expected.stream() |
| .map(n -> n.toString()) |
| .collect(Collectors.joining("\n")); |
| String rightContents = actual.stream() |
| .map(n -> n.toString()) |
| .collect(Collectors.joining("\n")); |
| String summary = "\nExpected:" + leftContents |
| + "\n-----------\n" |
| + "Actual:" + rightContents |
| + "\n-----------\n"; |
| |
| if (expected.size() != actual.size()) { |
| LOG.error(message + summary); |
| } |
| assertEquals(message + summary, expected.size(), actual.size()); |
| } |
| |
| /** |
| * Retrying v2 list directly through the s3 client. |
| * @param fs filesystem |
| * @param key key to list under |
| * @param delimiter any delimiter |
| * @return the listing |
| * @throws IOException on error |
| */ |
| @Retries.RetryRaw |
| private ListObjectsV2Result listObjectsV2(S3AFileSystem fs, |
| String key, String delimiter) throws IOException { |
| ListObjectsV2Request k = fs.createListObjectsRequest(key, delimiter) |
| .getV2(); |
| return invoker.retryUntranslated("list", true, |
| () -> { |
| return fs.getAmazonS3ClientForTesting("list").listObjectsV2(k); |
| }); |
| } |
| |
| } |
