+ * Set to true when a completed MPU commit consumes the ID so it is no + * longer visible in list operations; and abort reports {@code NoSuchUploadException}. + *
+ * This will change assertions in relevant tests. + *
+ * Can be set as a per-bucket setting; test runner will pick this up. + */ + String MULTIPART_COMMIT_CONSUMES_UPLOAD_ID = + "fs.s3a.ext.test.multipart.commit.consumes.upload.id"; + + /** + * Default value of {@link #MULTIPART_COMMIT_CONSUMES_UPLOAD_ID}: {@value}. + */ + boolean DEFAULT_MULTIPART_COMMIT_CONSUMES_UPLOAD_ID = false; } diff --git a/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/S3ATestUtils.java b/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/S3ATestUtils.java index d74c84463252e..02a7419fa5a1b 100644 --- a/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/S3ATestUtils.java +++ b/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/S3ATestUtils.java @@ -528,6 +528,32 @@ public static void skipIfNotEnabled(final Configuration configuration, } } + /** + * Skip a test suite/case if a configuration option is true. + * @param configuration configuration to probe + * @param key key to resolve + * @param defVal default value. + * @param message assertion text + */ + public static void skipIfEnabled(final Configuration configuration, + final String key, + final boolean defVal, + final String message) { + if (!configuration.getBoolean(key, defVal)) { + skip(message); + } + } + + /** + * Require multipart uploads; skip tests if not enabled in the configuration. + * @param conf filesystem configuration. + */ + public static void assumeMultipartUploads(Configuration conf) { + skipIfNotEnabled(conf, + MULTIPART_UPLOADS_ENABLED, + "Store has disabled multipart uploads; skipping tests"); + } + /** * Skip a test if storage class tests are disabled, * or the bucket is an S3Express bucket. diff --git a/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/TestS3AExceptionTranslation.java b/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/TestS3AExceptionTranslation.java index 6b894a6813704..edb6f277cda35 100644 --- a/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/TestS3AExceptionTranslation.java +++ b/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/TestS3AExceptionTranslation.java @@ -447,6 +447,19 @@ public void testOpenSSLErrorRetry() throws Throwable { sdkClientException(WFOPENSSL_0035_STREAM_IS_CLOSED, null)))); } + @Test + public void testS3ExpressPreconditionFailure() throws Throwable { + AwsServiceException ase = AwsServiceException.builder() + .message("unwind") + .statusCode(SC_200_OK) + .awsErrorDetails(AwsErrorDetails.builder() + .errorCode(PRECONDITION_FAILED) + .build()) + .build(); + verifyExceptionClass(RemoteFileChangedException.class, + translateException("commit", "/path", ase)); + } + /** * Create a shaded NoHttpResponseException. * @return an exception. diff --git a/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/auth/ITestAssumeRole.java b/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/auth/ITestAssumeRole.java index 592529b553d24..2680e1a712a8e 100644 --- a/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/auth/ITestAssumeRole.java +++ b/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/auth/ITestAssumeRole.java @@ -84,6 +84,10 @@ /** * Tests use of assumed roles. * Only run if an assumed role is provided. + *
+ * S3Express buckets only support access restrictions at the bucket level,
+ * rather than at paths underneath.
+ * All partial permission tests are disabled.
*/
@SuppressWarnings("ThrowableNotThrown")
public class ITestAssumeRole extends AbstractS3ATestBase {
@@ -197,9 +201,6 @@ protected Configuration createValidRoleConf() throws JsonProcessingException {
conf.set(ASSUMED_ROLE_ARN, roleARN);
conf.set(ASSUMED_ROLE_SESSION_NAME, "valid");
conf.set(ASSUMED_ROLE_SESSION_DURATION, "45m");
- // disable create session so there's no need to
- // add a role policy for it.
- disableCreateSession(conf);
bindRolePolicy(conf, RESTRICTED_POLICY);
return conf;
@@ -458,12 +459,15 @@ public void testAssumeRolePoliciesOverrideRolePerms() throws Throwable {
public void testReadOnlyOperations() throws Throwable {
describe("Restrict role to read only");
+ skipIfS3ExpressBucket(getConfiguration());
Configuration conf = createAssumedRoleConfig();
bindRolePolicy(conf,
policy(
statement(false, S3_ALL_BUCKETS, S3_PATH_WRITE_OPERATIONS),
- STATEMENT_ALL_S3, STATEMENT_ALLOW_KMS_RW));
+ STATEMENT_ALL_S3,
+ STATEMENT_S3EXPRESS,
+ STATEMENT_ALLOW_KMS_RW));
Path path = methodPath();
roleFS = (S3AFileSystem) path.getFileSystem(conf);
// list the root path, expect happy
@@ -501,6 +505,7 @@ public void testRestrictedWriteSubdir() throws Throwable {
describe("Attempt writing to paths where a role only has"
+ " write access to a subdir of the bucket");
+ skipIfS3ExpressBucket(getConfiguration());
Path restrictedDir = methodPath();
Path child = new Path(restrictedDir, "child");
// the full FS
@@ -563,6 +568,7 @@ public void testAssumedRoleRetryHandler() throws Throwable {
@Test
public void testRestrictedCommitActions() throws Throwable {
describe("Attempt commit operations against a path with restricted rights");
+ skipIfS3ExpressBucket(getConfiguration());
Configuration conf = createAssumedRoleConfig();
final int uploadPartSize = 5 * 1024 * 1024;
@@ -700,12 +706,14 @@ public void writeCSVData(final File localSrc) throws IOException {
@Test
public void testPartialDelete() throws Throwable {
describe("delete with part of the child tree read only; multidelete");
+ skipIfS3ExpressBucket(getConfiguration());
executePartialDelete(createAssumedRoleConfig(), false);
}
@Test
public void testPartialDeleteSingleDelete() throws Throwable {
describe("delete with part of the child tree read only");
+ skipIfS3ExpressBucket(getConfiguration());
executePartialDelete(createAssumedRoleConfig(), true);
}
@@ -718,6 +726,7 @@ public void testBulkDeleteOnReadOnlyAccess() throws Throwable {
@Test
public void testBulkDeleteWithReadWriteAccess() throws Throwable {
describe("Bulk delete with read write access");
+ skipIfS3ExpressBucket(getConfiguration());
executeBulkDeleteOnSomeReadOnlyFiles(createAssumedRoleConfig());
}
@@ -807,6 +816,7 @@ private static void bindReadOnlyRolePolicy(Configuration assumedRoleConfig,
throws JsonProcessingException {
bindRolePolicyStatements(assumedRoleConfig, STATEMENT_ALLOW_KMS_RW,
statement(true, S3_ALL_BUCKETS, S3_ALL_OPERATIONS),
+ STATEMENT_S3EXPRESS,
new Statement(Effects.Deny)
.addActions(S3_PATH_WRITE_OPERATIONS)
.addResources(directory(readOnlyDir))
diff --git a/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/auth/ITestAssumedRoleCommitOperations.java b/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/auth/ITestAssumedRoleCommitOperations.java
index 0cb42b0f31096..67df9a77df1a0 100644
--- a/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/auth/ITestAssumedRoleCommitOperations.java
+++ b/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/auth/ITestAssumedRoleCommitOperations.java
@@ -67,7 +67,9 @@ public void setup() throws Exception {
restrictedDir = super.path("restricted");
Configuration conf = newAssumedRoleConfig(getConfiguration(),
getAssumedRoleARN());
- bindRolePolicyStatements(conf, STATEMENT_ALLOW_KMS_RW,
+ bindRolePolicyStatements(conf,
+ STATEMENT_ALLOW_KMS_RW,
+ STATEMENT_S3EXPRESS,
statement(true, S3_ALL_BUCKETS, S3_BUCKET_READ_OPERATIONS),
new RoleModel.Statement(RoleModel.Effects.Allow)
.addActions(S3_PATH_RW_OPERATIONS)
diff --git a/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/auth/ITestCustomSigner.java b/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/auth/ITestCustomSigner.java
index e6eb60ad06134..c357c8aac0f50 100644
--- a/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/auth/ITestCustomSigner.java
+++ b/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/auth/ITestCustomSigner.java
@@ -25,6 +25,7 @@
import java.util.HashMap;
import java.util.Map;
import java.util.Objects;
+import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import org.junit.runner.RunWith;
@@ -51,14 +52,19 @@
import org.apache.hadoop.fs.s3a.S3AFileSystem;
import org.apache.hadoop.fs.s3a.auth.ITestCustomSigner.CustomSignerInitializer.StoreValue;
import org.apache.hadoop.fs.s3a.auth.delegation.DelegationTokenProvider;
+import org.apache.hadoop.fs.s3a.impl.ChecksumSupport;
import org.apache.hadoop.security.UserGroupInformation;
+import static org.apache.hadoop.fs.s3a.Constants.CHECKSUM_ALGORITHM;
+import static org.apache.hadoop.fs.s3a.Constants.CHECKSUM_VALIDATION;
import static org.apache.hadoop.fs.s3a.Constants.CUSTOM_SIGNERS;
import static org.apache.hadoop.fs.s3a.Constants.ENABLE_MULTI_DELETE;
+import static org.apache.hadoop.fs.s3a.Constants.PATH_STYLE_ACCESS;
import static org.apache.hadoop.fs.s3a.Constants.SIGNING_ALGORITHM_S3;
import static org.apache.hadoop.fs.s3a.MultipartTestUtils.createMagicFile;
import static org.apache.hadoop.fs.s3a.S3ATestUtils.disableFilesystemCaching;
import static org.apache.hadoop.fs.s3a.S3ATestUtils.removeBaseAndBucketOverrides;
+import static org.apache.hadoop.fs.s3a.S3ATestUtils.skipIfNotEnabled;
/**
* Tests for custom Signers and SignerInitializers.
@@ -74,6 +80,13 @@ public class ITestCustomSigner extends AbstractS3ATestBase {
private static final String TEST_ID_KEY = "TEST_ID_KEY";
private static final String TEST_REGION_KEY = "TEST_REGION_KEY";
+ /**
+ * Is the store using path style access?
+ */
+ private static final AtomicBoolean PATH_STYLE_ACCESS_IN_USE = new AtomicBoolean(false);
+
+ public static final String BUCKET = "bucket";
+
/**
* Parameterization.
*/
@@ -106,7 +119,11 @@ public void setup() throws Exception {
super.setup();
final S3AFileSystem fs = getFileSystem();
final Configuration conf = fs.getConf();
+ if (bulkDelete) {
+ skipIfNotEnabled(conf, ENABLE_MULTI_DELETE, "no bulk delete");
+ }
endpoint = conf.getTrimmed(Constants.ENDPOINT, Constants.CENTRAL_ENDPOINT);
+ PATH_STYLE_ACCESS_IN_USE.set(conf.getBoolean(PATH_STYLE_ACCESS, false));
LOG.info("Test endpoint is {}", endpoint);
regionName = conf.getTrimmed(Constants.AWS_REGION, "");
if (regionName.isEmpty()) {
@@ -153,6 +170,7 @@ private S3AFileSystem runStoreOperationsAndVerify(UserGroupInformation ugi,
throws IOException, InterruptedException {
Configuration conf = createTestConfig(identifier);
return ugi.doAs((PrivilegedExceptionAction
+ * When working with buckets using path-style resolution, the store bucket name
+ * is just {@link #BUCKET}.
+ */
private static final Map
+ * In path-style-access the bucket name is mapped to {@link #BUCKET} so only
+ * one bucket per UGI instance is supported.
+ */
private static class StoreKey {
private final String bucketName;
private final UserGroupInformation ugi;
diff --git a/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/auth/ITestHttpSigner.java b/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/auth/ITestHttpSigner.java
index db0aaa6be0eca..e886fca5d71c6 100644
--- a/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/auth/ITestHttpSigner.java
+++ b/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/auth/ITestHttpSigner.java
@@ -140,11 +140,13 @@ private S3AFileSystem runStoreOperationsAndVerify(UserGroupInformation ugi,
ContractTestUtils.touch(fs, new Path(subdir, "file1"));
// create a magic file.
- createMagicFile(fs, subdir);
- ContentSummary summary = fs.getContentSummary(finalPath);
- fs.getS3AInternals().abortMultipartUploads(subdir);
- fs.rename(subdir, new Path(finalPath, "renamed"));
- fs.delete(finalPath, true);
+ if (fs.isMagicCommitEnabled()) {
+ createMagicFile(fs, subdir);
+ ContentSummary summary = fs.getContentSummary(finalPath);
+ fs.getS3AInternals().abortMultipartUploads(subdir);
+ fs.rename(subdir, new Path(finalPath, "renamed"));
+ fs.delete(finalPath, true);
+ }
return fs;
});
}
diff --git a/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/auth/ITestRestrictedReadAccess.java b/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/auth/ITestRestrictedReadAccess.java
index 7151c38ad3e27..85e9f3f142c4c 100644
--- a/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/auth/ITestRestrictedReadAccess.java
+++ b/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/auth/ITestRestrictedReadAccess.java
@@ -47,6 +47,7 @@
import static org.apache.hadoop.fs.s3a.Constants.ASSUMED_ROLE_ARN;
import static org.apache.hadoop.fs.s3a.S3ATestUtils.assume;
import static org.apache.hadoop.fs.s3a.S3ATestUtils.lsR;
+import static org.apache.hadoop.fs.s3a.S3ATestUtils.skipIfS3ExpressBucket;
import static org.apache.hadoop.fs.s3a.auth.RoleModel.Effects;
import static org.apache.hadoop.fs.s3a.auth.RoleModel.Statement;
import static org.apache.hadoop.fs.s3a.auth.RoleModel.directory;
@@ -86,7 +87,9 @@
* To simplify maintenance, the operations tested are broken up into
* their own methods, with fields used to share the restricted role and
* created paths.
- *
+ *
+ * Test are skipped if no assumed role was provided, or if the test bucket
+ * is an S3Express bucket, whose permissions model is different.
*/
public class ITestRestrictedReadAccess extends AbstractS3ATestBase {
@@ -158,6 +161,7 @@ public class ITestRestrictedReadAccess extends AbstractS3ATestBase {
public void setup() throws Exception {
super.setup();
assumeRoleTests();
+ skipIfS3ExpressBucket(getConfiguration());
}
@Override
diff --git a/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/auth/RoleTestUtils.java b/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/auth/RoleTestUtils.java
index b0f685b076094..ac244be668b9d 100644
--- a/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/auth/RoleTestUtils.java
+++ b/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/auth/RoleTestUtils.java
@@ -164,8 +164,8 @@ public static Configuration newAssumedRoleConfig(
conf.set(ASSUMED_ROLE_ARN, roleARN);
conf.set(ASSUMED_ROLE_SESSION_NAME, "test");
conf.set(ASSUMED_ROLE_SESSION_DURATION, "15m");
- // force in bucket resolution during startup
- conf.setInt(S3A_BUCKET_PROBE, 1);
+ // disable bucket resolution during startup as s3 express doesn't like it
+ conf.setInt(S3A_BUCKET_PROBE, 0);
disableCreateSession(conf);
disableFilesystemCaching(conf);
return conf;
diff --git a/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/commit/AbstractITCommitProtocol.java b/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/commit/AbstractITCommitProtocol.java
index 3a7cceb2369ee..1aaf58152473e 100644
--- a/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/commit/AbstractITCommitProtocol.java
+++ b/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/commit/AbstractITCommitProtocol.java
@@ -170,7 +170,7 @@ public void setup() throws Exception {
taskAttempt0 = TaskAttemptID.forName(attempt0);
attempt1 = "attempt_" + jobId + "_m_000001_0";
taskAttempt1 = TaskAttemptID.forName(attempt1);
-
+ assumeMultipartUploads(getFileSystem().getConf());
outDir = path(getMethodName());
abortMultipartUploadsUnderPath(outDir);
cleanupDestDir();
diff --git a/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/commit/ITestCommitOperationCost.java b/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/commit/ITestCommitOperationCost.java
index aea0f1883243a..2b288bfec0ad4 100644
--- a/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/commit/ITestCommitOperationCost.java
+++ b/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/commit/ITestCommitOperationCost.java
@@ -40,6 +40,7 @@
import org.apache.hadoop.fs.s3a.performance.AbstractS3ACostTest;
import org.apache.hadoop.fs.statistics.IOStatisticsLogging;
+import static org.apache.hadoop.fs.s3a.S3ATestUtils.assumeMultipartUploads;
import static org.apache.hadoop.fs.s3a.S3ATestUtils.skipIfAnalyticsAcceleratorEnabled;
import static org.apache.hadoop.fs.s3a.Statistic.ACTION_HTTP_GET_REQUEST;
import static org.apache.hadoop.fs.s3a.Statistic.COMMITTER_MAGIC_FILES_CREATED;
@@ -91,6 +92,7 @@ public ITestCommitOperationCost() {
@Override
public void setup() throws Exception {
super.setup();
+ assumeMultipartUploads(getFileSystem().getConf());
testHelper = new CommitterTestHelper(getFileSystem());
}
diff --git a/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/commit/ITestCommitOperations.java b/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/commit/ITestCommitOperations.java
index 34b856c21b7e8..dd07500684785 100644
--- a/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/commit/ITestCommitOperations.java
+++ b/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/commit/ITestCommitOperations.java
@@ -92,6 +92,7 @@ protected Configuration createConfiguration() {
public void setup() throws Exception {
FileSystem.closeAll();
super.setup();
+ assumeMultipartUploads(getFileSystem().getConf());
verifyIsMagicCommitFS(getFileSystem());
progress = new ProgressCounter();
progress.assertCount("progress", 0);
diff --git a/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/commit/ITestS3ACommitterFactory.java b/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/commit/ITestS3ACommitterFactory.java
index 2561a69f60b59..9abbfd97cbe42 100644
--- a/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/commit/ITestS3ACommitterFactory.java
+++ b/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/commit/ITestS3ACommitterFactory.java
@@ -46,6 +46,7 @@
import org.apache.hadoop.mapreduce.task.TaskAttemptContextImpl;
import org.apache.hadoop.security.UserGroupInformation;
+import static org.apache.hadoop.fs.s3a.S3ATestUtils.assumeMultipartUploads;
import static org.apache.hadoop.fs.s3a.S3ATestUtils.removeBaseAndBucketOverrides;
import static org.apache.hadoop.fs.s3a.commit.CommitConstants.*;
import static org.apache.hadoop.fs.s3a.commit.InternalCommitterConstants.COMMITTER_NAME_STAGING;
@@ -182,6 +183,7 @@ public void setup() throws Exception {
// destroy all filesystems from previous runs.
FileSystem.closeAllForUGI(UserGroupInformation.getCurrentUser());
super.setup();
+ assumeMultipartUploads(getFileSystem().getConf());
jobId = randomJobId();
attempt0 = "attempt_" + jobId + "_m_000000_0";
taskAttempt0 = TaskAttemptID.forName(attempt0);
diff --git a/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/commit/ITestUploadRecovery.java b/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/commit/ITestUploadRecovery.java
index 2ede6d82798d0..ec5e807b23d08 100644
--- a/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/commit/ITestUploadRecovery.java
+++ b/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/commit/ITestUploadRecovery.java
@@ -19,6 +19,7 @@
package org.apache.hadoop.fs.s3a.commit;
import java.io.File;
+import java.io.FileNotFoundException;
import java.util.Arrays;
import java.util.Collection;
import java.util.UUID;
@@ -57,11 +58,13 @@
import static org.apache.hadoop.fs.s3a.Constants.FS_S3A_CREATE_PERFORMANCE;
import static org.apache.hadoop.fs.s3a.Constants.MAX_ERROR_RETRIES;
import static org.apache.hadoop.fs.s3a.Constants.RETRY_HTTP_5XX_ERRORS;
+import static org.apache.hadoop.fs.s3a.S3ATestUtils.assumeMultipartUploads;
import static org.apache.hadoop.fs.s3a.S3ATestUtils.removeBaseAndBucketOverrides;
import static org.apache.hadoop.fs.s3a.audit.S3AAuditConstants.AUDIT_EXECUTION_INTERCEPTORS;
import static org.apache.hadoop.fs.s3a.commit.CommitConstants.BASE;
import static org.apache.hadoop.fs.s3a.commit.CommitConstants.MAGIC_PATH_PREFIX;
import static org.apache.hadoop.fs.s3a.test.SdkFaultInjector.setRequestFailureConditions;
+import static org.apache.hadoop.test.LambdaTestUtils.intercept;
/**
* Test upload recovery by injecting failures into the response chain.
@@ -156,6 +159,10 @@ public Configuration createConfiguration() {
public void setup() throws Exception {
SdkFaultInjector.resetFaultInjector();
super.setup();
+ if (!FAST_UPLOAD_BUFFER_DISK.equals(buffer)) {
+ assumeMultipartUploads(getFileSystem().getConf());
+ }
+
}
@Override
@@ -163,7 +170,6 @@ public void teardown() throws Exception {
// safety check in case the evaluation is failing any
// request needed in cleanup.
SdkFaultInjector.resetFaultInjector();
-
super.teardown();
}
@@ -260,9 +266,18 @@ public void testCommitOperations() throws Throwable {
setRequestFailureConditions(2,
SdkFaultInjector::isCompleteMultipartUploadRequest);
+ boolean mpuCommitConsumesUploadId = getFileSystem().getConf().getBoolean(
+ MULTIPART_COMMIT_CONSUMES_UPLOAD_ID,
+ DEFAULT_MULTIPART_COMMIT_CONSUMES_UPLOAD_ID);
try (CommitContext commitContext
= actions.createCommitContextForTesting(dest, JOB_ID, 0)) {
- commitContext.commitOrFail(commit);
+
+ if (mpuCommitConsumesUploadId) {
+ intercept(FileNotFoundException.class, () ->
+ commitContext.commitOrFail(commit));
+ } else {
+ commitContext.commitOrFail(commit);
+ }
}
// make sure the saved data is as expected
verifyFileContents(fs, dest, dataset);
diff --git a/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/commit/integration/ITestS3ACommitterMRJob.java b/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/commit/integration/ITestS3ACommitterMRJob.java
index 7488de41ce638..fda87d2d67d60 100644
--- a/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/commit/integration/ITestS3ACommitterMRJob.java
+++ b/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/commit/integration/ITestS3ACommitterMRJob.java
@@ -72,6 +72,7 @@
import org.apache.hadoop.security.UserGroupInformation;
import org.apache.hadoop.util.DurationInfo;
+import static org.apache.hadoop.fs.s3a.S3ATestUtils.assumeMultipartUploads;
import static org.apache.hadoop.fs.s3a.S3ATestUtils.disableFilesystemCaching;
import static org.apache.hadoop.fs.s3a.S3ATestUtils.lsR;
import static org.apache.hadoop.fs.s3a.S3AUtils.applyLocatedFiles;
@@ -173,6 +174,7 @@ public ITestS3ACommitterMRJob(
@Override
public void setup() throws Exception {
super.setup();
+ assumeMultipartUploads(getFileSystem().getConf());
// configure the test binding for this specific test case.
committerTestBinding.setup(getClusterBinding(), getFileSystem());
}
diff --git a/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/commit/magic/ITestS3AHugeMagicCommits.java b/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/commit/magic/ITestS3AHugeMagicCommits.java
index 116d48e9de5fc..89b471bbd29fc 100644
--- a/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/commit/magic/ITestS3AHugeMagicCommits.java
+++ b/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/commit/magic/ITestS3AHugeMagicCommits.java
@@ -112,6 +112,15 @@ public void setup() throws Exception {
pendingDataFile = new Path(jobDir, filename + PENDING_SUFFIX);
}
+ /**
+ * Skip this test suite when MPUS are not avaialable.
+ * @return false
+ */
+ @Override
+ protected boolean requireMultipartUploads() {
+ return true;
+ }
+
/**
* Returns the path to the commit metadata file, not that of the huge file.
* @return a file in the job dir
diff --git a/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/commit/terasort/ITestTerasortOnS3A.java b/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/commit/terasort/ITestTerasortOnS3A.java
index da1580076dbb8..e2b70a9b83b80 100644
--- a/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/commit/terasort/ITestTerasortOnS3A.java
+++ b/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/commit/terasort/ITestTerasortOnS3A.java
@@ -56,6 +56,7 @@
import org.apache.hadoop.util.ToolRunner;
import static java.util.Optional.empty;
+import static org.apache.hadoop.fs.s3a.S3ATestUtils.assumeMultipartUploads;
import static org.apache.hadoop.fs.s3a.S3ATestUtils.lsR;
/**
@@ -142,6 +143,7 @@ protected String committerName() {
public void setup() throws Exception {
super.setup();
requireScaleTestsEnabled();
+ assumeMultipartUploads(getFileSystem().getConf());
prepareToTerasort();
}
diff --git a/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/impl/ITestConnectionTimeouts.java b/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/impl/ITestConnectionTimeouts.java
index aeb9629b3a6d1..bdb61a4e46786 100644
--- a/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/impl/ITestConnectionTimeouts.java
+++ b/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/impl/ITestConnectionTimeouts.java
@@ -52,6 +52,8 @@
import static org.apache.hadoop.fs.s3a.Constants.ESTABLISH_TIMEOUT;
import static org.apache.hadoop.fs.s3a.Constants.FS_S3A_CREATE_PERFORMANCE;
import static org.apache.hadoop.fs.s3a.Constants.FS_S3A_PERFORMANCE_FLAGS;
+import static org.apache.hadoop.fs.s3a.Constants.INPUT_STREAM_TYPE;
+import static org.apache.hadoop.fs.s3a.Constants.INPUT_STREAM_TYPE_CLASSIC;
import static org.apache.hadoop.fs.s3a.Constants.MAXIMUM_CONNECTIONS;
import static org.apache.hadoop.fs.s3a.Constants.MAX_ERROR_RETRIES;
import static org.apache.hadoop.fs.s3a.Constants.PART_UPLOAD_TIMEOUT;
@@ -110,6 +112,7 @@ private Configuration timingOutConfiguration() {
CONNECTION_ACQUISITION_TIMEOUT,
CONNECTION_IDLE_TIME,
ESTABLISH_TIMEOUT,
+ INPUT_STREAM_TYPE,
MAX_ERROR_RETRIES,
MAXIMUM_CONNECTIONS,
PART_UPLOAD_TIMEOUT,
@@ -125,6 +128,7 @@ private Configuration timingOutConfiguration() {
// needed to ensure that streams are kept open.
// without this the tests is unreliable in batch runs.
disablePrefetching(conf);
+ conf.set(INPUT_STREAM_TYPE, INPUT_STREAM_TYPE_CLASSIC);
conf.setInt(RETRY_LIMIT, 0);
conf.setBoolean(FS_S3A_CREATE_PERFORMANCE, true);
final Duration ms10 = Duration.ofMillis(10);
@@ -190,7 +194,7 @@ public void testObjectUploadTimeouts() throws Throwable {
AWSClientConfig.setMinimumOperationDuration(Duration.ZERO);
final Path dir = methodPath();
Path file = new Path(dir, "file");
- Configuration conf = new Configuration(getConfiguration());
+ Configuration conf = new Configuration(getFileSystem().getConf());
removeBaseAndBucketOverrides(conf,
PART_UPLOAD_TIMEOUT,
REQUEST_TIMEOUT,
@@ -257,20 +261,21 @@ public void testObjectUploadTimeouts() throws Throwable {
// and try a multipart upload to verify that its requests also outlast
// the short requests
- SdkFaultInjector.setRequestFailureConditions(999,
- SdkFaultInjector::isPartUpload);
- Path magicFile = new Path(dir, MAGIC_PATH_PREFIX + "0001/__base/file2");
- totalSleepTime.set(0);
- OperationDuration dur2 = new DurationInfo(LOG, "Creating File");
- ContractTestUtils.createFile(brittleFS, magicFile, true, DATASET);
- dur2.finished();
- Assertions.assertThat(totalSleepTime.get())
- .describedAs("total sleep time of magic write")
- .isGreaterThan(0);
- Assertions.assertThat(dur2.asDuration())
- .describedAs("Duration of magic write")
- .isGreaterThan(shortTimeout);
- brittleFS.delete(dir, true);
+ if (fs.isMagicCommitEnabled()) {
+ SdkFaultInjector.setRequestFailureConditions(999,
+ SdkFaultInjector::isPartUpload);
+ Path magicFile = new Path(dir, MAGIC_PATH_PREFIX + "0001/__base/file2");
+ totalSleepTime.set(0);
+ OperationDuration dur2 = new DurationInfo(LOG, "Creating File");
+ ContractTestUtils.createFile(brittleFS, magicFile, true, DATASET);
+ dur2.finished();
+ Assertions.assertThat(totalSleepTime.get())
+ .describedAs("total sleep time of magic write")
+ .isGreaterThan(0);
+ Assertions.assertThat(dur2.asDuration())
+ .describedAs("Duration of magic write")
+ .isGreaterThan(shortTimeout);
+ }
}
}
}
diff --git a/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/impl/ITestPartialRenamesDeletes.java b/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/impl/ITestPartialRenamesDeletes.java
index 8a285a4db40a8..3bf338814f6dc 100644
--- a/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/impl/ITestPartialRenamesDeletes.java
+++ b/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/impl/ITestPartialRenamesDeletes.java
@@ -72,7 +72,9 @@
* Test partial failures of delete and rename operations,.
*
* All these test have a unique path for each run, with a roleFS having
- * full RW access to part of it, and R/O access to a restricted subdirectory
+ * full RW access to part of it, and R/O access to a restricted subdirectory.
+ *
+ * Tests are skipped on S3Express buckets or if no assumed role is provided.
*
*
* Marked as a scale test even though it tries to aggressively abort streams being written
* and should, if working, complete fast.
+ *
+ * Assumes multipart uploads are enabled; single part upload interruptions aren't the complicated
+ * ones.
*/
@RunWith(Parameterized.class)
public class ITestS3ABlockOutputStreamInterruption extends S3AScaleTestBase {
@@ -168,6 +172,7 @@ protected Configuration createScaleConfiguration() {
public void setup() throws Exception {
SdkFaultInjector.resetFaultInjector();
super.setup();
+ assumeMultipartUploads(getFileSystem().getConf());
}
@Override
@@ -283,6 +288,7 @@ public void testAbortDuringUpload() throws Throwable {
@Test
public void testPartUploadFailure() throws Throwable {
describe("Trigger a failure during a multipart upload");
+ assumeMultipartUploads(getFileSystem().getConf());
int len = 6 * _1MB;
final byte[] dataset = dataset(len, 'a', 'z' - 'a');
final String text = "Simulated failure";
diff --git a/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/scale/ITestS3AHugeFilesArrayBlocks.java b/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/scale/ITestS3AHugeFilesArrayBlocks.java
index d6f15c8c84476..c85f5d54b537e 100644
--- a/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/scale/ITestS3AHugeFilesArrayBlocks.java
+++ b/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/scale/ITestS3AHugeFilesArrayBlocks.java
@@ -28,4 +28,13 @@ public class ITestS3AHugeFilesArrayBlocks extends AbstractSTestS3AHugeFiles {
protected String getBlockOutputBufferName() {
return Constants.FAST_UPLOAD_BUFFER_ARRAY;
}
+
+ /**
+ * Skip this test suite when MPUS are not avaialable.
+ * @return false
+ */
+ @Override
+ protected boolean requireMultipartUploads() {
+ return true;
+ }
}
diff --git a/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/scale/ITestS3AHugeFilesByteBufferBlocks.java b/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/scale/ITestS3AHugeFilesByteBufferBlocks.java
index 1e74d715b88fa..6cc62f8512746 100644
--- a/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/scale/ITestS3AHugeFilesByteBufferBlocks.java
+++ b/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/scale/ITestS3AHugeFilesByteBufferBlocks.java
@@ -40,6 +40,15 @@ protected String getBlockOutputBufferName() {
return FAST_UPLOAD_BYTEBUFFER;
}
+ /**
+ * Skip this test suite when MPUS are not avaialable.
+ * @return false
+ */
+ @Override
+ protected boolean requireMultipartUploads() {
+ return true;
+ }
+
/**
* Rename the parent directory, rather than the file itself.
* @param src source file
diff --git a/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/scale/ITestS3AMultipartUploadSizeLimits.java b/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/scale/ITestS3AMultipartUploadSizeLimits.java
index b83d12b4c1a66..121a8c07b3dd7 100644
--- a/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/scale/ITestS3AMultipartUploadSizeLimits.java
+++ b/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/scale/ITestS3AMultipartUploadSizeLimits.java
@@ -43,6 +43,7 @@
import static org.apache.hadoop.fs.contract.ContractTestUtils.verifyFileContents;
import static org.apache.hadoop.fs.contract.ContractTestUtils.writeTextFile;
import static org.apache.hadoop.fs.s3a.Constants.MULTIPART_SIZE;
+import static org.apache.hadoop.fs.s3a.S3ATestUtils.assumeMultipartUploads;
import static org.apache.hadoop.fs.s3a.S3ATestUtils.removeBaseAndBucketOverrides;
import static org.apache.hadoop.fs.s3a.Statistic.INVOCATION_ABORT;
import static org.apache.hadoop.fs.s3a.Statistic.OBJECT_MULTIPART_UPLOAD_ABORTED;
@@ -74,6 +75,12 @@ protected Configuration createScaleConfiguration() {
return configuration;
}
+ @Override
+ public void setup() throws Exception {
+ super.setup();
+ assumeMultipartUploads(getFileSystem().getConf());
+ }
+
/**
* Uploads under the limit are valid.
*/
diff --git a/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/statistics/ITestAWSStatisticCollection.java b/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/statistics/ITestAWSStatisticCollection.java
index 594cb0cdafb87..e6d7afd824fb7 100644
--- a/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/statistics/ITestAWSStatisticCollection.java
+++ b/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/statistics/ITestAWSStatisticCollection.java
@@ -19,12 +19,17 @@
package org.apache.hadoop.fs.s3a.statistics;
import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.fs.s3a.S3AFileSystem;
+import org.apache.hadoop.fs.s3a.S3ATestUtils;
import org.apache.hadoop.fs.s3a.performance.AbstractS3ACostTest;
-import static org.apache.hadoop.fs.s3a.Constants.FS_S3A_CREATE_PERFORMANCE;
+import static org.apache.hadoop.fs.s3a.Constants.S3EXPRESS_CREATE_SESSION;
+import static org.apache.hadoop.fs.s3a.S3ATestUtils.setPerformanceFlags;
import static org.apache.hadoop.fs.s3a.Statistic.STORE_IO_REQUEST;
/**
@@ -32,19 +37,27 @@
*/
public class ITestAWSStatisticCollection extends AbstractS3ACostTest {
+ private static final Logger LOG =
+ LoggerFactory.getLogger(ITestAWSStatisticCollection.class);
+
@Override
public Configuration createConfiguration() {
final Configuration conf = super.createConfiguration();
- conf.setBoolean(FS_S3A_CREATE_PERFORMANCE, true);
+ S3ATestUtils.removeBaseAndBucketOverrides(conf,
+ S3EXPRESS_CREATE_SESSION);
+ setPerformanceFlags(conf, "create");
+ conf.setBoolean(S3EXPRESS_CREATE_SESSION, false);
return conf;
}
@Test
public void testSDKMetricsCostOfGetFileStatusOnFile() throws Throwable {
- describe("performing getFileStatus on a file");
+ describe("Performing getFileStatus() on a file");
Path simpleFile = file(methodPath());
// and repeat on the file looking at AWS wired up stats
- verifyMetrics(() -> getFileSystem().getFileStatus(simpleFile),
+ final S3AFileSystem fs = getFileSystem();
+ LOG.info("Initiating GET request for {}", simpleFile);
+ verifyMetrics(() -> fs.getFileStatus(simpleFile),
with(STORE_IO_REQUEST, 1));
}
diff --git a/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/yarn/ITestS3AMiniYarnCluster.java b/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/yarn/ITestS3AMiniYarnCluster.java
index 87b37b7f8ffbc..5ab153df31626 100644
--- a/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/yarn/ITestS3AMiniYarnCluster.java
+++ b/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/yarn/ITestS3AMiniYarnCluster.java
@@ -44,6 +44,7 @@
import org.junit.Test;
+import static org.apache.hadoop.fs.s3a.S3ATestUtils.assumeMultipartUploads;
import static org.apache.hadoop.fs.s3a.commit.CommitConstants.FS_S3A_COMMITTER_NAME;
import static org.apache.hadoop.fs.s3a.commit.CommitConstants.FS_S3A_COMMITTER_STAGING_UNIQUE_FILENAMES;
import static org.apache.hadoop.fs.s3a.commit.CommitConstants._SUCCESS;
@@ -71,6 +72,7 @@ public void setup() throws Exception {
super.setup();
S3AFileSystem fs = getFileSystem();
Configuration conf = getConfiguration();
+ assumeMultipartUploads(fs.getConf());
rootPath = path("MiniClusterWordCount");
Path workingDir = path("working");
fs.setWorkingDirectory(workingDir);
diff --git a/hadoop-tools/hadoop-aws/src/test/resources/log4j.properties b/hadoop-tools/hadoop-aws/src/test/resources/log4j.properties
index f61668643a1b5..fe651742e02be 100644
--- a/hadoop-tools/hadoop-aws/src/test/resources/log4j.properties
+++ b/hadoop-tools/hadoop-aws/src/test/resources/log4j.properties
@@ -68,8 +68,13 @@ log4j.logger.org.apache.hadoop.fs.s3a.SDKV2Upgrade=WARN
# include sensitive information such as account IDs in HTTP headers.
# log4j.logger.software.amazon.awssdk.request=DEBUG
+# Log TLS info
+#log4j.logger.software.amazon.awssdk.thirdparty.org.apache.http.conn.ssl.SSLConnectionSocketFactory=DEBUG
+
+
# Turn on low level HTTP protocol debugging
-#log4j.logger.org.apache.http.wire=DEBUG
+#log4j.logger.software.amazon.awssdk.thirdparty.org.apache.http.wire=DEBUG
+#log4j.logger.software.amazon.awssdk.thirdparty.org.apache.http=DEBUG
# async client
#log4j.logger.io.netty.handler.logging=DEBUG
*