You are viewing a plain text version of this content. The canonical link for it is here.
Posted to common-commits@hadoop.apache.org by ah...@apache.org on 2023/12/04 09:47:14 UTC
(hadoop) branch feature-HADOOP-18073-sdk-v2-upgrade-3.4 updated (1aeae7a1fc26 -> c2c9e3885262)
This is an automated email from the ASF dual-hosted git repository.
ahmar pushed a change to branch feature-HADOOP-18073-sdk-v2-upgrade-3.4
in repository https://gitbox.apache.org/repos/asf/hadoop.git
omit 1aeae7a1fc26 HADOOP-18850. S3A: Enable dual-layer server-side encryption with AWS KMS keys (#6140)
new f738f4321f4d HADOOP-18930. Make fs.s3a.create.performance a bucket-wide setting. (#6168)
new 6f69a2fb1851 HADOOP-18915. Tune/extend S3A http connection and thread pool settings (#6180)
new b327fcac96d1 HADOOP-18925. S3A: option to enable/disable CopyFromLocalOperation (#6163)
new 8f4fd0472210 S3A: Upgrade AWS SDK version to 2.21.33 for Amazon S3 Express One Zone support (#6306)
new c2c9e3885262 HADOOP-18996. S3A to provide full support for S3 Express One Zone (#6308)
This update added new revisions after undoing existing revisions.
That is to say, some revisions that were in the old version of the
branch are not in the new version. This situation occurs
when a user --force pushes a change and generates a repository
containing something like this:
* -- * -- B -- O -- O -- O (1aeae7a1fc26)
\
N -- N -- N refs/heads/feature-HADOOP-18073-sdk-v2-upgrade-3.4 (c2c9e3885262)
You should already have received notification emails for all of the O
revisions, and so the following emails describe only the N revisions
from the common base, B.
Any revisions marked "omit" are not gone; other references still
refer to them. Any revisions marked "discard" are gone forever.
The 5 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
LICENSE-binary | 2 +-
.../apache/hadoop/fs/CommonPathCapabilities.java | 11 +
.../main/java/org/apache/hadoop/fs/FileUtil.java | 29 ++
.../java/org/apache/hadoop/fs/shell/Command.java | 17 +-
.../src/main/resources/core-default.xml | 32 +-
.../filesystem/fsdataoutputstreambuilder.md | 6 +
.../org/apache/hadoop/fs/FileContextURIBase.java | 32 +-
.../AbstractContractRootDirectoryTest.java | 27 +-
.../hadoop/fs/contract/ContractTestUtils.java | 66 ++-
.../hadoop/mapred/LocatedFileStatusFetcher.java | 34 +-
.../mapreduce/lib/input/FileInputFormat.java | 30 +-
hadoop-project/pom.xml | 2 +-
...eption.java => AWSApiCallTimeoutException.java} | 23 +-
.../java/org/apache/hadoop/fs/s3a/Constants.java | 236 ++++++++++-
.../java/org/apache/hadoop/fs/s3a/Listing.java | 2 +-
.../apache/hadoop/fs/s3a/S3AEncryptionMethods.java | 5 +-
.../org/apache/hadoop/fs/s3a/S3AFileSystem.java | 279 ++++++++++---
.../org/apache/hadoop/fs/s3a/S3AInternals.java | 10 +
.../org/apache/hadoop/fs/s3a/S3ARetryPolicy.java | 3 +
.../java/org/apache/hadoop/fs/s3a/S3AUtils.java | 44 +-
.../hadoop/fs/s3a/audit/AWSRequestAnalyzer.java | 4 +-
.../apache/hadoop/fs/s3a/auth/CustomSdkSigner.java | 178 ++++++++
.../delegation/EncryptionSecretOperations.java | 3 +-
.../apache/hadoop/fs/s3a/impl/AWSClientConfig.java | 351 +++++++++++++---
.../hadoop/fs/s3a/impl/ConfigurationHelper.java | 132 ++++++
.../hadoop/fs/s3a/impl/CreateFileBuilder.java | 7 +
.../apache/hadoop/fs/s3a/impl/DeleteOperation.java | 20 +-
.../hadoop/fs/s3a/impl/InternalConstants.java | 50 ++-
.../apache/hadoop/fs/s3a/impl/NetworkBinding.java | 11 +
.../hadoop/fs/s3a/impl/RequestFactoryImpl.java | 109 ++---
.../hadoop/fs/s3a/impl/S3ExpressStorage.java | 76 ++++
.../apache/hadoop/fs/s3a/s3guard/S3GuardTool.java | 52 ++-
.../org/apache/hadoop/fs/s3a/tools/BucketTool.java | 304 ++++++++++++++
.../org/apache/hadoop/fs/s3a/tools/MarkerTool.java | 8 +-
.../markdown/tools/hadoop-aws/aws_sdk_upgrade.md | 10 +
.../site/markdown/tools/hadoop-aws/encryption.md | 86 +---
.../src/site/markdown/tools/hadoop-aws/index.md | 103 +----
.../site/markdown/tools/hadoop-aws/performance.md | 113 +++--
.../src/site/markdown/tools/hadoop-aws/testing.md | 2 +-
.../tools/hadoop-aws/third_party_stores.md | 11 +-
.../tools/hadoop-aws/troubleshooting_s3a.md | 10 +
.../fs/contract/s3a/ITestS3AContractCreate.java | 93 +++++
.../hadoop/fs/s3a/AbstractTestS3AEncryption.java | 1 +
.../apache/hadoop/fs/s3a/EncryptionTestUtils.java | 44 +-
.../hadoop/fs/s3a/ITestS3AConfiguration.java | 2 +-
.../hadoop/fs/s3a/ITestS3ACopyFromLocalFile.java | 51 +++
...TestS3ADSSEEncryptionWithDefaultS3Settings.java | 167 --------
.../ITestS3AEncryptionDSSEKMSUserDefinedKey.java | 61 ---
.../hadoop/fs/s3a/ITestS3AEncryptionSSEC.java | 2 +-
.../ITestS3AEncryptionWithDefaultS3Settings.java | 26 +-
.../hadoop/fs/s3a/ITestS3AFSMainOperations.java | 20 +
.../hadoop/fs/s3a/ITestS3AFileOperationCost.java | 18 +-
.../hadoop/fs/s3a/ITestS3AFileSystemContract.java | 20 +
.../hadoop/fs/s3a/ITestS3AMiscOperations.java | 77 ----
.../apache/hadoop/fs/s3a/MultipartTestUtils.java | 55 ++-
.../org/apache/hadoop/fs/s3a/S3ATestUtils.java | 110 ++++-
.../hadoop/fs/s3a/TestS3AExceptionTranslation.java | 53 ++-
.../hadoop/fs/s3a/auth/ITestCustomSigner.java | 124 ++++--
.../fs/s3a/commit/ITestCommitOperations.java | 6 +-
.../fs/s3a/impl/ITestConnectionTimeouts.java | 149 +++++++
.../hadoop/fs/s3a/impl/ITestTreewalkProblems.java | 463 +++++++++++++++++++++
.../ITestUploadPurgeOnDirectoryOperations.java | 22 +-
.../apache/hadoop/fs/s3a/impl/ITestXAttrCost.java | 7 +-
.../hadoop/fs/s3a/impl/TestAwsClientConfig.java | 188 +++++++++
.../hadoop/fs/s3a/impl/TestS3ExpressStorage.java | 63 +++
.../fs/s3a/performance/ITestCreateFileCost.java | 118 +++++-
.../performance/ITestDirectoryMarkerListing.java | 31 +-
.../fs/s3a/performance/ITestS3ADeleteCost.java | 11 +
.../fs/s3a/performance/ITestUnbufferDraining.java | 30 +-
.../s3a/s3guard/AbstractS3GuardToolTestBase.java | 6 +-
.../fs/s3a/scale/AbstractSTestS3AHugeFiles.java | 72 +++-
.../scale/ITestS3AHugeFilesByteBufferBlocks.java | 29 ++
.../fs/s3a/scale/ITestS3AHugeFilesEncryption.java | 13 +-
.../fs/s3a/tools/AbstractMarkerToolTest.java | 5 +-
.../hadoop/fs/s3a/tools/ITestBucketTool.java | 183 ++++++++
.../hadoop-aws/src/test/resources/core-site.xml | 2 +-
.../hadoop-aws/src/test/resources/log4j.properties | 4 +
.../apache/hadoop/tools/util/DistCpTestUtils.java | 10 +-
78 files changed, 3768 insertions(+), 1098 deletions(-)
copy hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/{AWSRedirectException.java => AWSApiCallTimeoutException.java} (62%)
create mode 100644 hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/auth/CustomSdkSigner.java
create mode 100644 hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/impl/ConfigurationHelper.java
create mode 100644 hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/impl/S3ExpressStorage.java
create mode 100644 hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/tools/BucketTool.java
delete mode 100644 hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/ITestS3ADSSEEncryptionWithDefaultS3Settings.java
delete mode 100644 hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/ITestS3AEncryptionDSSEKMSUserDefinedKey.java
create mode 100644 hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/impl/ITestConnectionTimeouts.java
create mode 100644 hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/impl/ITestTreewalkProblems.java
create mode 100644 hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/impl/TestAwsClientConfig.java
create mode 100644 hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/impl/TestS3ExpressStorage.java
create mode 100644 hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/tools/ITestBucketTool.java
---------------------------------------------------------------------
To unsubscribe, e-mail: common-commits-unsubscribe@hadoop.apache.org
For additional commands, e-mail: common-commits-help@hadoop.apache.org
(hadoop) 01/05: HADOOP-18930. Make fs.s3a.create.performance a bucket-wide setting. (#6168)
Posted by ah...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
ahmar pushed a commit to branch feature-HADOOP-18073-sdk-v2-upgrade-3.4
in repository https://gitbox.apache.org/repos/asf/hadoop.git
commit f738f4321f4de7e13f63555bd2ddea5f6fd2a92a
Author: Steve Loughran <st...@cloudera.com>
AuthorDate: Fri Oct 27 12:23:55 2023 +0100
HADOOP-18930. Make fs.s3a.create.performance a bucket-wide setting. (#6168)
If fs.s3a.create.performance is set on a bucket
- All file overwrite checks are skipped, even if the caller says
otherwise.
- All directory existence checks are skipped.
- Marker deletion is *always* skipped.
This eliminates a HEAD and a LIST for every creation.
* New path capability "fs.s3a.create.performance.enabled" true
if the option is enabled.
* Parameterize ITestS3AContractCreate to expect the different
outcomes
* Parameterize ITestCreateFileCost similarly, with
changed cost assertions there.
* create(/) raises an IOE. existing bug only noticed here.
Contributed by Steve Loughran
---
.../filesystem/fsdataoutputstreambuilder.md | 6 ++
.../java/org/apache/hadoop/fs/s3a/Constants.java | 18 +++-
.../org/apache/hadoop/fs/s3a/S3AFileSystem.java | 54 +++++++---
.../hadoop/fs/s3a/impl/CreateFileBuilder.java | 7 ++
.../fs/contract/s3a/ITestS3AContractCreate.java | 93 ++++++++++++++++
.../hadoop/fs/s3a/ITestS3AFSMainOperations.java | 20 ++++
.../hadoop/fs/s3a/ITestS3AFileOperationCost.java | 18 +++-
.../hadoop/fs/s3a/ITestS3AFileSystemContract.java | 20 ++++
.../org/apache/hadoop/fs/s3a/S3ATestUtils.java | 13 +++
.../apache/hadoop/fs/s3a/impl/ITestXAttrCost.java | 7 +-
.../fs/s3a/performance/ITestCreateFileCost.java | 118 +++++++++++++++++----
.../performance/ITestDirectoryMarkerListing.java | 8 +-
.../fs/s3a/performance/ITestS3ADeleteCost.java | 11 ++
.../fs/s3a/tools/AbstractMarkerToolTest.java | 5 +-
14 files changed, 356 insertions(+), 42 deletions(-)
diff --git a/hadoop-common-project/hadoop-common/src/site/markdown/filesystem/fsdataoutputstreambuilder.md b/hadoop-common-project/hadoop-common/src/site/markdown/filesystem/fsdataoutputstreambuilder.md
index 59a93c5887a1..7bacf0e2436a 100644
--- a/hadoop-common-project/hadoop-common/src/site/markdown/filesystem/fsdataoutputstreambuilder.md
+++ b/hadoop-common-project/hadoop-common/src/site/markdown/filesystem/fsdataoutputstreambuilder.md
@@ -224,6 +224,12 @@ be used as evidence at the inquest as proof that they made a
conscious decision to choose speed over safety and
that the outcome was their own fault.
+Note: the option can be set for an entire filesystem. Again, the safety checks
+are there to more closely match the semantics of a classic filesystem,
+and to reduce the likelihood that the object store ends up in a state which
+diverges so much from the classic directory + tree structur that applications
+get confused.
+
Accordingly: *Use if and only if you are confident that the conditions are met.*
### `fs.s3a.create.header` User-supplied header support
diff --git a/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/Constants.java b/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/Constants.java
index 944ab5dcdcb0..16012871ff48 100644
--- a/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/Constants.java
+++ b/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/Constants.java
@@ -1192,12 +1192,26 @@ public final class Constants {
/**
* Flag for create performance.
- * This is *not* a configuration option; it is for use in the
- * {code createFile()} builder.
+ * This can be set in the {code createFile()} builder.
* Value {@value}.
*/
public static final String FS_S3A_CREATE_PERFORMANCE = "fs.s3a.create.performance";
+ /**
+ * Default value for create performance in an S3A FS.
+ * Value {@value}.
+ */
+ public static final boolean FS_S3A_CREATE_PERFORMANCE_DEFAULT = true;
+
+
+ /**
+ * Capability to indicate that the FS has been instantiated with
+ * {@link #FS_S3A_CREATE_PERFORMANCE} set to true.
+ * Value {@value}.
+ */
+ public static final String FS_S3A_CREATE_PERFORMANCE_ENABLED =
+ FS_S3A_CREATE_PERFORMANCE + ".enabled";
+
/**
* Prefix for adding a header to the object when created.
* The actual value must have a "." suffix and then the actual header.
diff --git a/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/S3AFileSystem.java b/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/S3AFileSystem.java
index db2cedd37fc9..c7511adf8ab1 100644
--- a/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/S3AFileSystem.java
+++ b/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/S3AFileSystem.java
@@ -235,6 +235,7 @@ import static org.apache.hadoop.fs.s3a.commit.CommitConstants.FS_S3A_COMMITTER_S
import static org.apache.hadoop.fs.s3a.impl.CallableSupplier.submit;
import static org.apache.hadoop.fs.s3a.impl.CreateFileBuilder.OPTIONS_CREATE_FILE_NO_OVERWRITE;
import static org.apache.hadoop.fs.s3a.impl.CreateFileBuilder.OPTIONS_CREATE_FILE_OVERWRITE;
+import static org.apache.hadoop.fs.s3a.impl.CreateFileBuilder.OPTIONS_CREATE_FILE_PERFORMANCE;
import static org.apache.hadoop.fs.s3a.impl.ErrorTranslation.isObjectNotFound;
import static org.apache.hadoop.fs.s3a.impl.ErrorTranslation.isUnknownBucket;
import static org.apache.hadoop.fs.s3a.impl.InternalConstants.AP_REQUIRED_EXCEPTION;
@@ -348,7 +349,8 @@ public class S3AFileSystem extends FileSystem implements StreamCapabilities,
private S3AStatisticsContext statisticsContext;
/** Storage Statistics Bonded to the instrumentation. */
private S3AStorageStatistics storageStatistics;
-
+ /** Should all create files be "performance" unless unset. */
+ private boolean performanceCreation;
/**
* Default input policy; may be overridden in
* {@code openFile()}.
@@ -662,6 +664,11 @@ public class S3AFileSystem extends FileSystem implements StreamCapabilities,
// verify there's no S3Guard in the store config.
checkNoS3Guard(this.getUri(), getConf());
+ // performance creation flag for code which wants performance
+ // at the risk of overwrites.
+ performanceCreation = conf.getBoolean(FS_S3A_CREATE_PERFORMANCE,
+ FS_S3A_CREATE_PERFORMANCE_DEFAULT);
+ LOG.debug("{} = {}", FS_S3A_CREATE_PERFORMANCE, performanceCreation);
allowAuthoritativePaths = S3Guard.getAuthoritativePaths(this);
// directory policy, which may look at authoritative paths
@@ -1878,14 +1885,22 @@ public class S3AFileSystem extends FileSystem implements StreamCapabilities,
Progressable progress) throws IOException {
final Path path = qualify(f);
+ // work out the options to pass down
+ CreateFileBuilder.CreateFileOptions options;
+ if (performanceCreation) {
+ options = OPTIONS_CREATE_FILE_PERFORMANCE;
+ }else {
+ options = overwrite
+ ? OPTIONS_CREATE_FILE_OVERWRITE
+ : OPTIONS_CREATE_FILE_NO_OVERWRITE;
+ }
+
// the span will be picked up inside the output stream
return trackDurationAndSpan(INVOCATION_CREATE, path, () ->
innerCreateFile(path,
progress,
getActiveAuditSpan(),
- overwrite
- ? OPTIONS_CREATE_FILE_OVERWRITE
- : OPTIONS_CREATE_FILE_NO_OVERWRITE));
+ options));
}
/**
@@ -1912,14 +1927,19 @@ public class S3AFileSystem extends FileSystem implements StreamCapabilities,
final CreateFileBuilder.CreateFileOptions options) throws IOException {
auditSpan.activate();
String key = pathToKey(path);
+ if (key.isEmpty()) {
+ // no matter the creation options, root cannot be written to.
+ throw new PathIOException("/", "Can't create root path");
+ }
EnumSet<CreateFlag> flags = options.getFlags();
- boolean overwrite = flags.contains(CreateFlag.OVERWRITE);
- boolean performance = options.isPerformance();
- boolean skipProbes = performance || isUnderMagicCommitPath(path);
+
+ boolean skipProbes = options.isPerformance() || isUnderMagicCommitPath(path);
if (skipProbes) {
LOG.debug("Skipping existence/overwrite checks");
} else {
try {
+ boolean overwrite = flags.contains(CreateFlag.OVERWRITE);
+
// get the status or throw an FNFE.
// when overwriting, there is no need to look for any existing file,
// just a directory (for safety)
@@ -1951,7 +1971,7 @@ public class S3AFileSystem extends FileSystem implements StreamCapabilities,
// put options are derived from the path and the
// option builder.
- boolean keep = performance || keepDirectoryMarkers(path);
+ boolean keep = options.isPerformance() || keepDirectoryMarkers(path);
final PutObjectOptions putOptions =
new PutObjectOptions(keep, null, options.getHeaders());
@@ -2031,11 +2051,14 @@ public class S3AFileSystem extends FileSystem implements StreamCapabilities,
final AuditSpan span = entryPoint(INVOCATION_CREATE_FILE,
pathToKey(qualified),
null);
- return new CreateFileBuilder(this,
+ final CreateFileBuilder builder = new CreateFileBuilder(this,
qualified,
- new CreateFileBuilderCallbacksImpl(INVOCATION_CREATE_FILE, span))
- .create()
- .overwrite(true);
+ new CreateFileBuilderCallbacksImpl(INVOCATION_CREATE_FILE, span));
+ builder
+ .create()
+ .overwrite(true)
+ .must(FS_S3A_CREATE_PERFORMANCE, performanceCreation);
+ return builder;
} catch (IOException e) {
// catch any IOEs raised in span creation and convert to
// an UncheckedIOException
@@ -2098,7 +2121,8 @@ public class S3AFileSystem extends FileSystem implements StreamCapabilities,
.create()
.withFlags(flags)
.blockSize(blockSize)
- .bufferSize(bufferSize);
+ .bufferSize(bufferSize)
+ .must(FS_S3A_CREATE_PERFORMANCE, performanceCreation);
if (progress != null) {
builder.progress(progress);
}
@@ -5365,6 +5389,10 @@ public class S3AFileSystem extends FileSystem implements StreamCapabilities,
case FS_S3A_CREATE_HEADER:
return true;
+ // is the FS configured for create file performance
+ case FS_S3A_CREATE_PERFORMANCE_ENABLED:
+ return performanceCreation;
+
default:
return super.hasPathCapability(p, cap);
}
diff --git a/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/impl/CreateFileBuilder.java b/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/impl/CreateFileBuilder.java
index 0392afac59d9..ae2945989ddd 100644
--- a/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/impl/CreateFileBuilder.java
+++ b/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/impl/CreateFileBuilder.java
@@ -71,6 +71,12 @@ public class CreateFileBuilder extends
public static final CreateFileOptions OPTIONS_CREATE_FILE_NO_OVERWRITE =
new CreateFileOptions(CREATE_NO_OVERWRITE_FLAGS, true, false, null);
+ /**
+ * Performance create options.
+ */
+ public static final CreateFileOptions OPTIONS_CREATE_FILE_PERFORMANCE =
+ new CreateFileOptions(CREATE_OVERWRITE_FLAGS, true, true, null);
+
/**
* Callback interface.
*/
@@ -129,6 +135,7 @@ public class CreateFileBuilder extends
if (flags.contains(CreateFlag.APPEND)) {
throw new UnsupportedOperationException("Append is not supported");
}
+
if (!flags.contains(CreateFlag.CREATE) &&
!flags.contains(CreateFlag.OVERWRITE)) {
throw new PathIOException(path.toString(),
diff --git a/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/contract/s3a/ITestS3AContractCreate.java b/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/contract/s3a/ITestS3AContractCreate.java
index d2a858f615ef..7a2a10879dd8 100644
--- a/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/contract/s3a/ITestS3AContractCreate.java
+++ b/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/contract/s3a/ITestS3AContractCreate.java
@@ -18,18 +18,111 @@
package org.apache.hadoop.fs.contract.s3a;
+import java.util.Arrays;
+import java.util.Collection;
+
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.contract.AbstractContractCreateTest;
import org.apache.hadoop.fs.contract.AbstractFSContract;
+import org.apache.hadoop.fs.s3a.S3ATestUtils;
+
+import static org.apache.hadoop.fs.s3a.Constants.FS_S3A_CREATE_PERFORMANCE;
+import static org.apache.hadoop.fs.s3a.S3ATestUtils.removeBaseAndBucketOverrides;
/**
* S3A contract tests creating files.
+ * Parameterized on the create performance flag as all overwrite
+ * tests are required to fail in create performance mode.
*/
+@RunWith(Parameterized.class)
public class ITestS3AContractCreate extends AbstractContractCreateTest {
+ /**
+ * This test suite is parameterized for the different create file
+ * options.
+ * @return a list of test parameters.
+ */
+ @Parameterized.Parameters
+ public static Collection<Object[]> params() {
+ return Arrays.asList(new Object[][]{
+ {false},
+ {true}
+ });
+ }
+
+ /**
+ * Is this test run in create performance mode?
+ */
+ private final boolean createPerformance;
+
+ public ITestS3AContractCreate(final boolean createPerformance) {
+ this.createPerformance = createPerformance;
+ }
+
@Override
protected AbstractFSContract createContract(Configuration conf) {
return new S3AContract(conf);
}
+ @Override
+ protected Configuration createConfiguration() {
+ final Configuration conf = super.createConfiguration();
+ removeBaseAndBucketOverrides(conf,
+ FS_S3A_CREATE_PERFORMANCE);
+ conf.setBoolean(FS_S3A_CREATE_PERFORMANCE, createPerformance);
+ S3ATestUtils.disableFilesystemCaching(conf);
+ return conf;
+ }
+
+ @Override
+ public void testOverwriteNonEmptyDirectory() throws Throwable {
+ try {
+ super.testOverwriteNonEmptyDirectory();
+ failWithCreatePerformance();
+ } catch (AssertionError e) {
+ swallowWithCreatePerformance(e);
+ }
+ }
+
+ @Override
+ public void testOverwriteEmptyDirectory() throws Throwable {
+ try {
+ super.testOverwriteEmptyDirectory();
+ failWithCreatePerformance();
+ } catch (AssertionError e) {
+ swallowWithCreatePerformance(e);
+ }
+ }
+
+ @Override
+ public void testCreateFileOverExistingFileNoOverwrite() throws Throwable {
+ try {
+ super.testCreateFileOverExistingFileNoOverwrite();
+ failWithCreatePerformance();
+ } catch (AssertionError e) {
+ swallowWithCreatePerformance(e);
+ }
+ }
+
+ private void failWithCreatePerformance() {
+ if (createPerformance) {
+ fail("expected an assertion error in create performance mode");
+ }
+ }
+
+ /**
+ * Swallow an assertion error if the create performance flag is set.
+ * @param e assertion error
+ */
+ private void swallowWithCreatePerformance(final AssertionError e) {
+ // this is expected in create performance modea
+ if (!createPerformance) {
+ // but if the create performance flag is set, then it is supported
+ // and the assertion error is unexpected
+ throw e;
+ }
+ }
}
diff --git a/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/ITestS3AFSMainOperations.java b/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/ITestS3AFSMainOperations.java
index 6669e8426af0..013ec901d0a7 100644
--- a/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/ITestS3AFSMainOperations.java
+++ b/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/ITestS3AFSMainOperations.java
@@ -18,6 +18,9 @@
package org.apache.hadoop.fs.s3a;
+import java.io.IOException;
+
+import org.assertj.core.api.Assertions;
import org.junit.Ignore;
import org.apache.hadoop.conf.Configuration;
@@ -27,6 +30,7 @@ import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.contract.s3a.S3AContract;
import static org.apache.hadoop.fs.s3a.S3ATestUtils.createTestPath;
+import static org.apache.hadoop.fs.s3a.S3ATestUtils.isCreatePerformanceEnabled;
/**
* S3A Test suite for the FSMainOperationsBaseTest tests.
@@ -70,4 +74,20 @@ public class ITestS3AFSMainOperations extends FSMainOperationsBaseTest {
throws Exception {
}
+ @Override
+ public void testOverwrite() throws IOException {
+ boolean createPerformance = isCreatePerformanceEnabled(fSys);
+ try {
+ super.testOverwrite();
+ Assertions.assertThat(createPerformance)
+ .describedAs("create performance enabled")
+ .isFalse();
+ } catch (AssertionError e) {
+ // swallow the exception if create performance is enabled,
+ // else rethrow
+ if (!createPerformance) {
+ throw e;
+ }
+ }
+ }
}
diff --git a/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/ITestS3AFileOperationCost.java b/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/ITestS3AFileOperationCost.java
index dae6312d4809..0e4a8eda5b29 100644
--- a/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/ITestS3AFileOperationCost.java
+++ b/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/ITestS3AFileOperationCost.java
@@ -18,6 +18,7 @@
package org.apache.hadoop.fs.s3a;
+import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.s3a.impl.StatusProbeEnum;
@@ -39,6 +40,8 @@ import java.util.EnumSet;
import static org.apache.hadoop.fs.contract.ContractTestUtils.*;
+import static org.apache.hadoop.fs.s3a.Constants.FS_S3A_CREATE_PERFORMANCE;
+import static org.apache.hadoop.fs.s3a.S3ATestUtils.removeBaseAndBucketOverrides;
import static org.apache.hadoop.fs.s3a.Statistic.*;
import static org.apache.hadoop.fs.s3a.performance.OperationCost.*;
import static org.apache.hadoop.test.GenericTestUtils.getTestDir;
@@ -47,6 +50,9 @@ import static org.apache.hadoop.test.LambdaTestUtils.intercept;
/**
* Use metrics to assert about the cost of file API calls.
* Parameterized on directory marker keep vs delete.
+ * When the FS is instantiated with creation performance, things
+ * behave differently...its value is that of the marker keep flag,
+ * so deletion costs are the same.
*/
@RunWith(Parameterized.class)
public class ITestS3AFileOperationCost extends AbstractS3ACostTest {
@@ -71,6 +77,14 @@ public class ITestS3AFileOperationCost extends AbstractS3ACostTest {
super(keepMarkers);
}
+ @Override
+ public Configuration createConfiguration() {
+ final Configuration conf = super.createConfiguration();
+ removeBaseAndBucketOverrides(conf, FS_S3A_CREATE_PERFORMANCE);
+ conf.setBoolean(FS_S3A_CREATE_PERFORMANCE, isKeepingMarkers());
+ return conf;
+ }
+
/**
* Test the cost of {@code listLocatedStatus(file)}.
*/
@@ -377,7 +391,7 @@ public class ITestS3AFileOperationCost extends AbstractS3ACostTest {
// create a bunch of files
int filesToCreate = 10;
for (int i = 0; i < filesToCreate; i++) {
- create(basePath.suffix("/" + i));
+ file(basePath.suffix("/" + i));
}
fs.globStatus(basePath.suffix("/*"));
@@ -396,7 +410,7 @@ public class ITestS3AFileOperationCost extends AbstractS3ACostTest {
// create a single file, globStatus returning a single file on a pattern
// triggers attempts at symlinks resolution if configured
String fileName = "/notASymlinkDOntResolveMeLikeOne";
- create(basePath.suffix(fileName));
+ file(basePath.suffix(fileName));
// unguarded: 2 head + 1 list from getFileStatus on path,
// plus 1 list to match the glob pattern
// no additional operations from symlink resolution
diff --git a/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/ITestS3AFileSystemContract.java b/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/ITestS3AFileSystemContract.java
index 7ce7b8385cec..56827043c9b8 100644
--- a/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/ITestS3AFileSystemContract.java
+++ b/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/ITestS3AFileSystemContract.java
@@ -19,7 +19,9 @@
package org.apache.hadoop.fs.s3a;
import java.io.FileNotFoundException;
+import java.io.IOException;
+import org.assertj.core.api.Assertions;
import org.junit.Before;
import org.junit.Rule;
import org.junit.Test;
@@ -32,6 +34,7 @@ import org.apache.hadoop.fs.FileSystemContractBaseTest;
import org.apache.hadoop.fs.Path;
import static org.apache.hadoop.fs.contract.ContractTestUtils.skip;
+import static org.apache.hadoop.fs.s3a.S3ATestUtils.isCreatePerformanceEnabled;
import static org.apache.hadoop.test.LambdaTestUtils.intercept;
import static org.junit.Assume.*;
import static org.junit.Assert.*;
@@ -137,4 +140,21 @@ public class ITestS3AFileSystemContract extends FileSystemContractBaseTest {
() -> super.testRenameNonExistentPath());
}
+
+ @Override
+ public void testOverwrite() throws IOException {
+ boolean createPerformance = isCreatePerformanceEnabled(fs);
+ try {
+ super.testOverwrite();
+ Assertions.assertThat(createPerformance)
+ .describedAs("create performance enabled")
+ .isFalse();
+ } catch (AssertionError e) {
+ // swallow the exception if create performance is enabled,
+ // else rethrow
+ if (!createPerformance) {
+ throw e;
+ }
+ }
+ }
}
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 3382c300b931..aa38186c6503 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
@@ -1554,4 +1554,17 @@ public final class S3ATestUtils {
return fs.getConf().getBoolean(Constants.ENABLE_MULTI_DELETE,
true);
}
+
+ /**
+ * Does this FS have create performance enabled?
+ * @param fs filesystem
+ * @return true if create performance is enabled
+ * @throws IOException IO problems
+ */
+ public static boolean isCreatePerformanceEnabled(FileSystem fs)
+ throws IOException {
+ return fs.hasPathCapability(new Path("/"), FS_S3A_CREATE_PERFORMANCE_ENABLED);
+ }
+
+
}
diff --git a/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/impl/ITestXAttrCost.java b/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/impl/ITestXAttrCost.java
index f6fe68b8e6d3..71fdb7aeaa62 100644
--- a/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/impl/ITestXAttrCost.java
+++ b/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/impl/ITestXAttrCost.java
@@ -33,6 +33,7 @@ import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.s3a.S3AFileSystem;
import org.apache.hadoop.fs.s3a.performance.AbstractS3ACostTest;
+import static org.apache.hadoop.fs.s3a.S3ATestUtils.isCreatePerformanceEnabled;
import static org.apache.hadoop.fs.s3a.Statistic.INVOCATION_OP_XATTR_LIST;
import static org.apache.hadoop.fs.s3a.Statistic.INVOCATION_XATTR_GET_MAP;
import static org.apache.hadoop.fs.s3a.Statistic.INVOCATION_XATTR_GET_NAMED;
@@ -43,6 +44,7 @@ import static org.apache.hadoop.fs.s3a.impl.HeaderProcessing.XA_CONTENT_TYPE;
import static org.apache.hadoop.fs.s3a.impl.HeaderProcessing.XA_STANDARD_HEADERS;
import static org.apache.hadoop.fs.s3a.impl.HeaderProcessing.decodeBytes;
import static org.apache.hadoop.fs.s3a.performance.OperationCost.CREATE_FILE_OVERWRITE;
+import static org.apache.hadoop.fs.s3a.performance.OperationCost.NO_HEAD_OR_LIST;
/**
* Invoke XAttr API calls against objects in S3 and validate header
@@ -95,8 +97,11 @@ public class ITestXAttrCost extends AbstractS3ACostTest {
public void testXAttrFile() throws Throwable {
describe("Test xattr on a file");
Path testFile = methodPath();
- create(testFile, true, CREATE_FILE_OVERWRITE);
S3AFileSystem fs = getFileSystem();
+ boolean createPerformance = isCreatePerformanceEnabled(fs);
+
+ create(testFile, true,
+ createPerformance ? NO_HEAD_OR_LIST : CREATE_FILE_OVERWRITE);
Map<String, byte[]> xAttrs = verifyMetrics(() ->
fs.getXAttrs(testFile),
with(INVOCATION_XATTR_GET_MAP, GET_METADATA_ON_OBJECT));
diff --git a/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/performance/ITestCreateFileCost.java b/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/performance/ITestCreateFileCost.java
index 39530d97bf79..2d128cffc5af 100644
--- a/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/performance/ITestCreateFileCost.java
+++ b/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/performance/ITestCreateFileCost.java
@@ -19,16 +19,22 @@
package org.apache.hadoop.fs.s3a.performance;
import java.io.IOException;
+import java.util.Arrays;
+import java.util.Collection;
import org.assertj.core.api.Assertions;
import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FSDataOutputStreamBuilder;
import org.apache.hadoop.fs.FileAlreadyExistsException;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.contract.ContractTestUtils;
import org.apache.hadoop.fs.s3a.S3AFileSystem;
+import org.apache.hadoop.fs.s3a.S3ATestUtils;
import static java.util.Objects.requireNonNull;
@@ -36,6 +42,7 @@ import static org.apache.hadoop.fs.contract.ContractTestUtils.toChar;
import static org.apache.hadoop.fs.s3a.Constants.FS_S3A_CREATE_HEADER;
import static org.apache.hadoop.fs.s3a.Constants.FS_S3A_CREATE_PERFORMANCE;
import static org.apache.hadoop.fs.s3a.Constants.XA_HEADER_PREFIX;
+import static org.apache.hadoop.fs.s3a.S3ATestUtils.removeBaseAndBucketOverrides;
import static org.apache.hadoop.fs.s3a.Statistic.OBJECT_BULK_DELETE_REQUEST;
import static org.apache.hadoop.fs.s3a.Statistic.OBJECT_DELETE_REQUEST;
import static org.apache.hadoop.fs.s3a.performance.OperationCost.CREATE_FILE_NO_OVERWRITE;
@@ -45,6 +52,7 @@ import static org.apache.hadoop.fs.s3a.performance.OperationCost.FILE_STATUS_FIL
import static org.apache.hadoop.fs.s3a.performance.OperationCost.GET_FILE_STATUS_ON_DIR_MARKER;
import static org.apache.hadoop.fs.s3a.performance.OperationCost.GET_FILE_STATUS_ON_FILE;
import static org.apache.hadoop.fs.s3a.performance.OperationCost.HEAD_OPERATION;
+import static org.apache.hadoop.fs.s3a.performance.OperationCost.LIST_OPERATION;
import static org.apache.hadoop.fs.s3a.performance.OperationCost.NO_HEAD_OR_LIST;
/**
@@ -52,13 +60,55 @@ import static org.apache.hadoop.fs.s3a.performance.OperationCost.NO_HEAD_OR_LIST
* with the FS_S3A_CREATE_PERFORMANCE option.
*/
@SuppressWarnings("resource")
+@RunWith(Parameterized.class)
public class ITestCreateFileCost extends AbstractS3ACostTest {
+ /**
+ * This test suite is parameterized for the different create file
+ * options.
+ * @return a list of test parameters.
+ */
+ @Parameterized.Parameters
+ public static Collection<Object[]> params() {
+ return Arrays.asList(new Object[][]{
+ {false},
+ {true}
+ });
+ }
+
+ /**
+ * Flag for performance creation; all cost asserts need changing.
+ */
+ private final boolean createPerformance;
+
/**
* Create with markers kept, always.
*/
- public ITestCreateFileCost() {
+ public ITestCreateFileCost(final boolean createPerformance) {
+ // keep markers to permit assertions that create performance
+ // always skips marker deletion.
super(false);
+ this.createPerformance = createPerformance;
+ }
+
+ /**
+ * Determine the expected cost of a create operation;
+ * if {@link #createPerformance} is true, then the cost is always "no IO".
+ * @param source source cost
+ * @return cost to assert
+ */
+ private OperationCost expected(OperationCost source) {
+ return createPerformance ? NO_HEAD_OR_LIST : source;
+ }
+
+ @Override
+ public Configuration createConfiguration() {
+ final Configuration conf = super.createConfiguration();
+ removeBaseAndBucketOverrides(conf,
+ FS_S3A_CREATE_PERFORMANCE);
+ conf.setBoolean(FS_S3A_CREATE_PERFORMANCE, createPerformance);
+ S3ATestUtils.disableFilesystemCaching(conf);
+ return conf;
}
@Test
@@ -67,7 +117,7 @@ public class ITestCreateFileCost extends AbstractS3ACostTest {
Path testFile = methodPath();
// when overwrite is false, the path is checked for existence.
create(testFile, false,
- CREATE_FILE_NO_OVERWRITE);
+ expected(CREATE_FILE_NO_OVERWRITE));
}
@Test
@@ -75,7 +125,7 @@ public class ITestCreateFileCost extends AbstractS3ACostTest {
describe("Test file creation with overwrite");
Path testFile = methodPath();
// when overwrite is true: only the directory checks take place.
- create(testFile, true, CREATE_FILE_OVERWRITE);
+ create(testFile, true, expected(CREATE_FILE_OVERWRITE));
}
@Test
@@ -85,21 +135,45 @@ public class ITestCreateFileCost extends AbstractS3ACostTest {
// now there is a file there, an attempt with overwrite == false will
// fail on the first HEAD.
- interceptOperation(FileAlreadyExistsException.class, "",
- FILE_STATUS_FILE_PROBE,
- () -> file(testFile, false));
+ if (!createPerformance) {
+ interceptOperation(FileAlreadyExistsException.class, "",
+ FILE_STATUS_FILE_PROBE,
+ () -> file(testFile, false));
+ } else {
+ create(testFile, false, NO_HEAD_OR_LIST);
+ }
}
@Test
- public void testCreateFileOverDir() throws Throwable {
- describe("Test cost of create file failing with existing dir");
+ public void testCreateFileOverDirNoOverwrite() throws Throwable {
+ describe("Test cost of create file overwrite=false failing with existing dir");
Path testFile = dir(methodPath());
- // now there is a file there, an attempt with overwrite == false will
+ // now there is a dir marker there, an attempt with overwrite == true will
// fail on the first HEAD.
- interceptOperation(FileAlreadyExistsException.class, "",
- GET_FILE_STATUS_ON_DIR_MARKER,
- () -> file(testFile, false));
+ if (!createPerformance) {
+ interceptOperation(FileAlreadyExistsException.class, "",
+ GET_FILE_STATUS_ON_DIR_MARKER,
+ () -> file(testFile, false));
+ } else {
+ create(testFile, false, NO_HEAD_OR_LIST);
+ }
+ }
+
+ @Test
+ public void testCreateFileOverDirWithOverwrite() throws Throwable {
+ describe("Test cost of create file overwrite=false failing with existing dir");
+ Path testFile = dir(methodPath());
+
+ // now there is a dir marker there, an attempt with overwrite == true will
+ // fail on the LIST; no HEAD is issued.
+ if (!createPerformance) {
+ interceptOperation(FileAlreadyExistsException.class, "",
+ LIST_OPERATION,
+ () -> file(testFile, true));
+ } else {
+ create(testFile, true, NO_HEAD_OR_LIST);
+ }
}
/**
@@ -117,14 +191,18 @@ public class ITestCreateFileCost extends AbstractS3ACostTest {
// files and so briefly the path not being present
// only make sure the dest path isn't a directory.
buildFile(testFile, true, false,
- FILE_STATUS_DIR_PROBE);
+ expected(FILE_STATUS_DIR_PROBE));
// now there is a file there, an attempt with overwrite == false will
// fail on the first HEAD.
- interceptOperation(FileAlreadyExistsException.class, "",
- GET_FILE_STATUS_ON_FILE,
- () -> buildFile(testFile, false, true,
- GET_FILE_STATUS_ON_FILE));
+ if (!createPerformance) {
+ interceptOperation(FileAlreadyExistsException.class, "",
+ GET_FILE_STATUS_ON_FILE,
+ () -> buildFile(testFile, false, true,
+ GET_FILE_STATUS_ON_FILE));
+ } else {
+ buildFile(testFile, false, true, NO_HEAD_OR_LIST);
+ }
}
@Test
@@ -162,7 +240,7 @@ public class ITestCreateFileCost extends AbstractS3ACostTest {
builder.must(FS_S3A_CREATE_HEADER + ".h1", custom);
verifyMetrics(() -> build(builder),
- always(CREATE_FILE_NO_OVERWRITE));
+ always(expected(CREATE_FILE_NO_OVERWRITE)));
// the header is there and the probe should be a single HEAD call.
String header = verifyMetrics(() ->
@@ -181,7 +259,7 @@ public class ITestCreateFileCost extends AbstractS3ACostTest {
verifyMetrics(() ->
build(fs.createFile(methodPath()).overwrite(true)),
- always(CREATE_FILE_OVERWRITE));
+ always(expected(CREATE_FILE_OVERWRITE)));
}
@@ -196,7 +274,7 @@ public class ITestCreateFileCost extends AbstractS3ACostTest {
.close();
return "";
},
- always(CREATE_FILE_OVERWRITE));
+ always(expected(CREATE_FILE_OVERWRITE)));
}
private FSDataOutputStream build(final FSDataOutputStreamBuilder builder)
diff --git a/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/performance/ITestDirectoryMarkerListing.java b/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/performance/ITestDirectoryMarkerListing.java
index cd4ee4440667..8dbf5baaa620 100644
--- a/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/performance/ITestDirectoryMarkerListing.java
+++ b/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/performance/ITestDirectoryMarkerListing.java
@@ -54,6 +54,7 @@ import static org.apache.hadoop.fs.contract.ContractTestUtils.touch;
import static org.apache.hadoop.fs.s3a.Constants.DIRECTORY_MARKER_POLICY;
import static org.apache.hadoop.fs.s3a.Constants.DIRECTORY_MARKER_POLICY_DELETE;
import static org.apache.hadoop.fs.s3a.Constants.DIRECTORY_MARKER_POLICY_KEEP;
+import static org.apache.hadoop.fs.s3a.Constants.FS_S3A_CREATE_PERFORMANCE;
import static org.apache.hadoop.fs.s3a.S3ATestUtils.getTestBucketName;
import static org.apache.hadoop.fs.s3a.S3ATestUtils.removeBaseAndBucketOverrides;
import static org.apache.hadoop.test.LambdaTestUtils.intercept;
@@ -80,8 +81,7 @@ import static org.apache.hadoop.util.functional.RemoteIterators.foreach;
* <p></p>
* Similarly: JUnit assertions over AssertJ.
* <p></p>
- * The tests work with unguarded buckets only -the bucket settings are changed
- * appropriately.
+ * s3a create performance is disabled for consistent assertions.
*/
@RunWith(Parameterized.class)
public class ITestDirectoryMarkerListing extends AbstractS3ATestBase {
@@ -199,11 +199,13 @@ public class ITestDirectoryMarkerListing extends AbstractS3ATestBase {
// directory marker options
removeBaseAndBucketOverrides(bucketName, conf,
- DIRECTORY_MARKER_POLICY);
+ DIRECTORY_MARKER_POLICY,
+ FS_S3A_CREATE_PERFORMANCE);
conf.set(DIRECTORY_MARKER_POLICY,
keepMarkers
? DIRECTORY_MARKER_POLICY_KEEP
: DIRECTORY_MARKER_POLICY_DELETE);
+ conf.setBoolean(FS_S3A_CREATE_PERFORMANCE, false);
return conf;
}
diff --git a/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/performance/ITestS3ADeleteCost.java b/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/performance/ITestS3ADeleteCost.java
index fae2df973af7..97f51fe2c8dc 100644
--- a/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/performance/ITestS3ADeleteCost.java
+++ b/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/performance/ITestS3ADeleteCost.java
@@ -30,6 +30,7 @@ import org.junit.runners.Parameterized;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.contract.ContractTestUtils;
@@ -38,6 +39,8 @@ import org.apache.hadoop.fs.s3a.S3AFileSystem;
import org.apache.hadoop.fs.s3a.Tristate;
import org.apache.hadoop.fs.s3a.impl.StatusProbeEnum;
+import static org.apache.hadoop.fs.s3a.Constants.FS_S3A_CREATE_PERFORMANCE;
+import static org.apache.hadoop.fs.s3a.S3ATestUtils.removeBaseAndBucketOverrides;
import static org.apache.hadoop.fs.s3a.Statistic.*;
import static org.apache.hadoop.fs.s3a.performance.OperationCost.*;
import static org.apache.hadoop.fs.s3a.performance.OperationCostValidator.probe;
@@ -74,6 +77,14 @@ public class ITestS3ADeleteCost extends AbstractS3ACostTest {
super(keepMarkers);
}
+ @Override
+ public Configuration createConfiguration() {
+ Configuration conf = super.createConfiguration();
+ removeBaseAndBucketOverrides(conf, FS_S3A_CREATE_PERFORMANCE);
+ conf.setBoolean(FS_S3A_CREATE_PERFORMANCE, false);
+ return conf;
+ }
+
@Override
public void teardown() throws Exception {
if (isKeepingMarkers()) {
diff --git a/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/tools/AbstractMarkerToolTest.java b/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/tools/AbstractMarkerToolTest.java
index 0c2473a5f61d..759a3bf129ee 100644
--- a/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/tools/AbstractMarkerToolTest.java
+++ b/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/tools/AbstractMarkerToolTest.java
@@ -73,12 +73,15 @@ public class AbstractMarkerToolTest extends AbstractS3ATestBase {
removeBaseAndBucketOverrides(bucketName, conf,
S3A_BUCKET_PROBE,
DIRECTORY_MARKER_POLICY,
- AUTHORITATIVE_PATH);
+ AUTHORITATIVE_PATH,
+ FS_S3A_CREATE_PERFORMANCE);
// base FS is legacy
conf.set(DIRECTORY_MARKER_POLICY, DIRECTORY_MARKER_POLICY_DELETE);
+ conf.setBoolean(FS_S3A_CREATE_PERFORMANCE, false);
// turn off bucket probes for a bit of speedup in the connectors we create.
conf.setInt(S3A_BUCKET_PROBE, 0);
+
return conf;
}
---------------------------------------------------------------------
To unsubscribe, e-mail: common-commits-unsubscribe@hadoop.apache.org
For additional commands, e-mail: common-commits-help@hadoop.apache.org
(hadoop) 05/05: HADOOP-18996. S3A to provide full support for S3 Express One Zone (#6308)
Posted by ah...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
ahmar pushed a commit to branch feature-HADOOP-18073-sdk-v2-upgrade-3.4
in repository https://gitbox.apache.org/repos/asf/hadoop.git
commit c2c9e388526296d5ac047f084fd71cda8b1d2b25
Author: Steve Loughran <st...@cloudera.com>
AuthorDate: Fri Dec 1 14:16:33 2023 +0000
HADOOP-18996. S3A to provide full support for S3 Express One Zone (#6308)
This adds borad support for Amazon S3 Express One Zone to the S3A connector,
particularly resilience of other parts of the codebase to LIST operations returning
paths under which only in-progress uploads are taking place.
hadoop-common and hadoop-mapreduce treewalking routines all cope with this;
distcp is left alone.
There are still some outstanding followup issues, and we expect more to surface
with extended use.
Contains HADOOP-18955. AWS SDK v2: add path capability probe "fs.s3a.capability.aws.v2
* lets us probe for AWS SDK version
* bucket-info reports it
Contains HADOOP-18961 S3A: add s3guard command "bucket"
hadoop s3guard bucket -create -region us-west-2 -zone usw2-az2 \
s3a://stevel--usw2-az2--x-s3/
* requires -zone if bucket is zonal
* rejects it if not
* rejects zonal bucket suffixes if endpoint is not aws (safety feature)
* imperfect, but a functional starting point.
New path capability "fs.s3a.capability.zonal.storage"
* Used in tests to determine whether pending uploads manifest paths
* cli tests can probe for this
* bucket-info reports it
* some tests disable/change assertions as appropriate
----
Shell commands fail on S3Express buckets if pending uploads.
New path capability in hadoop-common
"fs.capability.directory.listing.inconsistent"
1. S3AFS returns true on a S3 Express bucket
2. FileUtil.maybeIgnoreMissingDirectory(fs, path, fnfe)
decides whether to swallow the exception or not.
3. This is used in: Shell, FileInputFormat, LocatedFileStatusFetcher
Fixes with tests
* fs -ls -R
* fs -du
* fs -df
* fs -find
* S3AFS.getContentSummary() (maybe...should discuss)
* mapred LocatedFileStatusFetcher
* Globber, HADOOP-15478 already fixed that when dealing with
S3 inconsistencies
* FileInputFormat
S3Express CreateSession request is permitted outside audit spans.
S3 Bulk Delete calls request the store to return the list of deleted objects
if RequestFactoryImpl is set to trace.
log4j.logger.org.apache.hadoop.fs.s3a.impl.RequestFactoryImpl=TRACE
Test Changes
* ITestS3AMiscOperations removes all tests which require unencrypted
buckets. AWS S3 defaults to SSE-S3 everywhere.
* ITestBucketTool to test new tool without actually creating new
buckets.
* S3ATestUtils add methods to skip test suites/cases if store is/is not
S3Express
* Cutting down on "is this a S3Express bucket" logic to trailing --x-s3 string
and not worrying about AZ naming logic. commented out relevant tests.
* ITestTreewalkProblems validated against standard and S3Express stores
Outstanding
* Distcp: tests show it fails. Proposed: release notes.
---
x-amz-checksum header not found when signing S3Express messages
This modifies the custom signer in ITestCustomSigner to be a subclass
of AwsS3V4Signer with a goal of preventing signing problems with
S3 Express stores.
----
RemoteFileChanged renaming multipart file
Maps 412 status code to RemoteFileChangedException
Modifies huge file tests
-Adds a check on etag match for stat vs list
-ITestS3AHugeFilesByteBufferBlocks renames parent dirs, rather than
files, to replicate distcp better.
----
S3Express custom Signing cannot handle bulk delete
Copy custom signer into production JAR, so enable downstream testing
Extend ITestCustomSigner to cover more filesystem operations
- PUT
- POST
- COPY
- LIST
- Bulk delete through delete() and rename()
- list + abort multipart uploads
Suite is parameterized on bulk delete enabled/disabled.
To use the new signer for a full test run:
<property>
<name>fs.s3a.custom.signers</name>
<value>CustomSdkSigner:org.apache.hadoop.fs.s3a.auth.CustomSdkSigner</value>
</property>
<property>
<name>fs.s3a.s3.signing-algorithm</name>
<value>CustomSdkSigner</value>
</property>
---
.../apache/hadoop/fs/CommonPathCapabilities.java | 11 +
.../main/java/org/apache/hadoop/fs/FileUtil.java | 29 ++
.../java/org/apache/hadoop/fs/shell/Command.java | 17 +-
.../org/apache/hadoop/fs/FileContextURIBase.java | 32 +-
.../AbstractContractRootDirectoryTest.java | 27 +-
.../hadoop/fs/contract/ContractTestUtils.java | 66 ++-
.../hadoop/mapred/LocatedFileStatusFetcher.java | 34 +-
.../mapreduce/lib/input/FileInputFormat.java | 30 +-
.../java/org/apache/hadoop/fs/s3a/Constants.java | 18 +-
.../java/org/apache/hadoop/fs/s3a/Listing.java | 2 +-
.../org/apache/hadoop/fs/s3a/S3AFileSystem.java | 112 ++++-
.../org/apache/hadoop/fs/s3a/S3AInternals.java | 10 +
.../java/org/apache/hadoop/fs/s3a/S3AUtils.java | 16 +-
.../hadoop/fs/s3a/audit/AWSRequestAnalyzer.java | 4 +-
.../apache/hadoop/fs/s3a/auth/CustomSdkSigner.java | 178 ++++++++
.../apache/hadoop/fs/s3a/impl/DeleteOperation.java | 20 +-
.../hadoop/fs/s3a/impl/InternalConstants.java | 47 ++-
.../apache/hadoop/fs/s3a/impl/NetworkBinding.java | 11 +
.../hadoop/fs/s3a/impl/RequestFactoryImpl.java | 2 +-
.../hadoop/fs/s3a/impl/S3ExpressStorage.java | 76 ++++
.../apache/hadoop/fs/s3a/s3guard/S3GuardTool.java | 52 ++-
.../org/apache/hadoop/fs/s3a/tools/BucketTool.java | 304 ++++++++++++++
.../org/apache/hadoop/fs/s3a/tools/MarkerTool.java | 8 +-
.../markdown/tools/hadoop-aws/aws_sdk_upgrade.md | 10 +
.../tools/hadoop-aws/third_party_stores.md | 11 +-
.../hadoop/fs/s3a/AbstractTestS3AEncryption.java | 1 +
.../hadoop/fs/s3a/ITestS3AConfiguration.java | 2 +-
.../hadoop/fs/s3a/ITestS3AEncryptionSSEC.java | 2 +-
.../hadoop/fs/s3a/ITestS3AMiscOperations.java | 77 ----
.../apache/hadoop/fs/s3a/MultipartTestUtils.java | 55 ++-
.../org/apache/hadoop/fs/s3a/S3ATestUtils.java | 76 ++++
.../hadoop/fs/s3a/auth/ITestCustomSigner.java | 124 ++++--
.../fs/s3a/commit/ITestCommitOperations.java | 6 +-
.../hadoop/fs/s3a/impl/ITestTreewalkProblems.java | 463 +++++++++++++++++++++
.../ITestUploadPurgeOnDirectoryOperations.java | 22 +-
.../hadoop/fs/s3a/impl/TestS3ExpressStorage.java | 63 +++
.../performance/ITestDirectoryMarkerListing.java | 23 +-
.../s3a/s3guard/AbstractS3GuardToolTestBase.java | 6 +-
.../fs/s3a/scale/AbstractSTestS3AHugeFiles.java | 72 +++-
.../scale/ITestS3AHugeFilesByteBufferBlocks.java | 29 ++
.../hadoop/fs/s3a/tools/ITestBucketTool.java | 183 ++++++++
.../hadoop-aws/src/test/resources/core-site.xml | 2 +-
.../hadoop-aws/src/test/resources/log4j.properties | 4 +
.../apache/hadoop/tools/util/DistCpTestUtils.java | 10 +-
44 files changed, 2032 insertions(+), 315 deletions(-)
diff --git a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/CommonPathCapabilities.java b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/CommonPathCapabilities.java
index b898fde43cc8..9ec07cbe966e 100644
--- a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/CommonPathCapabilities.java
+++ b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/CommonPathCapabilities.java
@@ -170,4 +170,15 @@ public final class CommonPathCapabilities {
*/
public static final String LEASE_RECOVERABLE = "fs.capability.lease.recoverable";
+ /**
+ * Is this a store where parent directory listings are potentially inconsistent with
+ * direct list/getFileStatus calls?
+ * This can happen with Amazon S3 Express One Zone Storage when there are pending
+ * uploads under a path.
+ * Application code can use this flag to decide whether or not to treat
+ * FileNotFoundExceptions on treewalk as errors or something to downgrade.
+ * Value: {@value}.
+ */
+ public static final String DIRECTORY_LISTING_INCONSISTENT =
+ "fs.capability.directory.listing.inconsistent";
}
diff --git a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/FileUtil.java b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/FileUtil.java
index 933f56927741..79041e0d7b94 100644
--- a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/FileUtil.java
+++ b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/FileUtil.java
@@ -76,6 +76,7 @@ import org.apache.hadoop.util.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import static org.apache.hadoop.fs.CommonPathCapabilities.DIRECTORY_LISTING_INCONSISTENT;
import static org.apache.hadoop.fs.Options.OpenFileOptions.FS_OPTION_OPENFILE_READ_POLICY;
import static org.apache.hadoop.fs.Options.OpenFileOptions.FS_OPTION_OPENFILE_LENGTH;
import static org.apache.hadoop.fs.Options.OpenFileOptions.FS_OPTION_OPENFILE_READ_POLICY_WHOLE_FILE;
@@ -2058,4 +2059,32 @@ public class FileUtil {
final Options.Rename... options) throws IOException {
srcFs.rename(src, dst, options);
}
+
+ /**
+ * Method to call after a FNFE has been raised on a treewalk, so as to
+ * decide whether to throw the exception (default), or, if the FS
+ * supports inconsistent directory listings, to log and ignore it.
+ * If this returns then the caller should ignore the failure and continue.
+ * @param fs filesystem
+ * @param path path
+ * @param e exception caught
+ * @throws FileNotFoundException the exception passed in, if rethrown.
+ */
+ public static void maybeIgnoreMissingDirectory(FileSystem fs,
+ Path path,
+ FileNotFoundException e) throws FileNotFoundException {
+ final boolean b;
+ try {
+ b = !fs.hasPathCapability(path, DIRECTORY_LISTING_INCONSISTENT);
+ } catch (IOException ex) {
+ // something went wrong; rethrow the existing exception
+ e.addSuppressed(ex);
+ throw e;
+ }
+ if (b) {
+ throw e;
+ }
+ LOG.info("Ignoring missing directory {}", path);
+ LOG.debug("Directory missing", e);
+ }
}
diff --git a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/shell/Command.java b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/shell/Command.java
index 7858238ee71f..c4ce115def68 100644
--- a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/shell/Command.java
+++ b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/shell/Command.java
@@ -38,6 +38,7 @@ import org.apache.hadoop.util.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import static org.apache.hadoop.fs.FileUtil.maybeIgnoreMissingDirectory;
import static org.apache.hadoop.util.functional.RemoteIterators.cleanupRemoteIterator;
/**
@@ -448,12 +449,16 @@ abstract public class Command extends Configured {
protected void recursePath(PathData item) throws IOException {
try {
depth++;
- if (isSorted()) {
- // use the non-iterative method for listing because explicit sorting is
- // required. Iterators not guaranteed to return sorted elements
- processPaths(item, item.getDirectoryContents());
- } else {
- processPaths(item, item.getDirectoryContentsIterator());
+ try {
+ if (isSorted()) {
+ // use the non-iterative method for listing because explicit sorting is
+ // required. Iterators not guaranteed to return sorted elements
+ processPaths(item, item.getDirectoryContents());
+ } else {
+ processPaths(item, item.getDirectoryContentsIterator());
+ }
+ } catch (FileNotFoundException e) {
+ maybeIgnoreMissingDirectory(item.fs, item.path, e);
}
} finally {
depth--;
diff --git a/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/fs/FileContextURIBase.java b/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/fs/FileContextURIBase.java
index abb6d4f90159..fad1e3774ad2 100644
--- a/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/fs/FileContextURIBase.java
+++ b/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/fs/FileContextURIBase.java
@@ -26,6 +26,7 @@ import org.apache.hadoop.test.GenericTestUtils;
import org.junit.Assert;
import org.apache.hadoop.fs.permission.FsPermission;
+import org.apache.hadoop.test.LambdaTestUtils;
import org.apache.hadoop.util.Shell;
import org.junit.After;
import org.junit.Before;
@@ -129,7 +130,7 @@ public abstract class FileContextURIBase {
}
@Test
- public void testCreateExistingFile() throws IOException {
+ public void testCreateExistingFile() throws Exception {
String fileName = "testCreateExistingFile";
Path testPath = qualifiedPath(fileName, fc2);
@@ -140,15 +141,11 @@ public abstract class FileContextURIBase {
createFile(fc1, testPath);
// Create same file with fc1
- try {
- createFile(fc2, testPath);
- Assert.fail("Create existing file should throw an IOException.");
- } catch (IOException e) {
- // expected
- }
+ LambdaTestUtils.intercept(IOException.class, () ->
+ createFile(fc2, testPath));
// Ensure fc2 has the created file
- Assert.assertTrue(exists(fc2, testPath));
+ fc2.getFileStatus(testPath);
}
@Test
@@ -167,7 +164,7 @@ public abstract class FileContextURIBase {
Assert.assertTrue(isDir(fc2, testPath.getParent()));
Assert.assertEquals("testCreateFileInNonExistingDirectory",
testPath.getParent().getName());
- Assert.assertTrue(exists(fc2, testPath));
+ fc2.getFileStatus(testPath);
}
@@ -216,10 +213,11 @@ public abstract class FileContextURIBase {
// TestCase - Create multiple directories
String dirNames[] = {
"createTest/testDir", "createTest/test Dir",
- "deleteTest/test*Dir", "deleteTest/test#Dir",
- "deleteTest/test1234", "deleteTest/test_DIr",
- "deleteTest/1234Test", "deleteTest/test)Dir",
- "deleteTest/()&^%$#@!~_+}{><?", " ", "^ " };
+ "createTest/test*Dir", "createTest/test#Dir",
+ "createTest/test1234", "createTest/test_DIr",
+ "createTest/1234Test", "createTest/test)Dir",
+ "createTest/()&^%$#@!~_+}{><?",
+ "createTest/ ", "createTest/^ " };
for (String f : dirNames) {
if (!isTestableFileNameOnPlatform(f)) {
@@ -237,6 +235,10 @@ public abstract class FileContextURIBase {
Assert.assertTrue(exists(fc2, testPath));
Assert.assertTrue(isDir(fc2, testPath));
}
+ // delete the parent directory and verify that the dir no longer exists
+ final Path parent = qualifiedPath("createTest", fc2);
+ fc2.delete(parent, true);
+ Assert.assertFalse(exists(fc2, parent));
}
@@ -392,7 +394,9 @@ public abstract class FileContextURIBase {
"deleteTest/test*Dir", "deleteTest/test#Dir",
"deleteTest/test1234", "deleteTest/1234Test",
"deleteTest/test)Dir", "deleteTest/test_DIr",
- "deleteTest/()&^%$#@!~_+}{><?", " ", "^ " };
+ "deleteTest/()&^%$#@!~_+}{><?",
+ "deleteTest/ ",
+ "deleteTest/^ " };
for (String f : dirNames) {
if (!isTestableFileNameOnPlatform(f)) {
diff --git a/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/fs/contract/AbstractContractRootDirectoryTest.java b/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/fs/contract/AbstractContractRootDirectoryTest.java
index 924ebd0065d9..2988ebd215b8 100644
--- a/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/fs/contract/AbstractContractRootDirectoryTest.java
+++ b/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/fs/contract/AbstractContractRootDirectoryTest.java
@@ -29,7 +29,6 @@ import org.slf4j.LoggerFactory;
import java.io.IOException;
import java.util.Arrays;
import java.util.List;
-import java.util.concurrent.Callable;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors;
@@ -100,22 +99,18 @@ public abstract class AbstractContractRootDirectoryTest extends AbstractFSContra
final FileStatus[] originalChildren = listChildren(fs, root);
LambdaTestUtils.eventually(
OBJECTSTORE_RETRY_TIMEOUT,
- new Callable<Void>() {
- @Override
- public Void call() throws Exception {
- FileStatus[] deleted = deleteChildren(fs, root, true);
- FileStatus[] children = listChildren(fs, root);
- if (children.length > 0) {
- fail(String.format(
- "After %d attempts: listing after rm /* not empty"
- + "\n%s\n%s\n%s",
- iterations.incrementAndGet(),
- dumpStats("final", children),
+ () -> {
+ iterations.incrementAndGet();
+ FileStatus[] deleted = deleteChildren(fs, root, true);
+ FileStatus[] children = listChildren(fs, root);
+ Assertions.assertThat(children)
+ .describedAs("After %d attempts: listing after rm /* not empty"
+ + "\ndeleted: %s\n: original %s",
+ iterations.get(),
dumpStats("deleted", deleted),
- dumpStats("original", originalChildren)));
- }
- return null;
- }
+ dumpStats("original", originalChildren))
+ .isEmpty();
+ return null;
},
new LambdaTestUtils.ProportionalRetryInterval(50, 1000));
// then try to delete the empty one
diff --git a/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/fs/contract/ContractTestUtils.java b/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/fs/contract/ContractTestUtils.java
index b61abddd4342..13c86af08c4f 100644
--- a/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/fs/contract/ContractTestUtils.java
+++ b/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/fs/contract/ContractTestUtils.java
@@ -24,6 +24,7 @@ import org.apache.hadoop.fs.FileContext;
import org.apache.hadoop.fs.FileRange;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.FileUtil;
import org.apache.hadoop.fs.LocatedFileStatus;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.PathCapabilities;
@@ -34,6 +35,7 @@ import org.apache.hadoop.io.IOUtils;
import org.apache.hadoop.util.functional.RemoteIterators;
import org.apache.hadoop.util.functional.FutureIO;
+import org.assertj.core.api.Assertions;
import org.junit.Assert;
import org.junit.AssumptionViolatedException;
import org.slf4j.Logger;
@@ -61,6 +63,7 @@ import java.util.concurrent.TimeoutException;
import static org.apache.hadoop.fs.CommonConfigurationKeysPublic.IO_FILE_BUFFER_SIZE_DEFAULT;
import static org.apache.hadoop.fs.CommonConfigurationKeysPublic.IO_FILE_BUFFER_SIZE_KEY;
+import static org.apache.hadoop.util.functional.RemoteIterators.foreach;
/**
* Utilities used across test cases.
@@ -1506,19 +1509,39 @@ public class ContractTestUtils extends Assert {
*/
public static TreeScanResults treeWalk(FileSystem fs, Path path)
throws IOException {
- TreeScanResults dirsAndFiles = new TreeScanResults();
+ return treeWalk(fs, fs.getFileStatus(path), new TreeScanResults());
+ }
- FileStatus[] statuses = fs.listStatus(path);
- for (FileStatus status : statuses) {
- LOG.info("{}{}", status.getPath(), status.isDirectory() ? "*" : "");
- }
- for (FileStatus status : statuses) {
- dirsAndFiles.add(status);
- if (status.isDirectory()) {
- dirsAndFiles.add(treeWalk(fs, status.getPath()));
- }
+ /**
+ * Recursively list all entries, with a depth first traversal of the
+ * directory tree.
+ * @param dir status of the dir to scan
+ * @return the scan results
+ * @throws IOException IO problems
+ * @throws FileNotFoundException if the dir is not found and this FS does not
+ * have the listing inconsistency path capability/flag.
+ */
+ private static TreeScanResults treeWalk(FileSystem fs,
+ FileStatus dir,
+ TreeScanResults results) throws IOException {
+
+ Path path = dir.getPath();
+
+ try {
+ foreach(fs.listStatusIterator(path), (status) -> {
+ LOG.info("{}{}", status.getPath(), status.isDirectory() ? "*" : "");
+ if (status.isDirectory()) {
+ treeWalk(fs, status, results);
+ } else {
+ results.add(status);
+ }
+ });
+ // and ourselves
+ results.add(dir);
+ } catch (FileNotFoundException fnfe) {
+ FileUtil.maybeIgnoreMissingDirectory(fs, path, fnfe);
}
- return dirsAndFiles;
+ return results;
}
/**
@@ -1916,17 +1939,16 @@ public class ContractTestUtils extends Assert {
public void assertFieldsEquivalent(String fieldname,
TreeScanResults that,
List<Path> ours, List<Path> theirs) {
- String ourList = pathsToString(ours);
- String theirList = pathsToString(theirs);
- assertFalse("Duplicate " + fieldname + " in " + this
- +": " + ourList,
- containsDuplicates(ours));
- assertFalse("Duplicate " + fieldname + " in other " + that
- + ": " + theirList,
- containsDuplicates(theirs));
- assertTrue(fieldname + " mismatch: between " + ourList
- + " and " + theirList,
- collectionsEquivalent(ours, theirs));
+ Assertions.assertThat(ours).
+ describedAs("list of %s", fieldname)
+ .doesNotHaveDuplicates();
+ Assertions.assertThat(theirs).
+ describedAs("list of %s in %s", fieldname, that)
+ .doesNotHaveDuplicates();
+ Assertions.assertThat(ours)
+ .describedAs("Elements of %s", fieldname)
+ .containsExactlyInAnyOrderElementsOf(theirs);
+
}
public List<Path> getFiles() {
diff --git a/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapred/LocatedFileStatusFetcher.java b/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapred/LocatedFileStatusFetcher.java
index 5810698607c4..fcaa02a65586 100644
--- a/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapred/LocatedFileStatusFetcher.java
+++ b/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapred/LocatedFileStatusFetcher.java
@@ -18,6 +18,7 @@
package org.apache.hadoop.mapred;
+import java.io.FileNotFoundException;
import java.io.IOException;
import java.util.LinkedList;
import java.util.List;
@@ -56,6 +57,7 @@ import org.slf4j.LoggerFactory;
import org.apache.hadoop.util.concurrent.HadoopExecutors;
+import static org.apache.hadoop.fs.FileUtil.maybeIgnoreMissingDirectory;
import static org.apache.hadoop.fs.statistics.IOStatisticsSupport.retrieveIOStatistics;
import static org.apache.hadoop.fs.statistics.IOStatisticsSupport.snapshotIOStatistics;
@@ -304,23 +306,27 @@ public class LocatedFileStatusFetcher implements IOStatisticsSource {
Result result = new Result();
result.fs = fs;
LOG.debug("ProcessInputDirCallable {}", fileStatus);
- if (fileStatus.isDirectory()) {
- RemoteIterator<LocatedFileStatus> iter = fs
- .listLocatedStatus(fileStatus.getPath());
- while (iter.hasNext()) {
- LocatedFileStatus stat = iter.next();
- if (inputFilter.accept(stat.getPath())) {
- if (recursive && stat.isDirectory()) {
- result.dirsNeedingRecursiveCalls.add(stat);
- } else {
- result.locatedFileStatuses.add(stat);
+ try {
+ if (fileStatus.isDirectory()) {
+ RemoteIterator<LocatedFileStatus> iter = fs
+ .listLocatedStatus(fileStatus.getPath());
+ while (iter.hasNext()) {
+ LocatedFileStatus stat = iter.next();
+ if (inputFilter.accept(stat.getPath())) {
+ if (recursive && stat.isDirectory()) {
+ result.dirsNeedingRecursiveCalls.add(stat);
+ } else {
+ result.locatedFileStatuses.add(stat);
+ }
}
}
+ // aggregate any stats
+ result.stats = retrieveIOStatistics(iter);
+ } else {
+ result.locatedFileStatuses.add(fileStatus);
}
- // aggregate any stats
- result.stats = retrieveIOStatistics(iter);
- } else {
- result.locatedFileStatuses.add(fileStatus);
+ } catch (FileNotFoundException e) {
+ maybeIgnoreMissingDirectory(fs, fileStatus.getPath(), e);
}
return result;
}
diff --git a/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/lib/input/FileInputFormat.java b/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/lib/input/FileInputFormat.java
index 2d88d13a1546..a7c837628857 100644
--- a/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/lib/input/FileInputFormat.java
+++ b/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/lib/input/FileInputFormat.java
@@ -18,6 +18,7 @@
package org.apache.hadoop.mapreduce.lib.input;
+import java.io.FileNotFoundException;
import java.io.IOException;
import java.io.InterruptedIOException;
import java.util.ArrayList;
@@ -48,8 +49,9 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.hadoop.thirdparty.com.google.common.collect.Lists;
+import static org.apache.hadoop.fs.FileUtil.maybeIgnoreMissingDirectory;
-/**
+/**
* A base class for file-based {@link InputFormat}s.
*
* <p><code>FileInputFormat</code> is the base class for all file-based
@@ -357,16 +359,26 @@ public abstract class FileInputFormat<K, V> extends InputFormat<K, V> {
protected void addInputPathRecursively(List<FileStatus> result,
FileSystem fs, Path path, PathFilter inputFilter)
throws IOException {
- RemoteIterator<LocatedFileStatus> iter = fs.listLocatedStatus(path);
- while (iter.hasNext()) {
- LocatedFileStatus stat = iter.next();
- if (inputFilter.accept(stat.getPath())) {
- if (stat.isDirectory()) {
- addInputPathRecursively(result, fs, stat.getPath(), inputFilter);
- } else {
- result.add(stat);
+ // FNFE exceptions are caught whether raised in the list call,
+ // or in the hasNext() or next() calls, where async reporting
+ // may take place.
+ try {
+ RemoteIterator<LocatedFileStatus> iter = fs.listLocatedStatus(path);
+ while (iter.hasNext()) {
+ LocatedFileStatus stat = iter.next();
+ if (inputFilter.accept(stat.getPath())) {
+ if (stat.isDirectory()) {
+ addInputPathRecursively(result, fs, stat.getPath(), inputFilter);
+ } else {
+ result.add(stat);
+ }
}
}
+ } catch (FileNotFoundException e) {
+ // unless the store is capabile of list inconsistencies, rethrow.
+ // because this is recursive, the caller method may also end up catching
+ // and rethrowing, which is slighly inefficient but harmless.
+ maybeIgnoreMissingDirectory(fs, path, e);
}
}
diff --git a/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/Constants.java b/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/Constants.java
index 140ea53ceb6f..b0369ac735f5 100644
--- a/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/Constants.java
+++ b/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/Constants.java
@@ -1462,9 +1462,18 @@ public final class Constants {
/**
* Stream supports multipart uploads to the given path.
*/
- public static final String STORE_CAPABILITY_DIRECTORY_MARKER_MULTIPART_UPLOAD_ENABLED =
+ public static final String STORE_CAPABILITY_MULTIPART_UPLOAD_ENABLED =
"fs.s3a.capability.multipart.uploads.enabled";
+ /**
+ * Stream supports multipart uploads to the given path.
+ * This name is wrong, but it has shipped so must be
+ * retained.
+ */
+ @Deprecated
+ public static final String STORE_CAPABILITY_DIRECTORY_MARKER_MULTIPART_UPLOAD_ENABLED
+ = STORE_CAPABILITY_MULTIPART_UPLOAD_ENABLED;
+
/**
* Prefetch max blocks count config.
* Value = {@value}
@@ -1509,4 +1518,11 @@ public final class Constants {
* Value: {@value}.
*/
public static final boolean OPTIMIZED_COPY_FROM_LOCAL_DEFAULT = true;
+
+ /**
+ * Is this a v2 SDK build? value {@value}.
+ */
+ public static final String STORE_CAPABILITY_AWS_V2 =
+ "fs.s3a.capability.aws.v2";
+
}
diff --git a/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/Listing.java b/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/Listing.java
index 490deaaab04d..e0868a2e1308 100644
--- a/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/Listing.java
+++ b/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/Listing.java
@@ -259,7 +259,7 @@ public class Listing extends AbstractStoreOperation {
}
S3ListRequest request = createListObjectsRequest(key, "/", span);
- LOG.debug("listStatus: doing listObjects for directory {}", key);
+ LOG.debug("listStatus: doing listObjects for directory \"{}\"", key);
// return the results obtained from s3.
return createFileStatusListingIterator(
diff --git a/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/S3AFileSystem.java b/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/S3AFileSystem.java
index 3e995fb1ddb4..f02c8495914b 100644
--- a/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/S3AFileSystem.java
+++ b/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/S3AFileSystem.java
@@ -220,6 +220,7 @@ import static org.apache.hadoop.fs.CommonConfigurationKeys.IOSTATISTICS_LOGGING_
import static org.apache.hadoop.fs.CommonConfigurationKeys.IOSTATISTICS_LOGGING_LEVEL_DEFAULT;
import static org.apache.hadoop.fs.CommonConfigurationKeysPublic.IO_FILE_BUFFER_SIZE_DEFAULT;
import static org.apache.hadoop.fs.CommonConfigurationKeysPublic.IO_FILE_BUFFER_SIZE_KEY;
+import static org.apache.hadoop.fs.CommonPathCapabilities.DIRECTORY_LISTING_INCONSISTENT;
import static org.apache.hadoop.fs.impl.PathCapabilitiesSupport.validatePathCapabilityArgs;
import static org.apache.hadoop.fs.s3a.Constants.*;
import static org.apache.hadoop.fs.s3a.Invoker.*;
@@ -250,6 +251,8 @@ import static org.apache.hadoop.fs.s3a.impl.InternalConstants.SC_404_NOT_FOUND;
import static org.apache.hadoop.fs.s3a.impl.InternalConstants.UPLOAD_PART_COUNT_LIMIT;
import static org.apache.hadoop.fs.s3a.impl.NetworkBinding.fixBucketRegion;
import static org.apache.hadoop.fs.s3a.impl.NetworkBinding.logDnsLookup;
+import static org.apache.hadoop.fs.s3a.impl.S3ExpressStorage.STORE_CAPABILITY_S3_EXPRESS_STORAGE;
+import static org.apache.hadoop.fs.s3a.impl.S3ExpressStorage.isS3ExpressStore;
import static org.apache.hadoop.fs.s3a.s3guard.S3Guard.checkNoS3Guard;
import static org.apache.hadoop.fs.statistics.IOStatisticsLogging.logIOStatisticsAtLevel;
import static org.apache.hadoop.fs.statistics.StoreStatisticNames.OBJECT_CONTINUE_LIST_REQUEST;
@@ -289,11 +292,16 @@ public class S3AFileSystem extends FileSystem implements StreamCapabilities,
public static final int DEFAULT_BLOCKSIZE = 32 * 1024 * 1024;
private URI uri;
+
private Path workingDir;
+
private String username;
+
private S3Client s3Client;
+
/** Async client is used for transfer manager and s3 select. */
private S3AsyncClient s3AsyncClient;
+
// initial callback policy is fail-once; it's there just to assist
// some mock tests and other codepaths trying to call the low level
// APIs on an uninitialized filesystem.
@@ -470,6 +478,21 @@ public class S3AFileSystem extends FileSystem implements StreamCapabilities,
*/
private boolean optimizedCopyFromLocal;
+ /**
+ * Is this an S3 Express store?
+ */
+ private boolean s3ExpressStore;
+
+ /**
+ * Store endpoint from configuration info or access point ARN.
+ */
+ private String endpoint;
+
+ /**
+ * Region from configuration info or access point ARN.
+ */
+ private String configuredRegion;
+
/** Add any deprecated keys. */
@SuppressWarnings("deprecation")
private static void addDeprecatedKeys() {
@@ -581,9 +604,23 @@ public class S3AFileSystem extends FileSystem implements StreamCapabilities,
//check but do not store the block size
longBytesOption(conf, FS_S3A_BLOCK_SIZE, DEFAULT_BLOCKSIZE, 1);
enableMultiObjectsDelete = conf.getBoolean(ENABLE_MULTI_DELETE, true);
- // should the delete also purge uploads.
+
+ // determine and cache the endpoints
+ endpoint = accessPoint == null
+ ? conf.getTrimmed(ENDPOINT, DEFAULT_ENDPOINT)
+ : accessPoint.getEndpoint();
+
+ configuredRegion = accessPoint == null
+ ? conf.getTrimmed(AWS_REGION)
+ : accessPoint.getRegion();
+
+ // is this an S3Express store?
+ s3ExpressStore = isS3ExpressStore(bucket, endpoint);
+
+ // should the delete also purge uploads?
+ // happens if explicitly enabled, or if the store is S3Express storage.
dirOperationsPurgeUploads = conf.getBoolean(DIRECTORY_OPERATIONS_PURGE_UPLOADS,
- DIRECTORY_OPERATIONS_PURGE_UPLOADS_DEFAULT);
+ s3ExpressStore);
this.isMultipartUploadEnabled = conf.getBoolean(MULTIPART_UPLOADS_ENABLED,
DEFAULT_MULTIPART_UPLOAD_ENABLED);
@@ -996,14 +1033,6 @@ public class S3AFileSystem extends FileSystem implements StreamCapabilities,
S3_CLIENT_FACTORY_IMPL, DEFAULT_S3_CLIENT_FACTORY_IMPL,
S3ClientFactory.class);
- String endpoint = accessPoint == null
- ? conf.getTrimmed(ENDPOINT, DEFAULT_ENDPOINT)
- : accessPoint.getEndpoint();
-
- String configuredRegion = accessPoint == null
- ? conf.getTrimmed(AWS_REGION)
- : accessPoint.getRegion();
-
S3ClientFactory.S3ClientCreationParameters parameters =
new S3ClientFactory.S3ClientCreationParameters()
.withCredentialSet(credentials)
@@ -1445,7 +1474,19 @@ public class S3AFileSystem extends FileSystem implements StreamCapabilities,
public boolean isMultipartCopyEnabled() {
return S3AFileSystem.this.isMultipartUploadEnabled;
}
- }
+
+ @Override
+ public long abortMultipartUploads(final Path path) throws IOException {
+ final String prefix = pathToKey(path);
+ try (AuditSpan span = createSpan("object_multipart_bulk_abort", prefix, null)) {
+ return S3AFileSystem.this.abortMultipartUploadsUnderPrefix(
+ createStoreContext(),
+ span,
+ prefix);
+ }
+ }
+
+ } // end S3AInternals
/**
* Get the input policy for this FS instance.
@@ -2483,19 +2524,34 @@ public class S3AFileSystem extends FileSystem implements StreamCapabilities,
@Retries.RetryTranslated
public long abortMultipartUploadsUnderPrefix(String prefix)
throws IOException {
- getAuditSpan().activate();
- // this deactivates the audit span somehow
- final RemoteIterator<MultipartUpload> uploads =
- S3AFileSystem.this.listUploadsUnderPrefix(storeContext, prefix);
- // so reactivate it.
- getAuditSpan().activate();
- return foreach(uploads, upload ->
- invoker.retry("Aborting multipart commit", upload.key(), true, () ->
- S3AFileSystem.this.abortMultipartUpload(upload)));
+ return S3AFileSystem.this.abortMultipartUploadsUnderPrefix(storeContext, auditSpan, prefix);
}
} // end OperationCallbacksImpl
+ /**
+ * Abort multipart uploads under a prefix.
+ * @param storeContext store context
+ * @param span span for the operations
+ * @param prefix prefix for uploads to abort
+ * @return a count of aborts
+ * @throws IOException trouble; FileNotFoundExceptions are swallowed.
+ */
+ private long abortMultipartUploadsUnderPrefix(StoreContext storeContext,
+ AuditSpan span,
+ String prefix) throws IOException {
+
+ span.activate();
+ // this deactivates the audit span somehow
+ final RemoteIterator<MultipartUpload> uploads =
+ listUploadsUnderPrefix(storeContext, prefix);
+ // so reactivate it.
+ span.activate();
+ return foreach(uploads, upload ->
+ invoker.retry("Aborting multipart commit", upload.key(), true, () ->
+ abortMultipartUpload(upload)));
+ }
+
/**
* Callbacks from {@link Listing}.
* Auditing: the listing object is long-lived; the audit span
@@ -3327,7 +3383,7 @@ public class S3AFileSystem extends FileSystem implements StreamCapabilities,
LOG.debug("Initiating delete operation for {} objects",
keysToDelete.size());
for (ObjectIdentifier objectIdentifier : keysToDelete) {
- LOG.debug(" {} {}", objectIdentifier.key(),
+ LOG.debug(" \"{}\" {}", objectIdentifier.key(),
objectIdentifier.versionId() != null ? objectIdentifier.versionId() : "");
}
}
@@ -5400,10 +5456,24 @@ public class S3AFileSystem extends FileSystem implements StreamCapabilities,
case STORE_CAPABILITY_DIRECTORY_MARKER_AWARE:
return true;
+ // multi object delete flag
+ case ENABLE_MULTI_DELETE:
+ return enableMultiObjectsDelete;
+
// Do directory operations purge uploads.
case DIRECTORY_OPERATIONS_PURGE_UPLOADS:
return dirOperationsPurgeUploads;
+ // this is a v2 sdk release.
+ case STORE_CAPABILITY_AWS_V2:
+ return true;
+
+ // is this store S3 Express?
+ // if so, note that directory listings may be inconsistent
+ case STORE_CAPABILITY_S3_EXPRESS_STORAGE:
+ case DIRECTORY_LISTING_INCONSISTENT:
+ return s3ExpressStore;
+
// etags are avaialable in listings, but they
// are not consistent across renames.
// therefore, only availability is declared
diff --git a/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/S3AInternals.java b/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/S3AInternals.java
index 18d6c1af586f..b4116068565c 100644
--- a/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/S3AInternals.java
+++ b/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/S3AInternals.java
@@ -121,4 +121,14 @@ public interface S3AInternals {
* @return true if the transfer manager is used to copy files.
*/
boolean isMultipartCopyEnabled();
+
+ /**
+ * Abort multipart uploads under a path.
+ * @param path path to abort uploads under.
+ * @return a count of aborts
+ * @throws IOException trouble; FileNotFoundExceptions are swallowed.
+ */
+ @AuditEntryPoint
+ @Retries.RetryTranslated
+ long abortMultipartUploads(Path path) throws IOException;
}
diff --git a/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/S3AUtils.java b/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/S3AUtils.java
index 94fa2d35fc6f..703f64bfb6c2 100644
--- a/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/S3AUtils.java
+++ b/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/S3AUtils.java
@@ -270,6 +270,11 @@ public final class S3AUtils {
}
break;
+ // Caused by duplicate create bucket call.
+ case SC_409_CONFLICT:
+ ioe = new AWSBadRequestException(message, ase);
+ break;
+
// this also surfaces sometimes and is considered to
// be ~ a not found exception.
case SC_410_GONE:
@@ -280,12 +285,18 @@ public final class S3AUtils {
// errors which stores can return from requests which
// the store does not support.
case SC_405_METHOD_NOT_ALLOWED:
- case SC_412_PRECONDITION_FAILED:
case SC_415_UNSUPPORTED_MEDIA_TYPE:
case SC_501_NOT_IMPLEMENTED:
ioe = new AWSUnsupportedFeatureException(message, s3Exception);
break;
+ // precondition failure: the object is there, but the precondition
+ // (e.g. etag) didn't match. Assume remote file change during
+ // rename or status passed in to openfile had an etag which didn't match.
+ case SC_412_PRECONDITION_FAILED:
+ ioe = new RemoteFileChangedException(path, message, "", ase);
+ break;
+
// out of range. This may happen if an object is overwritten with
// a shorter one while it is being read.
case SC_416_RANGE_NOT_SATISFIABLE:
@@ -860,7 +871,7 @@ public final class S3AUtils {
*/
public static String stringify(S3Object s3Object) {
StringBuilder builder = new StringBuilder(s3Object.key().length() + 100);
- builder.append(s3Object.key()).append(' ');
+ builder.append("\"").append(s3Object.key()).append("\" ");
builder.append("size=").append(s3Object.size());
return builder.toString();
}
@@ -1633,4 +1644,5 @@ public final class S3AUtils {
public static String formatRange(long rangeStart, long rangeEnd) {
return String.format("bytes=%d-%d", rangeStart, rangeEnd);
}
+
}
diff --git a/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/audit/AWSRequestAnalyzer.java b/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/audit/AWSRequestAnalyzer.java
index 8a24a4e14db1..3cb8d9753244 100644
--- a/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/audit/AWSRequestAnalyzer.java
+++ b/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/audit/AWSRequestAnalyzer.java
@@ -24,6 +24,7 @@ import software.amazon.awssdk.core.SdkRequest;
import software.amazon.awssdk.services.s3.model.AbortMultipartUploadRequest;
import software.amazon.awssdk.services.s3.model.CompleteMultipartUploadRequest;
import software.amazon.awssdk.services.s3.model.CreateMultipartUploadRequest;
+import software.amazon.awssdk.services.s3.model.CreateSessionRequest;
import software.amazon.awssdk.services.s3.model.DeleteObjectRequest;
import software.amazon.awssdk.services.s3.model.DeleteObjectsRequest;
import software.amazon.awssdk.services.s3.model.GetBucketLocationRequest;
@@ -196,7 +197,8 @@ public class AWSRequestAnalyzer {
isRequestNotAlwaysInSpan(final Object request) {
return request instanceof UploadPartCopyRequest
|| request instanceof CompleteMultipartUploadRequest
- || request instanceof GetBucketLocationRequest;
+ || request instanceof GetBucketLocationRequest
+ || request instanceof CreateSessionRequest;
}
/**
diff --git a/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/auth/CustomSdkSigner.java b/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/auth/CustomSdkSigner.java
new file mode 100644
index 000000000000..b37860216507
--- /dev/null
+++ b/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/auth/CustomSdkSigner.java
@@ -0,0 +1,178 @@
+/*
+ * 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.auth;
+
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import software.amazon.awssdk.arns.Arn;
+import software.amazon.awssdk.auth.signer.Aws4Signer;
+import software.amazon.awssdk.auth.signer.AwsS3V4Signer;
+import software.amazon.awssdk.auth.signer.internal.AbstractAwsS3V4Signer;
+import software.amazon.awssdk.core.interceptor.ExecutionAttributes;
+import software.amazon.awssdk.core.signer.Signer;
+import software.amazon.awssdk.http.SdkHttpFullRequest;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.s3a.auth.delegation.DelegationTokenProvider;
+import org.apache.hadoop.security.UserGroupInformation;
+
+/**
+ * This class is for testing the SDK's signing: it
+ * can be declared as the signer class in the configuration
+ * and then the full test suite run with it.
+ * Derived from the inner class of {@code ITestCustomSigner}.
+ * <pre>
+ * fs.s3a.custom.signers=CustomSdkSigner:org.apache.hadoop.fs.s3a.auth.CustomSdkSigner
+ *
+ * fs.s3a.s3.signing-algorithm=CustomSdkSigner
+ * </pre>
+ */
+public class CustomSdkSigner extends AbstractAwsS3V4Signer implements Signer {
+
+ private static final Logger LOG = LoggerFactory
+ .getLogger(CustomSdkSigner.class);
+
+ private static final AtomicInteger INSTANTIATION_COUNT =
+ new AtomicInteger(0);
+ private static final AtomicInteger INVOCATION_COUNT =
+ new AtomicInteger(0);
+
+ /**
+ * Signer for all S3 requests.
+ */
+ private final AwsS3V4Signer s3Signer = AwsS3V4Signer.create();
+
+ /**
+ * Signer for other services.
+ */
+ private final Aws4Signer aws4Signer = Aws4Signer.create();
+
+
+ public CustomSdkSigner() {
+ int c = INSTANTIATION_COUNT.incrementAndGet();
+ LOG.info("Creating Signer #{}", c);
+ }
+
+
+ /**
+ * Method to sign the incoming request with credentials.
+ * <p>
+ * NOTE: In case of Client-side encryption, we do a "Generate Key" POST
+ * request to AWSKMS service rather than S3, this was causing the test to
+ * break. When this request happens, we have the endpoint in form of
+ * "kms.[REGION].amazonaws.com", and bucket-name becomes "kms". We can't
+ * use AWSS3V4Signer for AWSKMS service as it contains a header
+ * "x-amz-content-sha256:UNSIGNED-PAYLOAD", which returns a 400 bad
+ * request because the signature calculated by the service doesn't match
+ * what we sent.
+ * @param request the request to sign.
+ * @param executionAttributes request executionAttributes which contain the credentials.
+ */
+ @Override
+ public SdkHttpFullRequest sign(SdkHttpFullRequest request,
+ ExecutionAttributes executionAttributes) {
+ int c = INVOCATION_COUNT.incrementAndGet();
+
+ String host = request.host();
+ LOG.debug("Signing request #{} against {}: class {}",
+ c, host, request.getClass());
+ String bucketName = parseBucketFromHost(host);
+ if (bucketName.equals("kms")) {
+ return aws4Signer.sign(request, executionAttributes);
+ } else {
+ return s3Signer.sign(request, executionAttributes);
+ }
+ }
+
+ /**
+ * Parse the bucket name from the host.
+ * @param host hostname
+ * @return the parsed bucket name; if "kms" is KMS signing.
+ */
+ static String parseBucketFromHost(String host) {
+ String[] hostBits = host.split("\\.");
+ String bucketName = hostBits[0];
+ String service = hostBits[1];
+
+ if (bucketName.equals("kms")) {
+ return bucketName;
+ }
+
+ if (service.contains("s3-accesspoint") || service.contains("s3-outposts")
+ || service.contains("s3-object-lambda")) {
+ // If AccessPoint then bucketName is of format `accessPoint-accountId`;
+ String[] accessPointBits = bucketName.split("-");
+ String accountId = accessPointBits[accessPointBits.length - 1];
+ // Extract the access point name from bucket name. eg: if bucket name is
+ // test-custom-signer-<accountId>, get the access point name test-custom-signer by removing
+ // -<accountId> from the bucket name.
+ String accessPointName =
+ bucketName.substring(0, bucketName.length() - (accountId.length() + 1));
+ Arn arn = Arn.builder()
+ .accountId(accountId)
+ .partition("aws")
+ .region(hostBits[2])
+ .resource("accesspoint" + "/" + accessPointName)
+ .service("s3").build();
+
+ bucketName = arn.toString();
+ }
+
+ return bucketName;
+ }
+
+ public static int getInstantiationCount() {
+ return INSTANTIATION_COUNT.get();
+ }
+
+ public static int getInvocationCount() {
+ return INVOCATION_COUNT.get();
+ }
+
+ public static String description() {
+ return "CustomSigner{"
+ + "invocations=" + INVOCATION_COUNT.get()
+ + ", instantiations=" + INSTANTIATION_COUNT.get()
+ + "}";
+ }
+
+ public static class Initializer implements AwsSignerInitializer {
+
+ @Override
+ public void registerStore(
+ final String bucketName,
+ final Configuration storeConf,
+ final DelegationTokenProvider dtProvider,
+ final UserGroupInformation storeUgi) {
+
+ LOG.debug("Registering store for bucket {}", bucketName);
+ }
+
+ @Override
+ public void unregisterStore(final String bucketName,
+ final Configuration storeConf,
+ final DelegationTokenProvider dtProvider,
+ final UserGroupInformation storeUgi) {
+ LOG.debug("Unregistering store for bucket {}", bucketName);
+ }
+ }
+
+}
diff --git a/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/impl/DeleteOperation.java b/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/impl/DeleteOperation.java
index 7cc94cf89203..b79cf7aef5d7 100644
--- a/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/impl/DeleteOperation.java
+++ b/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/impl/DeleteOperation.java
@@ -263,7 +263,7 @@ public class DeleteOperation extends ExecutingStoreOperation<Boolean> {
final CompletableFuture<Long> abortUploads;
if (dirOperationsPurgeUploads) {
final StoreContext sc = getStoreContext();
- final String key = sc.pathToKey(path) + "/";
+ final String key = dirKey;
LOG.debug("All uploads under {} will be deleted", key);
abortUploads = submit(sc.getExecutor(), sc.getActiveAuditSpan(), () ->
callbacks.abortMultipartUploadsUnderPrefix(key));
@@ -439,14 +439,16 @@ public class DeleteOperation extends ExecutingStoreOperation<Boolean> {
.filter(e -> e.isDirMarker)
.map(e -> e.objectIdentifier)
.collect(Collectors.toList());
- LOG.debug("Deleting of {} directory markers", dirs.size());
- // This is invoked with deleteFakeDir.
- Invoker.once("Remove S3 Dir Markers",
- status.getPath().toString(),
- () -> callbacks.removeKeys(
- dirs,
- true
- ));
+ if (!dirs.isEmpty()) {
+ LOG.debug("Deleting {} directory markers", dirs.size());
+ // This is invoked with deleteFakeDir.
+ Invoker.once("Remove S3 Dir Markers",
+ status.getPath().toString(),
+ () -> callbacks.removeKeys(
+ dirs,
+ true
+ ));
+ }
}
}
}
diff --git a/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/impl/InternalConstants.java b/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/impl/InternalConstants.java
index 8b29313faf2d..4600f392e415 100644
--- a/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/impl/InternalConstants.java
+++ b/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/impl/InternalConstants.java
@@ -21,6 +21,7 @@ package org.apache.hadoop.fs.s3a.impl;
import java.util.Arrays;
import java.util.Collections;
import java.util.HashSet;
+import java.util.List;
import java.util.Set;
import java.util.stream.Collectors;
import java.util.stream.Stream;
@@ -30,7 +31,24 @@ import org.apache.hadoop.thirdparty.com.google.common.annotations.VisibleForTest
import org.apache.hadoop.classification.InterfaceStability;
import org.apache.hadoop.fs.s3a.Constants;
+import static org.apache.hadoop.fs.CommonPathCapabilities.DIRECTORY_LISTING_INCONSISTENT;
+import static org.apache.hadoop.fs.CommonPathCapabilities.ETAGS_AVAILABLE;
+import static org.apache.hadoop.fs.CommonPathCapabilities.FS_CHECKSUMS;
+import static org.apache.hadoop.fs.CommonPathCapabilities.FS_MULTIPART_UPLOADER;
import static org.apache.hadoop.fs.Options.OpenFileOptions.FS_OPTION_OPENFILE_STANDARD_OPTIONS;
+import static org.apache.hadoop.fs.s3a.Constants.DIRECTORY_OPERATIONS_PURGE_UPLOADS;
+import static org.apache.hadoop.fs.s3a.Constants.ENABLE_MULTI_DELETE;
+import static org.apache.hadoop.fs.s3a.Constants.FS_S3A_CREATE_PERFORMANCE;
+import static org.apache.hadoop.fs.s3a.Constants.FS_S3A_CREATE_PERFORMANCE_ENABLED;
+import static org.apache.hadoop.fs.s3a.Constants.STORE_CAPABILITY_AWS_V2;
+import static org.apache.hadoop.fs.s3a.impl.S3ExpressStorage.STORE_CAPABILITY_S3_EXPRESS_STORAGE;
+import static org.apache.hadoop.fs.s3a.Constants.STORE_CAPABILITY_DIRECTORY_MARKER_ACTION_DELETE;
+import static org.apache.hadoop.fs.s3a.Constants.STORE_CAPABILITY_DIRECTORY_MARKER_ACTION_KEEP;
+import static org.apache.hadoop.fs.s3a.Constants.STORE_CAPABILITY_DIRECTORY_MARKER_POLICY_AUTHORITATIVE;
+import static org.apache.hadoop.fs.s3a.Constants.STORE_CAPABILITY_DIRECTORY_MARKER_POLICY_DELETE;
+import static org.apache.hadoop.fs.s3a.Constants.STORE_CAPABILITY_DIRECTORY_MARKER_POLICY_KEEP;
+import static org.apache.hadoop.fs.s3a.Constants.STORE_CAPABILITY_MULTIPART_UPLOAD_ENABLED;
+import static org.apache.hadoop.fs.s3a.commit.CommitConstants.STORE_CAPABILITY_MAGIC_COMMITTER;
/**
* Internal constants private only to the S3A codebase.
@@ -142,6 +160,9 @@ public final class InternalConstants {
/** 405 status code: Method Not Allowed. */
public static final int SC_405_METHOD_NOT_ALLOWED = 405;
+ /** 409 status code: Conflict. Example: creating a bucket twice. */
+ public static final int SC_409_CONFLICT = 409;
+
/** 410 status code: Gone. */
public static final int SC_410_GONE = 410;
@@ -239,6 +260,30 @@ public final class InternalConstants {
*/
public static final Set<String> CREATE_FILE_KEYS =
Collections.unmodifiableSet(
- new HashSet<>(Arrays.asList(Constants.FS_S3A_CREATE_PERFORMANCE)));
+ new HashSet<>(Arrays.asList(FS_S3A_CREATE_PERFORMANCE)));
+ /**
+ * Dynamic Path capabilities to be evaluated
+ * in the BucketInfo tool.
+ */
+ public static final List<String> S3A_DYNAMIC_CAPABILITIES =
+ Collections.unmodifiableList(Arrays.asList(
+ ETAGS_AVAILABLE,
+ FS_CHECKSUMS,
+ FS_MULTIPART_UPLOADER,
+ DIRECTORY_LISTING_INCONSISTENT,
+
+ // s3 specific
+ STORE_CAPABILITY_AWS_V2,
+ STORE_CAPABILITY_DIRECTORY_MARKER_POLICY_KEEP,
+ STORE_CAPABILITY_DIRECTORY_MARKER_POLICY_DELETE,
+ STORE_CAPABILITY_DIRECTORY_MARKER_POLICY_AUTHORITATIVE,
+ STORE_CAPABILITY_DIRECTORY_MARKER_ACTION_KEEP,
+ STORE_CAPABILITY_DIRECTORY_MARKER_ACTION_DELETE,
+ STORE_CAPABILITY_MAGIC_COMMITTER,
+ STORE_CAPABILITY_MULTIPART_UPLOAD_ENABLED,
+ STORE_CAPABILITY_S3_EXPRESS_STORAGE,
+ FS_S3A_CREATE_PERFORMANCE_ENABLED,
+ DIRECTORY_OPERATIONS_PURGE_UPLOADS,
+ ENABLE_MULTI_DELETE));
}
diff --git a/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/impl/NetworkBinding.java b/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/impl/NetworkBinding.java
index 34b4049b0612..910723f909f5 100644
--- a/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/impl/NetworkBinding.java
+++ b/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/impl/NetworkBinding.java
@@ -97,6 +97,17 @@ public final class NetworkBinding {
}
}
+ /**
+ * Is this an AWS endpoint? looks at end of FQDN.
+ * @param endpoint endpoint
+ * @return true if the endpoint matches the requirements for an aws endpoint.
+ */
+ public static boolean isAwsEndpoint(final String endpoint) {
+ return (endpoint.isEmpty()
+ || endpoint.endsWith(".amazonaws.com")
+ || endpoint.endsWith(".amazonaws.com.cn"));
+ }
+
/**
* Interface used to bind to the socket factory, allows the code which
* works with the shaded AWS libraries to exist in their own class.
diff --git a/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/impl/RequestFactoryImpl.java b/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/impl/RequestFactoryImpl.java
index 0fddbe6726eb..ce6e8b3538f3 100644
--- a/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/impl/RequestFactoryImpl.java
+++ b/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/impl/RequestFactoryImpl.java
@@ -599,7 +599,7 @@ public class RequestFactoryImpl implements RequestFactory {
return prepareRequest(DeleteObjectsRequest
.builder()
.bucket(bucket)
- .delete(d -> d.objects(keysToDelete).quiet(true)));
+ .delete(d -> d.objects(keysToDelete).quiet(!LOG.isTraceEnabled())));
}
@Override
diff --git a/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/impl/S3ExpressStorage.java b/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/impl/S3ExpressStorage.java
new file mode 100644
index 000000000000..fde845c5d07d
--- /dev/null
+++ b/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/impl/S3ExpressStorage.java
@@ -0,0 +1,76 @@
+/*
+ * 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.impl;
+
+import static org.apache.hadoop.fs.s3a.impl.NetworkBinding.isAwsEndpoint;
+
+/**
+ * Anything needed to support Amazon S3 Express One Zone Storage.
+ * These have bucket names like {@code s3a://bucket--usw2-az2--x-s3/}
+ */
+public final class S3ExpressStorage {
+
+ /**
+ * Is this S3Express storage? value {@value}.
+ */
+ public static final String STORE_CAPABILITY_S3_EXPRESS_STORAGE =
+ "fs.s3a.capability.s3express.storage";
+
+ /**
+ * What is the official product name? used for error messages and logging: {@value}.
+ */
+ public static final String PRODUCT_NAME = "Amazon S3 Express One Zone Storage";
+
+ private S3ExpressStorage() {
+ }
+
+ /**
+ * Minimum length of a region.
+ */
+ private static final int SUFFIX_LENGTH = "--usw2-az2--x-s3".length();
+
+ public static final int ZONE_LENGTH = "usw2-az2".length();
+
+ /**
+ * Suffix of S3Express storage bucket names..
+ */
+ public static final String S3EXPRESS_STORE_SUFFIX = "--x-s3";
+
+ /**
+ * Is a bucket an S3Express store?
+ * This may get confused against third party stores, so takes the endpoint
+ * and only supports aws endpoints round the world.
+ * @param bucket bucket to probe
+ * @param endpoint endpoint string.
+ * @return true if the store is S3 Express.
+ */
+ public static boolean isS3ExpressStore(String bucket, final String endpoint) {
+ return isAwsEndpoint(endpoint) && hasS3ExpressSuffix(bucket);
+ }
+
+ /**
+ * Check for a bucket name matching -does not look at endpoint.
+ * @param bucket bucket to probe.
+ * @return true if the suffix is present
+ */
+ public static boolean hasS3ExpressSuffix(final String bucket) {
+ return bucket.endsWith(S3EXPRESS_STORE_SUFFIX);
+ }
+
+}
diff --git a/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/s3guard/S3GuardTool.java b/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/s3guard/S3GuardTool.java
index ea1ea908486e..41251d190c44 100644
--- a/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/s3guard/S3GuardTool.java
+++ b/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/s3guard/S3GuardTool.java
@@ -58,6 +58,7 @@ import org.apache.hadoop.fs.s3a.commit.InternalCommitterConstants;
import org.apache.hadoop.fs.s3a.impl.DirectoryPolicy;
import org.apache.hadoop.fs.s3a.impl.DirectoryPolicyImpl;
import org.apache.hadoop.fs.s3a.select.SelectTool;
+import org.apache.hadoop.fs.s3a.tools.BucketTool;
import org.apache.hadoop.fs.s3a.tools.MarkerTool;
import org.apache.hadoop.fs.shell.CommandFormat;
import org.apache.hadoop.fs.statistics.IOStatistics;
@@ -74,6 +75,7 @@ import static org.apache.hadoop.fs.s3a.Constants.*;
import static org.apache.hadoop.fs.s3a.Invoker.LOG_EVENT;
import static org.apache.hadoop.fs.s3a.commit.CommitConstants.*;
import static org.apache.hadoop.fs.s3a.commit.staging.StagingCommitterConstants.FILESYSTEM_TEMP_PATH;
+import static org.apache.hadoop.fs.s3a.impl.InternalConstants.S3A_DYNAMIC_CAPABILITIES;
import static org.apache.hadoop.fs.statistics.IOStatisticsLogging.ioStatisticsToPrettyString;
import static org.apache.hadoop.fs.statistics.IOStatisticsSupport.retrieveIOStatistics;
import static org.apache.hadoop.fs.statistics.StoreStatisticNames.MULTIPART_UPLOAD_ABORTED;
@@ -117,6 +119,7 @@ public abstract class S3GuardTool extends Configured implements Tool,
" [command] [OPTIONS] [s3a://BUCKET]\n\n" +
"Commands: \n" +
"\t" + BucketInfo.NAME + " - " + BucketInfo.PURPOSE + "\n" +
+ "\t" + BucketTool.NAME + " - " + BucketTool.PURPOSE + "\n" +
"\t" + MarkerTool.MARKERS + " - " + MarkerTool.PURPOSE + "\n" +
"\t" + SelectTool.NAME + " - " + SelectTool.PURPOSE + "\n" +
"\t" + Uploads.NAME + " - " + Uploads.PURPOSE + "\n";
@@ -164,8 +167,19 @@ public abstract class S3GuardTool extends Configured implements Tool,
* @param opts any boolean options to support
*/
protected S3GuardTool(Configuration conf, String... opts) {
+ this(conf, 0, Integer.MAX_VALUE, opts);
+ }
+
+ /**
+ * Constructor a S3Guard tool with HDFS configuration.
+ * @param conf Configuration.
+ * @param min min number of args
+ * @param max max number of args
+ * @param opts any boolean options to support
+ */
+ protected S3GuardTool(Configuration conf, int min, int max, String... opts) {
super(conf);
- commandFormat = new CommandFormat(0, Integer.MAX_VALUE, opts);
+ commandFormat = new CommandFormat(min, max, opts);
}
/**
@@ -208,7 +222,7 @@ public abstract class S3GuardTool extends Configured implements Tool,
return cliDelta;
}
- protected void addAgeOptions() {
+ protected final void addAgeOptions() {
CommandFormat format = getCommandFormat();
format.addOptionWithValue(DAYS_FLAG);
format.addOptionWithValue(HOURS_FLAG);
@@ -240,6 +254,22 @@ public abstract class S3GuardTool extends Configured implements Tool,
return getCommandFormat().parse(args, 1);
}
+ /**
+ * Process the arguments.
+ * @param args raw args
+ * @return process arg list.
+ * @throws ExitUtil.ExitException if there's an unknown option.
+ */
+ protected List<String> parseArgsWithErrorReporting(final String[] args)
+ throws ExitUtil.ExitException {
+ try {
+ return parseArgs(args);
+ } catch (CommandFormat.UnknownOptionException e) {
+ errorln(getUsage());
+ throw new ExitUtil.ExitException(EXIT_USAGE, e.getMessage(), e);
+ }
+ }
+
protected S3AFileSystem getFilesystem() {
return filesystem;
}
@@ -273,7 +303,7 @@ public abstract class S3GuardTool extends Configured implements Tool,
filesystem = null;
}
- protected CommandFormat getCommandFormat() {
+ protected final CommandFormat getCommandFormat() {
return commandFormat;
}
@@ -397,6 +427,7 @@ public abstract class S3GuardTool extends Configured implements Tool,
Configuration conf = fs.getConf();
URI fsUri = fs.getUri();
println(out, "Filesystem %s", fsUri);
+ final Path root = new Path("/");
try {
println(out, "Location: %s", fs.getBucketLocation());
} catch (IOException e) {
@@ -524,6 +555,13 @@ public abstract class S3GuardTool extends Configured implements Tool,
processMarkerOption(out, fs,
getCommandFormat().getOptValue(MARKERS_FLAG));
+ // and check for capabilitities
+ println(out, "%nStore Capabilities");
+ for (String capability : S3A_DYNAMIC_CAPABILITIES) {
+ out.printf("\t%s %s%n", capability,
+ fs.hasPathCapability(root, capability));
+ }
+ println(out, "");
// and finally flush the output and report a success.
out.flush();
return SUCCESS;
@@ -584,7 +622,7 @@ public abstract class S3GuardTool extends Configured implements Tool,
/**
* Command to list / abort pending multipart uploads.
*/
- static class Uploads extends S3GuardTool {
+ static final class Uploads extends S3GuardTool {
public static final String NAME = "uploads";
public static final String ABORT = "abort";
public static final String LIST = "list";
@@ -736,7 +774,7 @@ public abstract class S3GuardTool extends Configured implements Tool,
return ageDate.compareTo(Date.from(u.initiated())) >= 0;
}
- private void processArgs(List<String> args, PrintStream out)
+ protected void processArgs(List<String> args, PrintStream out)
throws IOException {
CommandFormat commands = getCommandFormat();
String err = "Can only specify one of -" + LIST + ", " +
@@ -947,9 +985,13 @@ public abstract class S3GuardTool extends Configured implements Tool,
throw s3guardUnsupported();
}
switch (subCommand) {
+
case BucketInfo.NAME:
command = new BucketInfo(conf);
break;
+ case BucketTool.NAME:
+ command = new BucketTool(conf);
+ break;
case MarkerTool.MARKERS:
command = new MarkerTool(conf);
break;
diff --git a/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/tools/BucketTool.java b/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/tools/BucketTool.java
new file mode 100644
index 000000000000..6bfcbcf776f1
--- /dev/null
+++ b/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/tools/BucketTool.java
@@ -0,0 +1,304 @@
+/*
+ * 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.tools;
+
+import java.io.FileNotFoundException;
+import java.io.PrintStream;
+import java.net.URI;
+import java.util.List;
+import java.util.Optional;
+import java.util.function.BiFunction;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import software.amazon.awssdk.services.s3.S3Client;
+import software.amazon.awssdk.services.s3.model.BucketType;
+import software.amazon.awssdk.services.s3.model.CreateBucketConfiguration;
+import software.amazon.awssdk.services.s3.model.CreateBucketRequest;
+import software.amazon.awssdk.services.s3.model.DataRedundancy;
+import software.amazon.awssdk.services.s3.model.LocationInfo;
+import software.amazon.awssdk.services.s3.model.LocationType;
+
+import org.apache.commons.lang3.StringUtils;
+import org.apache.hadoop.classification.VisibleForTesting;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.fs.s3a.S3AFileSystem;
+import org.apache.hadoop.fs.s3a.s3guard.S3GuardTool;
+import org.apache.hadoop.fs.shell.CommandFormat;
+import org.apache.hadoop.io.IOUtils;
+import org.apache.hadoop.util.DurationInfo;
+import org.apache.hadoop.util.ExitUtil;
+
+import static org.apache.commons.lang3.StringUtils.isNotEmpty;
+import static org.apache.hadoop.fs.s3a.Constants.AWS_REGION;
+import static org.apache.hadoop.fs.s3a.Constants.ENDPOINT;
+import static org.apache.hadoop.fs.s3a.Constants.S3A_BUCKET_PROBE;
+import static org.apache.hadoop.fs.s3a.impl.S3ExpressStorage.PRODUCT_NAME;
+import static org.apache.hadoop.fs.s3a.impl.S3ExpressStorage.STORE_CAPABILITY_S3_EXPRESS_STORAGE;
+import static org.apache.hadoop.fs.s3a.Invoker.once;
+import static org.apache.hadoop.fs.s3a.audit.S3AAuditConstants.REJECT_OUT_OF_SPAN_OPERATIONS;
+import static org.apache.hadoop.fs.s3a.impl.NetworkBinding.isAwsEndpoint;
+import static org.apache.hadoop.fs.s3a.impl.S3ExpressStorage.hasS3ExpressSuffix;
+import static org.apache.hadoop.service.launcher.LauncherExitCodes.EXIT_BAD_CONFIGURATION;
+import static org.apache.hadoop.service.launcher.LauncherExitCodes.EXIT_NOT_ACCEPTABLE;
+import static org.apache.hadoop.service.launcher.LauncherExitCodes.EXIT_USAGE;
+
+/**
+ * Bucket operations, e.g. create/delete/probe.
+ */
+public final class BucketTool extends S3GuardTool {
+
+ private static final Logger LOG = LoggerFactory.getLogger(BucketTool.class);
+
+ /**
+ * Name of this tool: {@value}.
+ */
+ public static final String NAME = "bucket";
+
+ /**
+ * Purpose of this tool: {@value}.
+ */
+ public static final String PURPOSE =
+ "View and manipulate S3 buckets";
+
+ /**
+ * create command.
+ */
+ public static final String CREATE = "create";
+
+ /**
+ * region {@value}.
+ */
+ public static final String OPT_REGION = "region";
+
+ /**
+ * endpoint {@value}.
+ */
+ public static final String OPT_ENDPOINT = "endpoint";
+
+ /**
+ * Zone for a store.
+ */
+ public static final String OPT_ZONE = "zone";
+
+ /**
+ * Error message if -zone is set but the name doesn't match.
+ * Value {@value}.
+ */
+ static final String UNSUPPORTED_ZONE_ARG =
+ "The -zone option is only supported for " + PRODUCT_NAME;
+
+ /**
+ * Error message if the bucket is S3 Express but -zone wasn't set.
+ * Value {@value}.
+ */
+ static final String NO_ZONE_SUPPLIED = "Required option -zone missing for "
+ + PRODUCT_NAME + " bucket";
+
+ /**
+ * Error Message logged/thrown when the tool could not start as
+ * the bucket probe was not disabled and the probe (inevitably)
+ * failed.
+ */
+ public static final String PROBE_FAILURE =
+ "Initialization failed because the bucket existence probe"
+ + S3A_BUCKET_PROBE + " was not disabled. Check core-site settings.";
+
+ public BucketTool(final Configuration conf) {
+ super(conf, 1, 1,
+ CREATE);
+ CommandFormat format = getCommandFormat();
+ format.addOptionWithValue(OPT_REGION);
+ format.addOptionWithValue(OPT_ENDPOINT);
+ format.addOptionWithValue(OPT_ZONE);
+ }
+
+ public String getUsage() {
+ return "bucket "
+ + "-" + CREATE + " "
+ + "[-" + OPT_ENDPOINT + " <endpoint>] "
+ + "[-" + OPT_REGION + " <region>] "
+ + "[-" + OPT_ZONE + " <zone>] "
+ + " <s3a-URL>";
+ }
+
+ public String getName() {
+ return NAME;
+ }
+
+ private Optional<String> getOptionalString(String key) {
+ String value = getCommandFormat().getOptValue(key);
+ return isNotEmpty(value) ? Optional.of(value) : Optional.empty();
+ }
+
+ @VisibleForTesting
+ int exec(final String...args) throws Exception {
+ return run(args, System.out);
+ }
+
+ @Override
+ public int run(final String[] args, final PrintStream out)
+ throws Exception, ExitUtil.ExitException {
+
+ LOG.debug("Supplied arguments: {}", String.join(", ", args));
+ final List<String> parsedArgs = parseArgsWithErrorReporting(args);
+
+ CommandFormat command = getCommandFormat();
+ boolean create = command.getOpt(CREATE);
+
+ Optional<String> endpoint = getOptionalString(OPT_ENDPOINT);
+ Optional<String> region = getOptionalString(OPT_REGION);
+ Optional<String> zone = getOptionalString(OPT_ZONE);
+
+
+ final String bucketPath = parsedArgs.get(0);
+ final Path source = new Path(bucketPath);
+ URI fsURI = source.toUri();
+ String bucket = fsURI.getHost();
+
+ println(out, "Filesystem %s", fsURI);
+ if (!"s3a".equals(fsURI.getScheme())) {
+ throw new ExitUtil.ExitException(EXIT_USAGE, "Filesystem is not S3A URL: " + fsURI);
+ }
+ println(out, "Options region=%s endpoint=%s zone=%s s3a://%s",
+ region.orElse("(unset)"),
+ endpoint.orElse("(unset)"),
+ zone.orElse("(unset)"),
+ bucket);
+ if (!create) {
+ errorln(getUsage());
+ println(out, "Supplied arguments: ["
+ + String.join(", ", parsedArgs)
+ + "]");
+ throw new ExitUtil.ExitException(EXIT_USAGE,
+ "required option not found: -create");
+ }
+
+ final Configuration conf = getConf();
+
+ removeBucketOverrides(bucket, conf,
+ S3A_BUCKET_PROBE,
+ REJECT_OUT_OF_SPAN_OPERATIONS,
+ AWS_REGION,
+ ENDPOINT);
+ // stop any S3 calls taking place against a bucket which
+ // may not exist
+ String bucketPrefix = "fs.s3a.bucket." + bucket + '.';
+ conf.setInt(bucketPrefix + S3A_BUCKET_PROBE, 0);
+ conf.setBoolean(bucketPrefix + REJECT_OUT_OF_SPAN_OPERATIONS, false);
+ // propagate an option
+ BiFunction<String, Optional<String>, Boolean> propagate = (key, opt) ->
+ opt.map(v -> {
+ conf.set(key, v);
+ LOG.info("{} = {}", key, v);
+ return true;
+ }).orElse(false);
+
+
+ propagate.apply(AWS_REGION, region);
+ propagate.apply(ENDPOINT, endpoint);
+
+ // fail fast on third party store
+ if (hasS3ExpressSuffix(bucket) && !isAwsEndpoint(endpoint.orElse(""))) {
+ throw new ExitUtil.ExitException(EXIT_NOT_ACCEPTABLE, UNSUPPORTED_ZONE_ARG);
+ }
+ S3AFileSystem fs;
+ try {
+ fs = (S3AFileSystem) FileSystem.newInstance(fsURI, conf);
+ } catch (FileNotFoundException e) {
+ // this happens if somehow the probe wasn't disabled.
+ errorln(PROBE_FAILURE);
+ throw new ExitUtil.ExitException(EXIT_BAD_CONFIGURATION, PROBE_FAILURE);
+ }
+
+ try {
+
+ // now build the configuration
+ final CreateBucketConfiguration.Builder builder = CreateBucketConfiguration.builder();
+
+ if (fs.hasPathCapability(new Path("/"), STORE_CAPABILITY_S3_EXPRESS_STORAGE)) {
+ // S3 Express store requires a zone and some other other settings
+ final String az = zone.orElseThrow(() ->
+ new ExitUtil.ExitException(EXIT_USAGE, NO_ZONE_SUPPLIED + bucket));
+ builder.location(LocationInfo.builder()
+ .type(LocationType.AVAILABILITY_ZONE).name(az).build())
+ .bucket(software.amazon.awssdk.services.s3.model.BucketInfo.builder()
+ .type(BucketType.DIRECTORY)
+ .dataRedundancy(DataRedundancy.SINGLE_AVAILABILITY_ZONE).build());
+
+ } else {
+ if (zone.isPresent()) {
+ throw new ExitUtil.ExitException(EXIT_USAGE, UNSUPPORTED_ZONE_ARG + " not " + bucket);
+ }
+ region.ifPresent(builder::locationConstraint);
+ }
+
+ final CreateBucketRequest request = CreateBucketRequest.builder()
+ .bucket(bucket)
+ .createBucketConfiguration(builder.build())
+ .build();
+
+ println(out, "Creating bucket %s", bucket);
+
+ final S3Client s3Client = fs.getS3AInternals().getAmazonS3Client(NAME);
+ try (DurationInfo ignored = new DurationInfo(LOG,
+ "Create %sbucket %s in region %s",
+ (fs.hasPathCapability(new Path("/"),
+ STORE_CAPABILITY_S3_EXPRESS_STORAGE) ? (PRODUCT_NAME + " "): ""),
+ bucket, region.orElse("(unset)"))) {
+ once("create", source.toString(), () ->
+ s3Client.createBucket(request));
+ }
+ } finally {
+ IOUtils.closeStream(fs);
+ }
+
+ return 0;
+ }
+
+ /**
+ * Remove any values from a bucket.
+ * @param bucket bucket whose overrides are to be removed. Can be null/empty
+ * @param conf config
+ * @param options list of fs.s3a options to remove
+ */
+ public static void removeBucketOverrides(final String bucket,
+ final Configuration conf,
+ final String... options) {
+
+ if (StringUtils.isEmpty(bucket)) {
+ return;
+ }
+ final String bucketPrefix = "fs.s3a.bucket." + bucket + '.';
+ for (String option : options) {
+ final String stripped = option.substring("fs.s3a.".length());
+ String target = bucketPrefix + stripped;
+ String v = conf.get(target);
+ if (v != null) {
+ conf.unset(target);
+ }
+ String extended = bucketPrefix + option;
+ if (conf.get(extended) != null) {
+ conf.unset(extended);
+ }
+ }
+ }
+}
diff --git a/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/tools/MarkerTool.java b/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/tools/MarkerTool.java
index fc60f4b623dd..b872d1431cb2 100644
--- a/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/tools/MarkerTool.java
+++ b/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/tools/MarkerTool.java
@@ -237,13 +237,7 @@ public final class MarkerTool extends S3GuardTool {
public int run(final String[] args, final PrintStream stream)
throws ExitUtil.ExitException, Exception {
this.out = stream;
- final List<String> parsedArgs;
- try {
- parsedArgs = parseArgs(args);
- } catch (CommandFormat.UnknownOptionException e) {
- errorln(getUsage());
- throw new ExitUtil.ExitException(EXIT_USAGE, e.getMessage(), e);
- }
+ final List<String> parsedArgs = parseArgsWithErrorReporting(args);
if (parsedArgs.size() != 1) {
errorln(getUsage());
println(out, "Supplied arguments: ["
diff --git a/hadoop-tools/hadoop-aws/src/site/markdown/tools/hadoop-aws/aws_sdk_upgrade.md b/hadoop-tools/hadoop-aws/src/site/markdown/tools/hadoop-aws/aws_sdk_upgrade.md
index 138e060155c0..8a3fd16ec172 100644
--- a/hadoop-tools/hadoop-aws/src/site/markdown/tools/hadoop-aws/aws_sdk_upgrade.md
+++ b/hadoop-tools/hadoop-aws/src/site/markdown/tools/hadoop-aws/aws_sdk_upgrade.md
@@ -288,6 +288,16 @@ public interface S3AInternals {
HeadObjectResponse getObjectMetadata(Path path) throws IOException;
AWSCredentialProviderList shareCredentials(final String purpose);
+
+ @AuditEntryPoint
+ @Retries.RetryTranslated
+ HeadBucketResponse getBucketMetadata() throws IOException;
+
+ boolean isMultipartCopyEnabled();
+
+ @AuditEntryPoint
+ @Retries.RetryTranslated
+ long abortMultipartUploads(Path path) throws IOException;
}
```
diff --git a/hadoop-tools/hadoop-aws/src/site/markdown/tools/hadoop-aws/third_party_stores.md b/hadoop-tools/hadoop-aws/src/site/markdown/tools/hadoop-aws/third_party_stores.md
index 0216e46014c7..1aa6e83b110e 100644
--- a/hadoop-tools/hadoop-aws/src/site/markdown/tools/hadoop-aws/third_party_stores.md
+++ b/hadoop-tools/hadoop-aws/src/site/markdown/tools/hadoop-aws/third_party_stores.md
@@ -41,9 +41,18 @@ The features which may be unavailable include:
* List API to use (`fs.s3a.list.version = 1`)
* Bucket lifecycle rules to clean up pending uploads.
-## Configuring s3a to connect to a third party store
+### Disabling Change Detection
+The (default) etag-based change detection logic expects stores to provide an Etag header in HEAD/GET requests,
+and to support it as a precondition in subsequent GET and COPY calls.
+If a store does not do this, disable the checks.
+```xml
+<property>
+ <name>fs.s3a.change.detection.mode</name>
+ <value>none</value>
+</property>
+```
## Connecting to a third party object store over HTTPS
The core setting for a third party store is to change the endpoint in `fs.s3a.endpoint`.
diff --git a/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/AbstractTestS3AEncryption.java b/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/AbstractTestS3AEncryption.java
index 828eb5127cbf..583a41fa62c5 100644
--- a/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/AbstractTestS3AEncryption.java
+++ b/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/AbstractTestS3AEncryption.java
@@ -49,6 +49,7 @@ public abstract class AbstractTestS3AEncryption extends AbstractS3ATestBase {
@Override
protected Configuration createConfiguration() {
Configuration conf = super.createConfiguration();
+ skipIfEncryptionTestsDisabled(conf);
S3ATestUtils.disableFilesystemCaching(conf);
patchConfigurationEncryptionSettings(conf);
return conf;
diff --git a/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/ITestS3AConfiguration.java b/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/ITestS3AConfiguration.java
index d9c847d4b2b0..8787fca431cc 100644
--- a/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/ITestS3AConfiguration.java
+++ b/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/ITestS3AConfiguration.java
@@ -131,7 +131,7 @@ public class ITestS3AConfiguration {
S3ATestConstants.CONFIGURATION_TEST_ENDPOINT, "");
if (endpoint.isEmpty()) {
LOG.warn("Custom endpoint test skipped as " +
- S3ATestConstants.CONFIGURATION_TEST_ENDPOINT + "config " +
+ S3ATestConstants.CONFIGURATION_TEST_ENDPOINT + " config " +
"setting was not detected");
} else {
conf.set(Constants.ENDPOINT, endpoint);
diff --git a/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/ITestS3AEncryptionSSEC.java b/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/ITestS3AEncryptionSSEC.java
index 45b0c6c206f2..321f831c0a8d 100644
--- a/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/ITestS3AEncryptionSSEC.java
+++ b/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/ITestS3AEncryptionSSEC.java
@@ -49,6 +49,7 @@ import static org.apache.hadoop.fs.s3a.S3ATestUtils.createTestPath;
import static org.apache.hadoop.fs.s3a.S3ATestUtils.disableFilesystemCaching;
import static org.apache.hadoop.fs.s3a.S3ATestUtils.getTestBucketName;
import static org.apache.hadoop.fs.s3a.S3ATestUtils.removeBaseAndBucketOverrides;
+import static org.apache.hadoop.fs.s3a.S3ATestUtils.skipIfEncryptionTestsDisabled;
import static org.apache.hadoop.test.LambdaTestUtils.intercept;
/**
@@ -104,7 +105,6 @@ public class ITestS3AEncryptionSSEC extends AbstractTestS3AEncryption {
@Override
protected Configuration createConfiguration() {
Configuration conf = super.createConfiguration();
- disableFilesystemCaching(conf);
String bucketName = getTestBucketName(conf);
// directory marker options
removeBaseAndBucketOverrides(bucketName, conf,
diff --git a/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/ITestS3AMiscOperations.java b/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/ITestS3AMiscOperations.java
index 66a211181bdd..a375044add43 100644
--- a/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/ITestS3AMiscOperations.java
+++ b/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/ITestS3AMiscOperations.java
@@ -19,18 +19,12 @@
package org.apache.hadoop.fs.s3a;
import java.io.ByteArrayInputStream;
-import java.io.FileNotFoundException;
import java.io.IOException;
import java.net.URI;
import java.nio.charset.StandardCharsets;
-import java.nio.file.AccessDeniedException;
-import software.amazon.awssdk.services.s3.S3Client;
-import software.amazon.awssdk.services.s3.model.GetBucketEncryptionRequest;
-import software.amazon.awssdk.services.s3.model.GetBucketEncryptionResponse;
import software.amazon.awssdk.services.s3.model.PutObjectRequest;
import org.assertj.core.api.Assertions;
-import org.junit.Assume;
import org.junit.Test;
@@ -45,7 +39,6 @@ import org.apache.hadoop.fs.store.audit.AuditSpan;
import org.apache.hadoop.fs.store.EtagChecksum;
import org.apache.hadoop.test.LambdaTestUtils;
-import static org.apache.hadoop.fs.contract.ContractTestUtils.assertHasPathCapabilities;
import static org.apache.hadoop.fs.contract.ContractTestUtils.assertLacksPathCapabilities;
import static org.apache.hadoop.fs.contract.ContractTestUtils.createFile;
import static org.apache.hadoop.fs.contract.ContractTestUtils.touch;
@@ -55,7 +48,6 @@ import static org.apache.hadoop.fs.s3a.Constants.SERVER_SIDE_ENCRYPTION_ALGORITH
import static org.apache.hadoop.fs.s3a.Constants.SERVER_SIDE_ENCRYPTION_KEY;
import static org.apache.hadoop.fs.s3a.S3ATestUtils.removeBaseAndBucketOverrides;
import static org.apache.hadoop.fs.s3a.impl.HeaderProcessing.XA_ETAG;
-import static org.hamcrest.Matchers.nullValue;
/**
* Tests of the S3A FileSystem which don't have a specific home and can share
@@ -153,39 +145,6 @@ public class ITestS3AMiscOperations extends AbstractS3ATestBase {
return f;
}
- /**
- * The assumption here is that 0-byte files uploaded in a single PUT
- * always have the same checksum, including stores with encryption.
- * This will be skipped if the bucket has S3 default encryption enabled.
- * @throws Throwable on a failure
- */
- @Test
- public void testEmptyFileChecksums() throws Throwable {
- assumeNoDefaultEncryption();
- final S3AFileSystem fs = getFileSystem();
- Path file1 = touchFile("file1");
- EtagChecksum checksum1 = fs.getFileChecksum(file1, 0);
- LOG.info("Checksum for {}: {}", file1, checksum1);
- assertHasPathCapabilities(fs, file1,
- CommonPathCapabilities.FS_CHECKSUMS);
- assertNotNull("Null file 1 checksum", checksum1);
- assertNotEquals("file 1 checksum", 0, checksum1.getLength());
- assertEquals("checksums of empty files", checksum1,
- fs.getFileChecksum(touchFile("file2"), 0));
- Assertions.assertThat(fs.getXAttr(file1, XA_ETAG))
- .describedAs("etag from xattr")
- .isEqualTo(checksum1.getBytes());
- }
-
- /**
- * Skip a test if we can get the default encryption on a bucket and it is
- * non-null.
- */
- private void assumeNoDefaultEncryption() throws IOException {
- skipIfClientSideEncryption();
- Assume.assumeThat(getDefaultEncryption(), nullValue());
- }
-
/**
* Make sure that when checksums are disabled, the caller
* gets null back.
@@ -226,24 +185,6 @@ public class ITestS3AMiscOperations extends AbstractS3ATestBase {
.isEqualTo(checksum1.getBytes());
}
- /**
- * Verify that on an unencrypted store, the checksum of two non-empty
- * (single PUT) files is the same if the data is the same.
- * This will be skipped if the bucket has S3 default encryption enabled.
- * @throws Throwable failure
- */
- @Test
- public void testNonEmptyFileChecksumsUnencrypted() throws Throwable {
- Assume.assumeTrue(encryptionAlgorithm().equals(S3AEncryptionMethods.NONE));
- assumeNoDefaultEncryption();
- final S3AFileSystem fs = getFileSystem();
- final EtagChecksum checksum1 =
- fs.getFileChecksum(mkFile("file5", HELLO), 0);
- assertNotNull("file 3 checksum", checksum1);
- assertEquals("checksums", checksum1,
- fs.getFileChecksum(mkFile("file6", HELLO), 0));
- }
-
private S3AEncryptionMethods encryptionAlgorithm() {
return getFileSystem().getS3EncryptionAlgorithm();
}
@@ -399,22 +340,4 @@ public class ITestS3AMiscOperations extends AbstractS3ATestBase {
return o;
}
- /**
- * Gets default encryption settings for the bucket or returns null if default
- * encryption is disabled.
- */
- private GetBucketEncryptionResponse getDefaultEncryption() throws IOException {
- S3AFileSystem fs = getFileSystem();
- S3Client s3 = getS3AInternals().getAmazonS3Client("check default encryption");
- try (AuditSpan s = span()){
- return Invoker.once("getBucketEncryption()",
- fs.getBucket(),
- () -> s3.getBucketEncryption(GetBucketEncryptionRequest.builder()
- .bucket(fs.getBucket())
- .build()));
- } catch (FileNotFoundException | AccessDeniedException | AWSBadRequestException e) {
- return null;
- }
- }
-
}
diff --git a/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/MultipartTestUtils.java b/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/MultipartTestUtils.java
index 3f6870be46b2..20e5c12ec79c 100644
--- a/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/MultipartTestUtils.java
+++ b/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/MultipartTestUtils.java
@@ -26,7 +26,6 @@ import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.RemoteIterator;
import org.apache.hadoop.fs.s3a.impl.PutObjectOptions;
import org.apache.hadoop.fs.store.audit.AuditSpan;
-import org.apache.hadoop.io.IOUtils;
import org.junit.Assert;
import org.slf4j.Logger;
@@ -35,22 +34,31 @@ import org.slf4j.LoggerFactory;
import java.io.ByteArrayInputStream;
import java.io.IOException;
import java.io.InputStream;
+import java.nio.charset.StandardCharsets;
import java.util.Date;
import java.util.List;
import java.util.Objects;
import java.util.Set;
import java.util.stream.Collectors;
+import static org.apache.hadoop.fs.contract.ContractTestUtils.assertFileHasLength;
+import static org.apache.hadoop.fs.contract.ContractTestUtils.createFile;
import static org.apache.hadoop.fs.contract.ContractTestUtils.dataset;
-import static org.apache.hadoop.fs.s3a.Invoker.LOG_EVENT;
+import static org.apache.hadoop.fs.s3a.commit.CommitConstants.MAGIC_PATH_PREFIX;
/**
* Utilities for S3A multipart upload tests.
*/
public final class MultipartTestUtils {
+
private static final Logger LOG = LoggerFactory.getLogger(
MultipartTestUtils.class);
+ /**
+ * Target file of {@code createMagicFile()}.
+ */
+ public static final String MAGIC_FILE_TARGET = "subdir/file.txt";
+
/** Not instantiated. */
private MultipartTestUtils() { }
@@ -94,24 +102,10 @@ public final class MultipartTestUtils {
/** Delete any uploads under given path (recursive). Silent on failure. */
public static void clearAnyUploads(S3AFileSystem fs, Path path) {
- String key = fs.pathToKey(path);
- AuditSpan span = null;
try {
- RemoteIterator<MultipartUpload> uploads = fs.listUploads(key);
- span = fs.createSpan("multipart", path.toString(), null);
- final WriteOperationHelper helper
- = fs.getWriteOperationHelper();
- while (uploads.hasNext()) {
- MultipartUpload upload = uploads.next();
- LOG.debug("Cleaning up upload: {} {}", upload.key(),
- truncatedUploadId(upload.uploadId()));
- helper.abortMultipartUpload(upload.key(),
- upload.uploadId(), true, LOG_EVENT);
- }
+ fs.getS3AInternals().abortMultipartUploads(path);
} catch (IOException ioe) {
LOG.info("Ignoring exception: ", ioe);
- } finally {
- IOUtils.closeStream(span);
}
}
@@ -165,6 +159,33 @@ public final class MultipartTestUtils {
return fullId.substring(0, 12) + " ...";
}
+ /**
+ * Given a dir, return the name of the magic subdir.
+ * The naming has changed across versions; this isolates
+ * the changes.
+ * @param dir directory
+ * @return the magic subdir
+ */
+ public static Path magicPath(Path dir) {
+ return new Path(dir, MAGIC_PATH_PREFIX + "001/");
+ }
+
+ /**
+ * Create a magic file of "real" length more than 0 bytes long.
+ * @param fs filesystem
+ * @param dir directory
+ * @return the path
+ * @throws IOException creation failure.p
+ */
+ public static Path createMagicFile(final S3AFileSystem fs, final Path dir) throws IOException {
+ Path magicFile = new Path(magicPath(dir), "__base/" + MAGIC_FILE_TARGET);
+ createFile(fs, magicFile, true, "123".getBytes(StandardCharsets.UTF_8));
+
+ // the file exists but is a 0 byte marker file.
+ assertFileHasLength(fs, magicFile, 0);
+ return magicFile;
+ }
+
/** Struct of object key, upload ID. */
public static class IdKey {
private String key;
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 aa38186c6503..dbe5aaf22288 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
@@ -22,6 +22,7 @@ import org.apache.commons.lang3.StringUtils;
import org.apache.hadoop.classification.InterfaceAudience;
import org.apache.hadoop.classification.InterfaceStability;
import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.EtagSource;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FileContext;
@@ -37,6 +38,7 @@ import org.apache.hadoop.fs.s3a.auth.MarshalledCredentials;
import org.apache.hadoop.fs.s3a.auth.delegation.EncryptionSecrets;
import org.apache.hadoop.fs.s3a.impl.ChangeDetectionPolicy;
import org.apache.hadoop.fs.s3a.impl.ContextAccessors;
+import org.apache.hadoop.fs.s3a.impl.NetworkBinding;
import org.apache.hadoop.fs.s3a.impl.StatusProbeEnum;
import org.apache.hadoop.fs.s3a.impl.StoreContext;
import org.apache.hadoop.fs.s3a.impl.StoreContextBuilder;
@@ -57,6 +59,8 @@ import org.apache.hadoop.thirdparty.com.google.common.util.concurrent.ListeningE
import org.apache.hadoop.thirdparty.com.google.common.util.concurrent.MoreExecutors;
import org.apache.hadoop.util.BlockingThreadPoolExecutorService;
import org.apache.hadoop.util.DurationInfo;
+import org.apache.hadoop.util.ExitUtil;
+import org.apache.hadoop.util.Preconditions;
import org.apache.hadoop.util.ReflectionUtils;
import org.apache.hadoop.util.functional.CallableRaisingIOE;
import org.apache.hadoop.util.functional.FutureIO;
@@ -89,6 +93,7 @@ import java.util.stream.Collectors;
import static org.apache.hadoop.fs.contract.ContractTestUtils.createFile;
import static org.apache.hadoop.fs.s3a.impl.CallableSupplier.submit;
import static org.apache.hadoop.fs.s3a.impl.CallableSupplier.waitForCompletion;
+import static org.apache.hadoop.fs.s3a.impl.S3ExpressStorage.STORE_CAPABILITY_S3_EXPRESS_STORAGE;
import static org.apache.hadoop.test.GenericTestUtils.buildPaths;
import static org.apache.hadoop.util.Preconditions.checkNotNull;
import static org.apache.hadoop.fs.CommonConfigurationKeysPublic.HADOOP_SECURITY_CREDENTIAL_PROVIDER_PATH;
@@ -98,6 +103,8 @@ import static org.apache.hadoop.fs.s3a.S3ATestConstants.*;
import static org.apache.hadoop.fs.s3a.Constants.*;
import static org.apache.hadoop.fs.s3a.S3AUtils.buildEncryptionSecrets;
import static org.apache.hadoop.test.LambdaTestUtils.intercept;
+import static org.apache.hadoop.util.functional.RemoteIterators.mappingRemoteIterator;
+import static org.apache.hadoop.util.functional.RemoteIterators.toList;
import static org.junit.Assert.*;
/**
@@ -981,6 +988,54 @@ public final class S3ATestUtils {
}
}
+ /**
+ * Given a RemoteIterator to a list of file statuses, return a list of paths.
+ * @param i iterator
+ * @return list of paths
+ * @param <T> type of status
+ * @throws IOException failure retrieving values from the iterator
+ */
+ public static <T extends FileStatus> List<Path> toPathList(RemoteIterator<T> i)
+ throws IOException {
+ return toList(mappingRemoteIterator(i, FileStatus::getPath));
+ }
+
+ /**
+ * Expect an error code from the exception.
+ * @param code error code
+ * @param error exception
+ */
+ public static void expectErrorCode(final int code, final ExitUtil.ExitException error) {
+ if (error.getExitCode() != code) {
+ throw error;
+ }
+ }
+
+ /**
+ * Require a test case to be against Amazon S3 Express store.
+ */
+ public static void assumeS3ExpressFileSystem(final FileSystem fs) throws IOException {
+ assume("store is not S3 Express: " + fs.getUri(),
+ fs.hasPathCapability(new Path("/"), STORE_CAPABILITY_S3_EXPRESS_STORAGE));
+ }
+
+ /**
+ * Require a test case to be against a standard S3 store.
+ */
+ public static void assumeNotS3ExpressFileSystem(final FileSystem fs) throws IOException {
+ assume("store is S3 Express: " + fs.getUri(),
+ !fs.hasPathCapability(new Path("/"), STORE_CAPABILITY_S3_EXPRESS_STORAGE));
+ }
+
+ /**
+ * Require a store to be hosted by Amazon -i.e. not a third party store.
+ */
+ public static void assumeStoreAwsHosted(final FileSystem fs) {
+ assume("store is not AWS S3",
+ !NetworkBinding.isAwsEndpoint(fs.getConf()
+ .getTrimmed(ENDPOINT, DEFAULT_ENDPOINT)));
+ }
+
/**
* Helper class to do diffs of metrics.
*/
@@ -1566,5 +1621,26 @@ public final class S3ATestUtils {
return fs.hasPathCapability(new Path("/"), FS_S3A_CREATE_PERFORMANCE_ENABLED);
}
+ /**
+ * Is the filesystem connector bonded to S3Express storage?
+ * @param fs filesystem.
+ * @return true if the store has the relevant path capability.
+ * @throws IOException IO failure
+ */
+ public static boolean isS3ExpressStorage(FileSystem fs) throws IOException {
+ return fs.hasPathCapability(new Path("/"), STORE_CAPABILITY_S3_EXPRESS_STORAGE);
+ }
+ /**
+ * Get an etag from a FileStatus which must implement
+ * the {@link EtagSource} interface -which S3AFileStatus does.
+ *
+ * @param status the status.
+ * @return the etag
+ */
+ public static String etag(FileStatus status) {
+ Preconditions.checkArgument(status instanceof EtagSource,
+ "Not an EtagSource: %s", status);
+ return ((EtagSource) status).getEtag();
+ }
}
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 fe48cd891157..58bb2a5e491f 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
@@ -20,14 +20,18 @@ package org.apache.hadoop.fs.s3a.auth;
import java.io.IOException;
import java.security.PrivilegedExceptionAction;
+import java.util.Arrays;
+import java.util.Collection;
import java.util.HashMap;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.atomic.AtomicInteger;
-import software.amazon.awssdk.arns.Arn;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
import software.amazon.awssdk.auth.signer.Aws4Signer;
import software.amazon.awssdk.auth.signer.AwsS3V4Signer;
+import software.amazon.awssdk.auth.signer.internal.AbstractAwsS3V4Signer;
import software.amazon.awssdk.core.interceptor.ExecutionAttributes;
import software.amazon.awssdk.core.signer.Signer;
import software.amazon.awssdk.http.SdkHttpFullRequest;
@@ -38,8 +42,10 @@ import org.slf4j.LoggerFactory;
import org.apache.hadoop.classification.InterfaceAudience.Private;
import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.ContentSummary;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.fs.contract.ContractTestUtils;
import org.apache.hadoop.fs.s3a.AbstractS3ATestBase;
import org.apache.hadoop.fs.s3a.Constants;
import org.apache.hadoop.fs.s3a.S3AFileSystem;
@@ -48,12 +54,18 @@ import org.apache.hadoop.fs.s3a.auth.delegation.DelegationTokenProvider;
import org.apache.hadoop.security.UserGroupInformation;
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.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;
/**
* Tests for custom Signers and SignerInitializers.
+ * Because the v2 sdk has had some problems with bulk delete
+ * and custom signing, this suite is parameterized.
*/
+@RunWith(Parameterized.class)
public class ITestCustomSigner extends AbstractS3ATestBase {
private static final Logger LOG = LoggerFactory
@@ -62,10 +74,33 @@ 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";
+ /**
+ * Parameterization.
+ */
+ @Parameterized.Parameters(name = "{0}")
+ public static Collection<Object[]> params() {
+ return Arrays.asList(new Object[][]{
+ {"bulk delete", true},
+ {"simple-delete", false},
+ });
+ }
+
+ private final boolean bulkDelete;
+
+ private final UserGroupInformation ugi1 = UserGroupInformation.createRemoteUser("user1");
+
+ private final UserGroupInformation ugi2 = UserGroupInformation.createRemoteUser("user2");
+
private String regionName;
private String endpoint;
+ public ITestCustomSigner(
+ final String ignored,
+ final boolean bulkDelete) {
+ this.bulkDelete = bulkDelete;
+ }
+
@Override
public void setup() throws Exception {
super.setup();
@@ -79,6 +114,17 @@ public class ITestCustomSigner extends AbstractS3ATestBase {
}
LOG.info("Determined region name to be [{}] for bucket [{}]", regionName,
fs.getBucket());
+ CustomSignerInitializer.reset();
+ }
+
+ /**
+ * Teardown closes all filesystems for the test UGIs.
+ */
+ @Override
+ public void teardown() throws Exception {
+ super.teardown();
+ FileSystem.closeAllForUGI(ugi1);
+ FileSystem.closeAllForUGI(ugi2);
}
@Test
@@ -86,12 +132,10 @@ public class ITestCustomSigner extends AbstractS3ATestBase {
throws IOException, InterruptedException {
final Path basePath = path(getMethodName());
- UserGroupInformation ugi1 = UserGroupInformation.createRemoteUser("user1");
- FileSystem fs1 = runMkDirAndVerify(ugi1,
+ FileSystem fs1 = runStoreOperationsAndVerify(ugi1,
new Path(basePath, "customsignerpath1"), "id1");
- UserGroupInformation ugi2 = UserGroupInformation.createRemoteUser("user2");
- FileSystem fs2 = runMkDirAndVerify(ugi2,
+ FileSystem fs2 = runStoreOperationsAndVerify(ugi2,
new Path(basePath, "customsignerpath2"), "id2");
Assertions.assertThat(CustomSignerInitializer.knownStores.size())
@@ -104,14 +148,15 @@ public class ITestCustomSigner extends AbstractS3ATestBase {
.as("Num registered stores mismatch").isEqualTo(0);
}
- private FileSystem runMkDirAndVerify(UserGroupInformation ugi,
+ private S3AFileSystem runStoreOperationsAndVerify(UserGroupInformation ugi,
Path finalPath, String identifier)
throws IOException, InterruptedException {
Configuration conf = createTestConfig(identifier);
- return ugi.doAs((PrivilegedExceptionAction<FileSystem>) () -> {
+ return ugi.doAs((PrivilegedExceptionAction<S3AFileSystem>) () -> {
int instantiationCount = CustomSigner.getInstantiationCount();
int invocationCount = CustomSigner.getInvocationCount();
- FileSystem fs = finalPath.getFileSystem(conf);
+ S3AFileSystem fs = (S3AFileSystem)finalPath.getFileSystem(conf);
+
fs.mkdirs(finalPath);
Assertions.assertThat(CustomSigner.getInstantiationCount())
.as("CustomSigner Instantiation count lower than expected")
@@ -130,6 +175,22 @@ public class ITestCustomSigner extends AbstractS3ATestBase {
.as("Configuration TEST_KEY mismatch in %s", CustomSigner.description())
.isEqualTo(identifier);
+ // now do some more operations to make sure all is good.
+ final Path subdir = new Path(finalPath, "year=1970/month=1/day=1");
+ fs.mkdirs(subdir);
+
+ final Path file1 = new Path(subdir, "file1");
+ ContractTestUtils.touch(fs, new Path(subdir, "file1"));
+ fs.listStatus(subdir);
+ fs.delete(file1, false);
+ 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);
return fs;
});
}
@@ -143,10 +204,15 @@ public class ITestCustomSigner extends AbstractS3ATestBase {
private Configuration createTestConfig(String identifier) {
Configuration conf = createConfiguration();
+ removeBaseAndBucketOverrides(conf,
+ CUSTOM_SIGNERS,
+ SIGNING_ALGORITHM_S3,
+ ENABLE_MULTI_DELETE);
conf.set(CUSTOM_SIGNERS,
"CustomS3Signer:" + CustomSigner.class.getName() + ":"
+ CustomSignerInitializer.class.getName());
conf.set(SIGNING_ALGORITHM_S3, "CustomS3Signer");
+ conf.setBoolean(ENABLE_MULTI_DELETE, bulkDelete);
conf.set(TEST_ID_KEY, identifier);
conf.set(TEST_REGION_KEY, regionName);
@@ -162,7 +228,7 @@ public class ITestCustomSigner extends AbstractS3ATestBase {
}
@Private
- public static final class CustomSigner implements Signer {
+ public static final class CustomSigner extends AbstractAwsS3V4Signer implements Signer {
private static final AtomicInteger INSTANTIATION_COUNT =
@@ -202,7 +268,8 @@ public class ITestCustomSigner extends AbstractS3ATestBase {
try {
lastStoreValue = CustomSignerInitializer
.getStoreValue(bucketName, UserGroupInformation.getCurrentUser());
- LOG.info("Store value for bucket {} is {}", bucketName, lastStoreValue);
+ LOG.info("Store value for bucket {} is {}", bucketName,
+ (lastStoreValue == null) ? "unset" : "set");
} catch (IOException e) {
throw new RuntimeException("Failed to get current Ugi " + e, e);
}
@@ -216,35 +283,7 @@ public class ITestCustomSigner extends AbstractS3ATestBase {
}
private String parseBucketFromHost(String host) {
- String[] hostBits = host.split("\\.");
- String bucketName = hostBits[0];
- String service = hostBits[1];
-
- if (bucketName.equals("kms")) {
- return bucketName;
- }
-
- if (service.contains("s3-accesspoint") || service.contains("s3-outposts")
- || service.contains("s3-object-lambda")) {
- // If AccessPoint then bucketName is of format `accessPoint-accountId`;
- String[] accessPointBits = bucketName.split("-");
- String accountId = accessPointBits[accessPointBits.length - 1];
- // Extract the access point name from bucket name. eg: if bucket name is
- // test-custom-signer-<accountId>, get the access point name test-custom-signer by removing
- // -<accountId> from the bucket name.
- String accessPointName =
- bucketName.substring(0, bucketName.length() - (accountId.length() + 1));
- Arn arn = Arn.builder()
- .accountId(accountId)
- .partition("aws")
- .region(hostBits[2])
- .resource("accesspoint" + "/" + accessPointName)
- .service("s3").build();
-
- bucketName = arn.toString();
- }
-
- return bucketName;
+ return CustomSdkSigner.parseBucketFromHost(host);
}
public static int getInstantiationCount() {
@@ -287,6 +326,13 @@ public class ITestCustomSigner extends AbstractS3ATestBase {
knownStores.remove(storeKey);
}
+ /**
+ * Clear the stores; invoke during test setup.
+ */
+ public static void reset() {
+ knownStores.clear();
+ }
+
public static StoreValue getStoreValue(String bucketName,
UserGroupInformation ugi) {
StoreKey storeKey = new StoreKey(bucketName, ugi);
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 64def00fd287..ee1e8328a771 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
@@ -619,8 +619,10 @@ public class ITestCommitOperations extends AbstractCommitITest {
commits.add(commit1);
}
- assertPathDoesNotExist("destination dir", destDir);
- assertPathDoesNotExist("subdirectory", subdir);
+ if (!isS3ExpressStorage(fs)) {
+ assertPathDoesNotExist("destination dir", destDir);
+ assertPathDoesNotExist("subdirectory", subdir);
+ }
LOG.info("Initiating commit operations");
try (CommitContext commitContext
= actions.createCommitContextForTesting(destDir, JOB_ID, 0)) {
diff --git a/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/impl/ITestTreewalkProblems.java b/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/impl/ITestTreewalkProblems.java
new file mode 100644
index 000000000000..286a92f78934
--- /dev/null
+++ b/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/impl/ITestTreewalkProblems.java
@@ -0,0 +1,463 @@
+/*
+ * 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.impl;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import java.util.stream.Collectors;
+
+import org.assertj.core.api.Assertions;
+import org.junit.Test;
+import software.amazon.awssdk.services.s3.model.MultipartUpload;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.ContentSummary;
+import org.apache.hadoop.fs.FileStatus;
+import org.apache.hadoop.fs.FsShell;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.fs.RemoteIterator;
+import org.apache.hadoop.fs.contract.ContractTestUtils;
+import org.apache.hadoop.fs.s3a.S3AFileSystem;
+import org.apache.hadoop.fs.s3a.performance.AbstractS3ACostTest;
+import org.apache.hadoop.fs.store.audit.AuditSpan;
+import org.apache.hadoop.mapred.JobConf;
+import org.apache.hadoop.mapred.LocatedFileStatusFetcher;
+import org.apache.hadoop.mapreduce.JobID;
+import org.apache.hadoop.mapreduce.lib.input.FileSplit;
+import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
+import org.apache.hadoop.mapreduce.task.JobContextImpl;
+import org.apache.hadoop.tools.DistCpConstants;
+import org.apache.hadoop.tools.util.DistCpTestUtils;
+
+import static org.apache.hadoop.fs.CommonPathCapabilities.DIRECTORY_LISTING_INCONSISTENT;
+import static org.apache.hadoop.fs.contract.ContractTestUtils.assertHasPathCapabilities;
+import static org.apache.hadoop.fs.contract.ContractTestUtils.touch;
+import static org.apache.hadoop.fs.contract.ContractTestUtils.treeWalk;
+import static org.apache.hadoop.fs.s3a.Constants.DIRECTORY_OPERATIONS_PURGE_UPLOADS;
+import static org.apache.hadoop.fs.s3a.MultipartTestUtils.assertNoUploadsAt;
+import static org.apache.hadoop.fs.s3a.MultipartTestUtils.clearAnyUploads;
+import static org.apache.hadoop.fs.s3a.MultipartTestUtils.createMagicFile;
+import static org.apache.hadoop.fs.s3a.MultipartTestUtils.magicPath;
+import static org.apache.hadoop.fs.s3a.S3ATestUtils.removeBaseAndBucketOverrides;
+import static org.apache.hadoop.fs.s3a.S3ATestUtils.toPathList;
+import static org.apache.hadoop.fs.s3a.S3AUtils.HIDDEN_FILE_FILTER;
+import static org.apache.hadoop.fs.s3a.commit.CommitConstants.MAGIC_COMMITTER_ENABLED;
+import static org.apache.hadoop.mapreduce.lib.input.FileInputFormat.LIST_STATUS_NUM_THREADS;
+import static org.apache.hadoop.util.ToolRunner.run;
+import static org.apache.hadoop.util.functional.RemoteIterators.foreach;
+import static org.apache.hadoop.util.functional.RemoteIterators.remoteIteratorFromArray;
+import static org.apache.hadoop.util.functional.RemoteIterators.toList;
+
+/**
+ * Test behavior of treewalking when there are pending
+ * uploads. All commands MUST work.
+ * Currently, the only one which doesn't is distcp;
+ * some tests do have different assertions about directories
+ * found.
+ */
+public class ITestTreewalkProblems extends AbstractS3ACostTest {
+
+ /**
+ * Exit code to expect on a shell failure.
+ */
+ public static final int SHELL_FAILURE = 1;
+
+ /**
+ * Are directory listings potentially inconsistent?
+ */
+ private boolean listingInconsistent;
+
+ @Override
+ public Configuration createConfiguration() {
+ final Configuration conf = super.createConfiguration();
+ removeBaseAndBucketOverrides(conf,
+ DIRECTORY_OPERATIONS_PURGE_UPLOADS,
+ MAGIC_COMMITTER_ENABLED);
+ conf.setBoolean(DIRECTORY_OPERATIONS_PURGE_UPLOADS, true);
+ conf.setBoolean(MAGIC_COMMITTER_ENABLED, true);
+ return conf;
+ }
+
+ @Override
+ public void setup() throws Exception {
+ super.setup();
+ final S3AFileSystem fs = getFileSystem();
+ final Path path = methodPath();
+ assertHasPathCapabilities(fs, path, DIRECTORY_OPERATIONS_PURGE_UPLOADS);
+ listingInconsistent = fs.hasPathCapability(path, DIRECTORY_LISTING_INCONSISTENT);
+ clearAnyUploads(fs, path);
+ }
+
+ @Test
+ public void testListFilesDeep() throws Throwable {
+ final S3AFileSystem fs = getFileSystem();
+ final Path src = createDirWithUpload();
+
+ LOG.info("listFiles({}, true)", src);
+ foreach(fs.listFiles(src, true), e -> LOG.info("{}", e));
+
+ LOG.info("listFiles({}, true)", src);
+ foreach(fs.listLocatedStatus(src), e -> LOG.info("{}", e));
+
+ // and just verify a cleanup works
+ Assertions.assertThat(fs.getS3AInternals().abortMultipartUploads(src))
+ .describedAs("Aborted uploads under %s", src)
+ .isEqualTo(1);
+ assertNoUploadsAt(fs, src);
+ }
+
+ /**
+ * Create a directory methodPath()/src with a magic upload underneath,
+ * with the upload pointing at {@code src/subdir/file.txt}.
+ * @return the directory created
+ * @throws IOException creation problems
+ */
+ private Path createDirWithUpload() throws IOException {
+ final S3AFileSystem fs = getFileSystem();
+ final Path src = new Path(methodPath(), "src");
+ // create a magic file.
+ createMagicFile(fs, src);
+ fs.delete(magicPath(src), true);
+ return src;
+ }
+
+ @Test
+ public void testLocatedFileStatusFetcher() throws Throwable {
+ describe("Validate mapreduce LocatedFileStatusFetcher");
+
+ final Path src = createDirWithUpload();
+
+ Configuration listConfig = new Configuration(getConfiguration());
+ listConfig.setInt(LIST_STATUS_NUM_THREADS, 2);
+
+ LocatedFileStatusFetcher fetcher =
+ new LocatedFileStatusFetcher(listConfig, new Path[]{src}, true, HIDDEN_FILE_FILTER, true);
+ Assertions.assertThat(fetcher.getFileStatuses()).hasSize(0);
+ }
+
+ @Test
+ public void testGetContentSummaryDirsAndFiles() throws Throwable {
+ describe("FileSystem.getContentSummary()");
+ final S3AFileSystem fs = getFileSystem();
+ final Path src = createDirWithUpload();
+ fs.mkdirs(new Path(src, "child"));
+ final Path path = methodPath();
+ file(new Path(path, "file"));
+ final int dirs = listingInconsistent ? 3 : 3;
+ assertContentSummary(path, dirs, 1);
+ }
+
+ /**
+ * Execute getContentSummary() down a directory tree which only
+ * contains a single real directory.
+ * This test case has been a bit inconsistent between different store
+ * types.
+ */
+ @Test
+ public void testGetContentSummaryPendingDir() throws Throwable {
+ describe("FileSystem.getContentSummary() with pending dir");
+ assertContentSummary(createDirWithUpload(), 1, 0);
+ }
+
+ /**
+ * Make an assertions about the content summary of a path.
+ * @param path path to scan
+ * @param dirs number of directories to find.
+ * @param files number of files to find
+ * @throws IOException scanning problems
+ */
+ private void assertContentSummary(
+ final Path path,
+ final int dirs,
+ final int files) throws IOException {
+ ContentSummary summary = getFileSystem().getContentSummary(path);
+ Assertions.assertThat(summary.getDirectoryCount())
+ .describedAs("dir count under %s of %s", path, summary)
+ .isEqualTo(dirs);
+ Assertions.assertThat(summary.getFileCount())
+ .describedAs("filecount count under %s of %s", path, summary)
+ .isEqualTo(files);
+ }
+
+ /**
+ * Execute getContentSummary() down a directory tree which only
+ * contains a single real directory.
+ */
+ @Test
+ public void testGetContentSummaryFiles() throws Throwable {
+ describe("FileSystem.getContentSummary()");
+ final S3AFileSystem fs = getFileSystem();
+ final Path src = createDirWithUpload();
+ fs.mkdirs(new Path(src, "child"));
+ final Path base = methodPath();
+ touch(fs, new Path(base, "file"));
+ assertContentSummary(base, 3, 1);
+ }
+
+ /**
+ * Test all the various filesystem.list* calls.
+ * Bundled into one test case to reduce setup/teardown overhead.
+ */
+ @Test
+ public void testListStatusOperations() throws Throwable {
+ describe("FileSystem liststtus calls");
+ final S3AFileSystem fs = getFileSystem();
+ final Path base = methodPath();
+ final Path src = createDirWithUpload();
+ final Path file = new Path(base, "file");
+ final Path dir2 = new Path(base, "dir2");
+ // path in a child dir
+ final Path childfile = new Path(dir2, "childfile");
+ file(childfile);
+ file(file);
+ fs.mkdirs(dir2);
+
+ Assertions.assertThat(toPathList(fs.listStatusIterator(base)))
+ .describedAs("listStatusIterator(%s)", base)
+ .contains(src, dir2, file);
+
+ Assertions.assertThat(toPathList(remoteIteratorFromArray(fs.listStatus(base))))
+ .describedAs("listStatusIterator(%s, false)", false)
+ .contains(src, dir2, file);
+
+ Assertions.assertThat(toPathList(fs.listFiles(base, false)))
+ .describedAs("listfiles(%s, false)", false)
+ .containsExactly(file);
+
+ Assertions.assertThat(toPathList(fs.listFiles(base, true)))
+ .describedAs("listfiles(%s, true)", false)
+ .containsExactlyInAnyOrder(file, childfile);
+
+ Assertions.assertThat(toPathList(fs.listLocatedStatus(base, (p) -> true)))
+ .describedAs("listLocatedStatus(%s, true)", false)
+ .contains(src, dir2, file);
+ Assertions.assertThat(toPathList(fs.listLocatedStatus(base)))
+ .describedAs("listLocatedStatus(%s, true)", false)
+ .contains(src, dir2, file);
+ }
+
+ @Test
+ public void testShellList() throws Throwable {
+ describe("Validate hadoop fs -ls sorted and unsorted");
+ final S3AFileSystem fs = getFileSystem();
+ final Path base = methodPath();
+ createDirWithUpload();
+ fs.mkdirs(new Path(base, "dir2"));
+ // recursive not sorted
+ shell(base, "-ls", "-R", base.toUri().toString());
+
+ // recursive sorted
+ shell(base, "-ls", "-R", "-S", base.toUri().toString());
+ }
+
+ @Test
+ public void testShellDu() throws Throwable {
+ describe("Validate hadoop fs -du");
+ final Path base = methodPath();
+ createDirWithUpload();
+
+ shell(base, "-du", base.toUri().toString());
+ }
+
+ @Test
+ public void testShellDf() throws Throwable {
+ describe("Validate hadoop fs -df");
+ final Path base = methodPath();
+
+ final String p = base.toUri().toString();
+ shell(SHELL_FAILURE, base, "-df", p);
+ createDirWithUpload();
+
+ shell(base, "-df", p);
+ }
+
+ @Test
+ public void testShellFind() throws Throwable {
+ describe("Validate hadoop fs -ls -R");
+ final Path base = methodPath();
+ final String p = base.toUri().toString();
+ shell(SHELL_FAILURE, base, "-find", p, "-print");
+ createDirWithUpload();
+ shell(base, "-find", p, "-print");
+ }
+
+ @Test
+ public void testDistCp() throws Throwable {
+ describe("Validate distcp");
+ final S3AFileSystem fs = getFileSystem();
+ final Path base = methodPath();
+ final Path src = createDirWithUpload();
+ final Path dest = new Path(base, "dest");
+ file(new Path(src, "real-file"));
+ // distcp fails if uploads are visible
+ DistCpTestUtils.assertRunDistCp(DistCpConstants.SUCCESS, src.toString(), dest.toString(),
+ "-useiterator -update -delete -direct", getConfiguration());
+ }
+
+ @Test
+ public void testDistCpNoIterator() throws Throwable {
+ describe("Validate distcp");
+ final S3AFileSystem fs = getFileSystem();
+ final Path base = methodPath();
+ final Path src = createDirWithUpload();
+ final Path dest = new Path(base, "dest");
+ file(new Path(src, "real-file"));
+
+ // distcp fails if uploads are visible
+ DistCpTestUtils.assertRunDistCp(DistCpConstants.SUCCESS, src.toString(), dest.toString(),
+ "-update -delete -direct", getConfiguration());
+ }
+
+ /**
+ * CTU is also doing treewalking, though it's test only.
+ */
+ @Test
+ public void testContractTestUtilsTreewalk() throws Throwable {
+ final S3AFileSystem fs = getFileSystem();
+ final Path base = methodPath();
+ createDirWithUpload();
+ final ContractTestUtils.TreeScanResults treeWalk = treeWalk(fs, base);
+
+ ContractTestUtils.TreeScanResults listing =
+ new ContractTestUtils.TreeScanResults(fs.listFiles(base, true));
+ treeWalk.assertFieldsEquivalent("treewalk vs listFiles(/, true)", listing, treeWalk.getFiles(),
+ listing.getFiles());
+ }
+
+ /**
+ * Globber is already resilient to missing directories; a relic
+ * of the time when HEAD requests on s3 objects could leave the
+ * 404 in S3 front end cache.
+ */
+ @Test
+ public void testGlobberTreewalk() throws Throwable {
+ final S3AFileSystem fs = getFileSystem();
+ final Path base = methodPath();
+ final Path src = createDirWithUpload();
+ // this is the pending dir
+ final Path subdir = new Path(src, "subdir/");
+ final Path dest = new Path(base, "dest");
+ final Path monday = new Path(dest, "day=monday");
+ final Path realFile = file(new Path(monday, "real-file.parquet"));
+ assertGlob(fs, new Path(base, "*/*/*.parquet"), realFile);
+ if (listingInconsistent) {
+ assertGlob(fs, new Path(base, "*"), src, dest);
+ assertGlob(fs, new Path(base, "*/*"), subdir, monday);
+ } else {
+ assertGlob(fs, new Path(base, "*"), src, dest);
+ assertGlob(fs, new Path(base, "*/*"), monday);
+ }
+ }
+
+ private static void assertGlob(final S3AFileSystem fs,
+ final Path pattern,
+ final Path... expected)
+ throws IOException {
+ final FileStatus[] globbed = fs.globStatus(pattern,
+ (f) -> true);
+ final List<Path> paths = Arrays.stream(globbed).map(s -> s.getPath())
+ .collect(Collectors.toList());
+ Assertions.assertThat(paths)
+ .describedAs("glob(%s)", pattern)
+ .containsExactlyInAnyOrder(expected);
+ }
+
+ @Test
+ public void testFileInputFormatSplits() throws Throwable {
+ final S3AFileSystem fs = getFileSystem();
+ final Path base = methodPath();
+ final Path src = createDirWithUpload();
+ final Path dest = new Path(base, "dest");
+ final Path monday = new Path(dest, "day=monday");
+ final int count = 4;
+ List<Path> files = new ArrayList<>();
+ for (int i = 0; i < count; i++) {
+ files.add(file(new Path(monday, "file-0" + i + ".parquet")));
+ }
+ final JobContextImpl jobContext = new JobContextImpl(getConfiguration(), new JobID("job", 1));
+ final JobConf jc = (JobConf) jobContext.getConfiguration();
+ jc.set("mapreduce.input.fileinputformat.inputdir", base.toUri().toString());
+ jc.setBoolean("mapreduce.input.fileinputformat.input.dir.recursive", true);
+ final TextInputFormat inputFormat = new TextInputFormat();
+ final List<Path> paths = inputFormat.getSplits(jobContext).stream().map(s ->
+ ((FileSplit) s).getPath())
+ .collect(Collectors.toList());
+
+ Assertions.assertThat(paths)
+ .describedAs("input split of base directory")
+ .containsExactlyInAnyOrderElementsOf(files);
+
+
+ }
+
+ /**
+ * Exec a shell command; require it to succeed.
+ * @param base base dir
+ * @param command command sequence
+ * @throws Exception failure
+ */
+
+ private void shell(final Path base, final String... command) throws Exception {
+ shell(0, base, command);
+ }
+
+ /**
+ * Exec a shell command; require the result to match the expected outcome.
+ * @param expected expected outcome
+ * @param base base dir
+ * @param command command sequence
+ * @throws Exception failure
+ */
+
+ private void shell(int expected, final Path base, final String... command) throws Exception {
+ Assertions.assertThat(run(getConfiguration(), new FsShell(), command))
+ .describedAs("%s %s", command[0], base)
+ .isEqualTo(expected);
+ }
+
+ /**
+ * Assert the upload count under a dir is the expected value.
+ * Failure message will include the list of entries.
+ * @param dir dir
+ * @param expected expected count
+ * @throws IOException listing problem
+ */
+ private void assertUploadCount(final Path dir, final int expected) throws IOException {
+ Assertions.assertThat(toList(listUploads(dir)))
+ .describedAs("uploads under %s", dir)
+ .hasSize(expected);
+ }
+
+ /**
+ * List uploads; use the same APIs that the directory operations use,
+ * so implicitly validating them.
+ * @param dir directory to list
+ * @return full list of entries
+ * @throws IOException listing problem
+ */
+ private RemoteIterator<MultipartUpload> listUploads(Path dir) throws IOException {
+ final S3AFileSystem fs = getFileSystem();
+ try (AuditSpan ignored = span()) {
+ final StoreContext sc = fs.createStoreContext();
+ return fs.listUploadsUnderPrefix(sc, sc.pathToKey(dir));
+ }
+ }
+}
diff --git a/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/impl/ITestUploadPurgeOnDirectoryOperations.java b/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/impl/ITestUploadPurgeOnDirectoryOperations.java
index 9e0702737598..80a44e22b8de 100644
--- a/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/impl/ITestUploadPurgeOnDirectoryOperations.java
+++ b/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/impl/ITestUploadPurgeOnDirectoryOperations.java
@@ -19,7 +19,6 @@
package org.apache.hadoop.fs.s3a.impl;
import java.io.IOException;
-import java.nio.charset.StandardCharsets;
import org.assertj.core.api.Assertions;
import org.junit.Test;
@@ -32,21 +31,20 @@ import org.apache.hadoop.fs.s3a.S3AFileSystem;
import org.apache.hadoop.fs.s3a.performance.AbstractS3ACostTest;
import org.apache.hadoop.fs.store.audit.AuditSpan;
-import static org.apache.hadoop.fs.contract.ContractTestUtils.assertFileHasLength;
import static org.apache.hadoop.fs.contract.ContractTestUtils.assertHasPathCapabilities;
-import static org.apache.hadoop.fs.contract.ContractTestUtils.createFile;
import static org.apache.hadoop.fs.s3a.Constants.DIRECTORY_OPERATIONS_PURGE_UPLOADS;
import static org.apache.hadoop.fs.s3a.MultipartTestUtils.clearAnyUploads;
+import static org.apache.hadoop.fs.s3a.MultipartTestUtils.createMagicFile;
import static org.apache.hadoop.fs.s3a.S3ATestUtils.removeBaseAndBucketOverrides;
import static org.apache.hadoop.fs.s3a.Statistic.MULTIPART_UPLOAD_LIST;
import static org.apache.hadoop.fs.s3a.Statistic.OBJECT_MULTIPART_UPLOAD_ABORTED;
import static org.apache.hadoop.fs.s3a.Statistic.OBJECT_MULTIPART_UPLOAD_LIST;
import static org.apache.hadoop.fs.s3a.commit.CommitConstants.MAGIC_COMMITTER_ENABLED;
-import static org.apache.hadoop.fs.s3a.commit.CommitConstants.MAGIC_PATH_PREFIX;
import static org.apache.hadoop.util.functional.RemoteIterators.toList;
/**
* Test behavior of purging uploads in rename and delete.
+ * S3 Express tests automatically set this; it is explicitly set for the rest.
*/
public class ITestUploadPurgeOnDirectoryOperations extends AbstractS3ACostTest {
@@ -117,22 +115,6 @@ public class ITestUploadPurgeOnDirectoryOperations extends AbstractS3ACostTest {
assertUploadCount(dir, 0);
}
- /**
- * Create a magic file of "real" length more than 0 bytes long.
- * @param fs filesystem
- * @param dir directory
- * @return the path
- * @throws IOException creation failure.p
- */
- private static Path createMagicFile(final S3AFileSystem fs, final Path dir) throws IOException {
- Path magicFile = new Path(dir, MAGIC_PATH_PREFIX + "001/file.txt");
- createFile(fs, magicFile, true, "123".getBytes(StandardCharsets.UTF_8));
-
- // the file exists but is a 0 byte marker file.
- assertFileHasLength(fs, magicFile, 0);
- return magicFile;
- }
-
/**
* Assert the upload count under a dir is the expected value.
* Failure message will include the list of entries.
diff --git a/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/impl/TestS3ExpressStorage.java b/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/impl/TestS3ExpressStorage.java
new file mode 100644
index 000000000000..2d5d69d2c90e
--- /dev/null
+++ b/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/impl/TestS3ExpressStorage.java
@@ -0,0 +1,63 @@
+/*
+ * 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.impl;
+
+import org.assertj.core.api.Assertions;
+import org.junit.Test;
+
+import org.apache.hadoop.test.AbstractHadoopTestBase;
+
+import static org.apache.hadoop.fs.s3a.impl.S3ExpressStorage.isS3ExpressStore;
+
+/**
+ * Test S3 Express Storage methods.
+ */
+public class TestS3ExpressStorage extends AbstractHadoopTestBase {
+
+ private static final String S3EXPRESS_BUCKET = "bucket--usw2-az2--x-s3";
+ private static final String AZ = "usw2-az2";
+
+ @Test
+ public void testS3ExpressStateDefaultEndpoint() {
+ assertS3ExpressState(S3EXPRESS_BUCKET, true, "");
+ assertS3ExpressState("bucket", false, "");
+ }
+
+ @Test
+ public void testS3ExpressStateAwsRegions() {
+ assertS3ExpressState(S3EXPRESS_BUCKET, true, "s3.amazonaws.com");
+ assertS3ExpressState(S3EXPRESS_BUCKET, true, "s3.cn-northwest-1.amazonaws.com.cn");
+ assertS3ExpressState(S3EXPRESS_BUCKET, true, "s3-fips.us-gov-east-1.amazonaws.com");
+ assertS3ExpressState(S3EXPRESS_BUCKET, true,
+ "s3-accesspoint-fips.dualstack.us-east-1.amazonaws.com");
+ assertS3ExpressState(S3EXPRESS_BUCKET, true, "s3.ca-central-1.amazonaws.com");
+ }
+
+ @Test
+ public void testS3ExpressStateThirdPartyStore() {
+ assertS3ExpressState(S3EXPRESS_BUCKET, false, "storeendpoint.example.com");
+ }
+
+ private void assertS3ExpressState(final String bucket, final boolean expected, String endpoint) {
+ Assertions.assertThat(isS3ExpressStore(bucket, endpoint))
+ .describedAs("isS3ExpressStore(%s) with endpoint %s", bucket, endpoint)
+ .isEqualTo(expected);
+ }
+
+}
diff --git a/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/performance/ITestDirectoryMarkerListing.java b/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/performance/ITestDirectoryMarkerListing.java
index 8dbf5baaa620..e00130047434 100644
--- a/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/performance/ITestDirectoryMarkerListing.java
+++ b/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/performance/ITestDirectoryMarkerListing.java
@@ -27,6 +27,7 @@ import java.util.List;
import java.util.concurrent.Callable;
import java.util.stream.Collectors;
+import org.assertj.core.api.Assertions;
import software.amazon.awssdk.core.exception.SdkException;
import software.amazon.awssdk.core.sync.RequestBody;
import software.amazon.awssdk.services.s3.S3Client;
@@ -684,7 +685,8 @@ public class ITestDirectoryMarkerListing extends AbstractS3ATestBase {
}
/**
- * Expect the list of status objects to match that of the paths.
+ * Expect the list of status objects to match that of the paths,
+ * without enforcing ordering of the values.
* @param statuses status object list
* @param paths ordered varargs list of paths
* @param <T> type of status objects
@@ -692,20 +694,11 @@ public class ITestDirectoryMarkerListing extends AbstractS3ATestBase {
private <T extends FileStatus> void assertContainsExactlyStatusOfPaths(
List<T> statuses, Path... paths) {
- String actual = statuses.stream()
- .map(Object::toString)
- .collect(Collectors.joining(";"));
- String expected = Arrays.stream(paths)
- .map(Object::toString)
- .collect(Collectors.joining(";"));
- String summary = "expected [" + expected + "]"
- + " actual = [" + actual + "]";
- assertEquals("mismatch in size of listing " + summary,
- paths.length, statuses.size());
- for (int i = 0; i < statuses.size(); i++) {
- assertEquals("Path mismatch at element " + i + " in " + summary,
- paths[i], statuses.get(i).getPath());
- }
+ final List<Path> pathList = statuses.stream()
+ .map(FileStatus::getPath)
+ .collect(Collectors.toList());
+ Assertions.assertThat(pathList)
+ .containsExactlyInAnyOrder(paths);
}
/**
diff --git a/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/s3guard/AbstractS3GuardToolTestBase.java b/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/s3guard/AbstractS3GuardToolTestBase.java
index 95f22644c91a..fd7a528a5d1b 100644
--- a/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/s3guard/AbstractS3GuardToolTestBase.java
+++ b/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/s3guard/AbstractS3GuardToolTestBase.java
@@ -219,8 +219,10 @@ public abstract class AbstractS3GuardToolTestBase extends AbstractS3ATestBase {
cmdR.getName(),
S3A_THIS_BUCKET_DOES_NOT_EXIST
};
- intercept(UnknownStoreException.class,
- () -> cmdR.run(argsR));
+ intercept(UnknownStoreException.class, () -> {
+ final int e = cmdR.run(argsR);
+ return String.format("Outcome of %s on missing bucket: %d", tool, e);
+ });
}
}
diff --git a/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/scale/AbstractSTestS3AHugeFiles.java b/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/scale/AbstractSTestS3AHugeFiles.java
index 628b56704182..53d6a4e0f99c 100644
--- a/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/scale/AbstractSTestS3AHugeFiles.java
+++ b/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/scale/AbstractSTestS3AHugeFiles.java
@@ -18,10 +18,12 @@
package org.apache.hadoop.fs.s3a.scale;
+import java.io.FileNotFoundException;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.List;
+import java.util.NoSuchElementException;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
import java.util.function.IntFunction;
@@ -72,6 +74,7 @@ import static org.apache.hadoop.fs.statistics.IOStatisticAssertions.verifyStatis
import static org.apache.hadoop.fs.statistics.IOStatisticsLogging.ioStatisticsSourceToString;
import static org.apache.hadoop.fs.statistics.IOStatisticsLogging.ioStatisticsToPrettyString;
import static org.apache.hadoop.fs.statistics.StreamStatisticNames.STREAM_WRITE_BLOCK_UPLOADS;
+import static org.apache.hadoop.util.functional.RemoteIterators.filteringRemoteIterator;
/**
* Scale test which creates a huge file.
@@ -103,8 +106,8 @@ public abstract class AbstractSTestS3AHugeFiles extends S3AScaleTestBase {
public void setup() throws Exception {
super.setup();
scaleTestDir = new Path(getTestPath(), getTestSuiteName());
- hugefile = new Path(scaleTestDir, "hugefile");
- hugefileRenamed = new Path(scaleTestDir, "hugefileRenamed");
+ hugefile = new Path(scaleTestDir, "src/hugefile");
+ hugefileRenamed = new Path(scaleTestDir, "dest/hugefile");
uploadBlockSize = uploadBlockSize();
filesize = getTestPropertyBytes(getConf(), KEY_HUGE_FILESIZE,
DEFAULT_HUGE_FILESIZE);
@@ -490,7 +493,40 @@ public abstract class AbstractSTestS3AHugeFiles extends S3AScaleTestBase {
ContractTestUtils.assertPathExists(fs, "Huge file", hugefile);
FileStatus status = fs.getFileStatus(hugefile);
ContractTestUtils.assertIsFile(hugefile, status);
+ LOG.info("Huge File Status: {}", status);
assertEquals("File size in " + status, filesize, status.getLen());
+
+ // now do some etag status checks asserting they are always the same
+ // across listing operations.
+ final Path path = hugefile;
+ final FileStatus listStatus = listFile(hugefile);
+ LOG.info("List File Status: {}", listStatus);
+
+ Assertions.assertThat(listStatus.getLen())
+ .describedAs("List file status length %s", listStatus)
+ .isEqualTo(filesize);
+ Assertions.assertThat(etag(listStatus))
+ .describedAs("List file status etag %s", listStatus)
+ .isEqualTo(etag(status));
+ }
+
+ /**
+ * Get a filestatus by listing the parent directory.
+ * @param path path
+ * @return status
+ * @throws IOException failure to read, file not found
+ */
+ private FileStatus listFile(final Path path)
+ throws IOException {
+ try {
+ return filteringRemoteIterator(
+ getFileSystem().listStatusIterator(path.getParent()),
+ st -> st.getPath().equals(path))
+ .next();
+ } catch (NoSuchElementException e) {
+ throw (FileNotFoundException)(new FileNotFoundException("Not found: " + path)
+ .initCause(e));
+ }
}
/**
@@ -509,7 +545,7 @@ public abstract class AbstractSTestS3AHugeFiles extends S3AScaleTestBase {
String filetype = encrypted ? "encrypted file" : "file";
describe("Positioned reads of %s %s", filetype, hugefile);
S3AFileSystem fs = getFileSystem();
- FileStatus status = fs.getFileStatus(hugefile);
+ FileStatus status = listFile(hugefile);
long size = status.getLen();
int ops = 0;
final int bufferSize = 8192;
@@ -518,7 +554,11 @@ public abstract class AbstractSTestS3AHugeFiles extends S3AScaleTestBase {
ContractTestUtils.NanoTimer timer = new ContractTestUtils.NanoTimer();
ContractTestUtils.NanoTimer readAtByte0, readAtByte0Again, readAtEOF;
- try (FSDataInputStream in = fs.open(hugefile, uploadBlockSize)) {
+ try (FSDataInputStream in = fs.openFile(hugefile)
+ .withFileStatus(status)
+ .opt(FS_OPTION_OPENFILE_READ_POLICY, "random")
+ .opt(FS_OPTION_OPENFILE_BUFFER_SIZE, uploadBlockSize)
+ .build().get()) {
readAtByte0 = new ContractTestUtils.NanoTimer();
in.readFully(0, buffer);
readAtByte0.end("time to read data at start of file");
@@ -661,27 +701,45 @@ public abstract class AbstractSTestS3AHugeFiles extends S3AScaleTestBase {
S3AFileSystem fs = getFileSystem();
FileStatus status = fs.getFileStatus(hugefile);
long size = status.getLen();
- fs.delete(hugefileRenamed, false);
ContractTestUtils.NanoTimer timer = new ContractTestUtils.NanoTimer();
- fs.rename(hugefile, hugefileRenamed);
+ renameFile(hugefile, hugefileRenamed);
long mb = Math.max(size / _1MB, 1);
timer.end("time to rename file of %d MB", mb);
LOG.info("Time per MB to rename = {} nS",
toHuman(timer.nanosPerOperation(mb)));
bandwidth(timer, size);
+ assertPathExists("renamed file", hugefileRenamed);
logFSState();
FileStatus destFileStatus = fs.getFileStatus(hugefileRenamed);
assertEquals(size, destFileStatus.getLen());
// rename back
ContractTestUtils.NanoTimer timer2 = new ContractTestUtils.NanoTimer();
- fs.rename(hugefileRenamed, hugefile);
+ renameFile(hugefileRenamed, hugefile);
+
timer2.end("Renaming back");
LOG.info("Time per MB to rename = {} nS",
toHuman(timer2.nanosPerOperation(mb)));
bandwidth(timer2, size);
}
+ /**
+ * Rename a file.
+ * Subclasses may do this differently.
+ * @param src source file
+ * @param dest dest file
+ * @throws IOException IO failure
+ */
+ protected void renameFile(final Path src,
+ final Path dest) throws IOException {
+ final S3AFileSystem fs = getFileSystem();
+ fs.delete(dest, false);
+ final boolean renamed = fs.rename(src, dest);
+ Assertions.assertThat(renamed)
+ .describedAs("rename(%s, %s)", src, dest)
+ .isTrue();
+ }
+
/**
* Test to verify target file encryption key.
* @throws IOException
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 b1323c49a8f6..1e74d715b88f 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
@@ -18,12 +18,20 @@
package org.apache.hadoop.fs.s3a.scale;
+import java.io.IOException;
+
+import org.assertj.core.api.Assertions;
+
+import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.s3a.Constants;
+import org.apache.hadoop.fs.s3a.S3AFileSystem;
import static org.apache.hadoop.fs.s3a.Constants.FAST_UPLOAD_BYTEBUFFER;
/**
* Use {@link Constants#FAST_UPLOAD_BYTEBUFFER} for buffering.
+ * This also renames by parent directory, so validates parent
+ * dir renaming of huge files.
*/
public class ITestS3AHugeFilesByteBufferBlocks
extends AbstractSTestS3AHugeFiles {
@@ -31,4 +39,25 @@ public class ITestS3AHugeFilesByteBufferBlocks
protected String getBlockOutputBufferName() {
return FAST_UPLOAD_BYTEBUFFER;
}
+
+ /**
+ * Rename the parent directory, rather than the file itself.
+ * @param src source file
+ * @param dest dest file
+ * @throws IOException
+ */
+ @Override
+ protected void renameFile(final Path src, final Path dest) throws IOException {
+
+ final S3AFileSystem fs = getFileSystem();
+
+ final Path srcDir = src.getParent();
+ final Path destDir = dest.getParent();
+ fs.delete(destDir, true);
+ fs.mkdirs(destDir, null);
+ final boolean renamed = fs.rename(srcDir, destDir);
+ Assertions.assertThat(renamed)
+ .describedAs("rename(%s, %s)", src, dest)
+ .isTrue();
+ }
}
diff --git a/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/tools/ITestBucketTool.java b/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/tools/ITestBucketTool.java
new file mode 100644
index 000000000000..3025646d239b
--- /dev/null
+++ b/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/tools/ITestBucketTool.java
@@ -0,0 +1,183 @@
+/*
+ * 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.tools;
+
+import java.io.IOException;
+import java.net.UnknownHostException;
+
+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.AWSBadRequestException;
+import org.apache.hadoop.fs.s3a.AbstractS3ATestBase;
+import org.apache.hadoop.fs.s3a.S3AFileSystem;
+import org.apache.hadoop.fs.s3a.S3ATestUtils;
+import org.apache.hadoop.util.ExitUtil;
+
+import static org.apache.hadoop.fs.s3a.Constants.AWS_REGION;
+import static org.apache.hadoop.fs.s3a.S3ATestUtils.assumeNotS3ExpressFileSystem;
+import static org.apache.hadoop.fs.s3a.S3ATestUtils.assumeS3ExpressFileSystem;
+import static org.apache.hadoop.fs.s3a.S3ATestUtils.assumeStoreAwsHosted;
+import static org.apache.hadoop.fs.s3a.S3ATestUtils.expectErrorCode;
+import static org.apache.hadoop.fs.s3a.impl.S3ExpressStorage.STORE_CAPABILITY_S3_EXPRESS_STORAGE;
+import static org.apache.hadoop.fs.s3a.tools.BucketTool.CREATE;
+import static org.apache.hadoop.fs.s3a.tools.BucketTool.NO_ZONE_SUPPLIED;
+import static org.apache.hadoop.fs.s3a.tools.BucketTool.OPT_ENDPOINT;
+import static org.apache.hadoop.fs.s3a.tools.BucketTool.OPT_REGION;
+import static org.apache.hadoop.fs.s3a.tools.BucketTool.OPT_ZONE;
+import static org.apache.hadoop.fs.s3a.tools.BucketTool.UNSUPPORTED_ZONE_ARG;
+import static org.apache.hadoop.service.launcher.LauncherExitCodes.EXIT_NOT_ACCEPTABLE;
+import static org.apache.hadoop.service.launcher.LauncherExitCodes.EXIT_USAGE;
+import static org.apache.hadoop.test.GenericTestUtils.assertExceptionContains;
+import static org.apache.hadoop.test.LambdaTestUtils.intercept;
+
+/**
+ * Test {@link BucketTool}.
+ * This is a tricky little test as it doesn't want to create any new bucket,
+ * including on third party stores.
+ */
+public class ITestBucketTool extends AbstractS3ATestBase {
+
+ public static final String USWEST_2 = "us-west-2";
+
+ private static final Logger LOG = LoggerFactory.getLogger(ITestBucketTool.class);
+
+ /**
+ * This is yours.
+ */
+ public static final String OWNED = "BucketAlreadyOwnedByYouException";
+
+ /**
+ * This is not yours. Not *yet* used in any tests.
+ */
+ public static final String EXISTS = "BucketAlreadyExists";
+
+ /**
+ * Sample S3 Express bucket name. {@value}.
+ */
+ private static final String S3EXPRESS_BUCKET = "bucket--usw2-az2--x-s3";
+
+ /**
+ * Sample S3 Express bucket URI. {@value}.
+ */
+ public static final String S3A_S3EXPRESS = "s3a://" + S3EXPRESS_BUCKET;
+
+ /**
+ * US west az. {@value}
+ */
+ public static final String USWEST_AZ_2 = "usw2-az2";
+
+ private String bucketName;
+
+ private boolean s3ExpressStore;
+
+ private BucketTool bucketTool;
+
+ private String fsURI;
+
+ private String region;
+
+ private S3AFileSystem fs;
+
+ @Override
+ public void setup() throws Exception {
+ super.setup();
+ fs = getFileSystem();
+ final Configuration conf = fs.getConf();
+ bucketName = S3ATestUtils.getTestBucketName(conf);
+ fsURI = fs.getUri().toString();
+ s3ExpressStore = fs.hasPathCapability(new Path("/"), STORE_CAPABILITY_S3_EXPRESS_STORAGE);
+ bucketTool = new BucketTool(conf);
+ region = conf.get(AWS_REGION, "");
+ }
+
+ @Test
+ public void testRecreateTestBucketS3Express() throws Throwable {
+ assumeS3ExpressFileSystem(getFileSystem());
+ final IOException ex = intercept(IOException.class,
+ () -> bucketTool.exec("bucket", d(CREATE), d(OPT_ZONE), USWEST_AZ_2, d(OPT_REGION), region,
+ fsURI));
+ if (ex instanceof AWSBadRequestException) {
+ // owned error
+ assertExceptionContains(OWNED, ex);
+ } else if (ex instanceof UnknownHostException) {
+ // endpoint region stuff, expect the error to include the s3express endpoint
+ // name
+ assertExceptionContains("s3express-control", ex);
+ }
+ }
+
+ /**
+ * Try and recreate the test bucket.
+ * Third party stores may allow this, so skip test on them.
+ */
+ @Test
+ public void testRecreateTestBucketNonS3Express() throws Throwable {
+ assumeNotS3ExpressFileSystem(fs);
+ assumeStoreAwsHosted(fs);
+ intercept(AWSBadRequestException.class, OWNED,
+ () -> bucketTool.exec("bucket", d(CREATE),
+ d(OPT_REGION), region,
+ fsURI));
+ }
+
+ @Test
+ public void testSimpleBucketWithZoneParam() throws Throwable {
+ expectErrorCode(EXIT_USAGE,
+ intercept(ExitUtil.ExitException.class, UNSUPPORTED_ZONE_ARG, () ->
+ bucketTool.exec("bucket", d(CREATE),
+ d(OPT_ZONE), USWEST_AZ_2,
+ "s3a://simple/")));
+ }
+
+ @Test
+ public void testS3ExpressBucketWithoutZoneParam() throws Throwable {
+ expectErrorCode(EXIT_USAGE,
+ intercept(ExitUtil.ExitException.class, NO_ZONE_SUPPLIED, () ->
+ bucketTool.exec("bucket", d(CREATE),
+ S3A_S3EXPRESS)));
+ }
+
+ /**
+ * To avoid accidentally trying to create S3 Express stores on third party
+ * endpoints, the tool rejects such attempts.
+ */
+ @Test
+ public void testS3ExpressStorageOnlyOnAwsEndpoint() throws Throwable {
+ describe("The tool only supports S3 Express bucket names on aws endpoints");
+ expectErrorCode(EXIT_NOT_ACCEPTABLE,
+ intercept(ExitUtil.ExitException.class, () ->
+ bucketTool.exec("bucket", d(CREATE),
+ d(OPT_ENDPOINT), "s3.example.org",
+ S3A_S3EXPRESS)));
+ }
+
+ /**
+ * Add a dash to a string.
+ * @param s source.
+ * @return "-s"
+ */
+ private String d(String s) {
+ return "-" + s;
+ }
+
+}
diff --git a/hadoop-tools/hadoop-aws/src/test/resources/core-site.xml b/hadoop-tools/hadoop-aws/src/test/resources/core-site.xml
index e6c3eeed3d7e..f871369ed571 100644
--- a/hadoop-tools/hadoop-aws/src/test/resources/core-site.xml
+++ b/hadoop-tools/hadoop-aws/src/test/resources/core-site.xml
@@ -108,7 +108,7 @@
<property>
<name>ireland.endpoint</name>
- <value>s3-eu-west-1.amazonaws.com</value>
+ <value>s3.eu-west-1.amazonaws.com</value>
</property>
<property>
diff --git a/hadoop-tools/hadoop-aws/src/test/resources/log4j.properties b/hadoop-tools/hadoop-aws/src/test/resources/log4j.properties
index 15c3b571b126..25247aaaabd3 100644
--- a/hadoop-tools/hadoop-aws/src/test/resources/log4j.properties
+++ b/hadoop-tools/hadoop-aws/src/test/resources/log4j.properties
@@ -92,3 +92,7 @@ log4j.logger.org.apache.hadoop.fs.s3a.S3AStorageStatistics=INFO
# uncomment this to trace where context entries are set
# log4j.logger.org.apache.hadoop.fs.audit.CommonAuditContext=TRACE
+
+# uncomment this to get S3 Delete requests to return the list of deleted objects
+# log4.logger.org.apache.hadoop.fs.s3a.impl.RequestFactoryImpl=TRACE
+
diff --git a/hadoop-tools/hadoop-distcp/src/test/java/org/apache/hadoop/tools/util/DistCpTestUtils.java b/hadoop-tools/hadoop-distcp/src/test/java/org/apache/hadoop/tools/util/DistCpTestUtils.java
index 624f7d5a0ef7..15211e261ee7 100644
--- a/hadoop-tools/hadoop-distcp/src/test/java/org/apache/hadoop/tools/util/DistCpTestUtils.java
+++ b/hadoop-tools/hadoop-distcp/src/test/java/org/apache/hadoop/tools/util/DistCpTestUtils.java
@@ -24,9 +24,13 @@ import org.apache.hadoop.fs.Path;
import org.apache.hadoop.tools.DistCp;
import org.apache.hadoop.util.ToolRunner;
+import java.util.Arrays;
import java.util.Iterator;
import java.util.Map;
import java.util.Map.Entry;
+import java.util.stream.Collectors;
+
+import org.assertj.core.api.Assertions;
import static org.junit.Assert.assertArrayEquals;
import static org.junit.Assert.assertEquals;
@@ -91,7 +95,9 @@ public class DistCpTestUtils {
optsArr[optsArr.length - 2] = src;
optsArr[optsArr.length - 1] = dst;
- assertEquals(exitCode,
- ToolRunner.run(conf, distCp, optsArr));
+ Assertions.assertThat(ToolRunner.run(conf, distCp, optsArr))
+ .describedAs("Exit code of distcp %s",
+ Arrays.stream(optsArr).collect(Collectors.joining(" ")))
+ .isEqualTo(exitCode);
}
}
---------------------------------------------------------------------
To unsubscribe, e-mail: common-commits-unsubscribe@hadoop.apache.org
For additional commands, e-mail: common-commits-help@hadoop.apache.org
(hadoop) 02/05: HADOOP-18915. Tune/extend S3A http connection and thread pool settings (#6180)
Posted by ah...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
ahmar pushed a commit to branch feature-HADOOP-18073-sdk-v2-upgrade-3.4
in repository https://gitbox.apache.org/repos/asf/hadoop.git
commit 6f69a2fb1851aae5ad1d150abca34d82a132079b
Author: Steve Loughran <st...@cloudera.com>
AuthorDate: Wed Nov 29 15:12:44 2023 +0000
HADOOP-18915. Tune/extend S3A http connection and thread pool settings (#6180)
Increases existing pool sizes, as with server scale and vector
IO, larger pools are needed
fs.s3a.connection.maximum 200
fs.s3a.threads.max 96
Adds new configuration options for v2 sdk internal timeouts,
both with default of 60s:
fs.s3a.connection.acquisition.timeout
fs.s3a.connection.idle.time
All the pool/timoeut options are covered in performance.md
Moves all timeout/duration options in the s3a FS to taking
temporal units (h, m, s, ms,...); retaining the previous default
unit (normally millisecond)
Adds a minimum duration for most of these, in order to recover from
deployments where a timeout has been set on the assumption the unit
was seconds, not millis.
Uses java.time.Duration throughout the codebase;
retaining the older numeric constants in
org.apache.hadoop.fs.s3a.Constants for backwards compatibility;
these are now deprecated.
Adds new class AWSApiCallTimeoutException to be raised on
sdk-related methods and also gateway timeouts. This is a subclass
of org.apache.hadoop.net.ConnectTimeoutException to support
existing retry logic.
+ reverted default value of fs.s3a.create.performance to false;
inadvertently set to true during testing.
Contributed by Steve Loughran.
---
.../src/main/resources/core-default.xml | 28 +-
.../hadoop/fs/s3a/AWSApiCallTimeoutException.java | 41 +++
.../java/org/apache/hadoop/fs/s3a/Constants.java | 191 +++++++++--
.../org/apache/hadoop/fs/s3a/S3AFileSystem.java | 20 +-
.../org/apache/hadoop/fs/s3a/S3ARetryPolicy.java | 3 +
.../java/org/apache/hadoop/fs/s3a/S3AUtils.java | 23 +-
.../apache/hadoop/fs/s3a/impl/AWSClientConfig.java | 351 +++++++++++++++++----
.../hadoop/fs/s3a/impl/ConfigurationHelper.java | 132 ++++++++
.../hadoop/fs/s3a/impl/InternalConstants.java | 3 +
.../src/site/markdown/tools/hadoop-aws/index.md | 103 ++----
.../site/markdown/tools/hadoop-aws/performance.md | 111 +++++--
.../hadoop/fs/s3a/TestS3AExceptionTranslation.java | 53 +++-
.../fs/s3a/impl/ITestConnectionTimeouts.java | 149 +++++++++
.../hadoop/fs/s3a/impl/TestAwsClientConfig.java | 188 +++++++++++
.../fs/s3a/performance/ITestUnbufferDraining.java | 30 +-
15 files changed, 1197 insertions(+), 229 deletions(-)
diff --git a/hadoop-common-project/hadoop-common/src/main/resources/core-default.xml b/hadoop-common-project/hadoop-common/src/main/resources/core-default.xml
index 1a391c9cf22f..b9b6ec3d8f20 100644
--- a/hadoop-common-project/hadoop-common/src/main/resources/core-default.xml
+++ b/hadoop-common-project/hadoop-common/src/main/resources/core-default.xml
@@ -1332,7 +1332,7 @@
<property>
<name>fs.s3a.connection.maximum</name>
- <value>96</value>
+ <value>200</value>
<description>Controls the maximum number of simultaneous connections to S3.
This must be bigger than the value of fs.s3a.threads.max so as to stop
threads being blocked waiting for new HTTPS connections.
@@ -1410,14 +1410,21 @@
<property>
<name>fs.s3a.connection.establish.timeout</name>
- <value>5000</value>
- <description>Socket connection setup timeout in milliseconds.</description>
+ <value>5s</value>
+ <description>Socket connection setup timeout in milliseconds; this will be retried
+ more than once.</description>
</property>
<property>
<name>fs.s3a.connection.timeout</name>
- <value>200000</value>
- <description>Socket connection timeout in milliseconds.</description>
+ <value>200s</value>
+ <description>Socket connection timeout.</description>
+</property>
+
+<property>
+ <name>fs.s3a.connection.ttl</name>
+ <value>5m</value>
+ <description>Expiry time for any active connection.</description>
</property>
<property>
@@ -1441,14 +1448,14 @@
<property>
<name>fs.s3a.threads.max</name>
- <value>64</value>
+ <value>96</value>
<description>The total number of threads available in the filesystem for data
uploads *or any other queued filesystem operation*.</description>
</property>
<property>
<name>fs.s3a.threads.keepalivetime</name>
- <value>60</value>
+ <value>60s</value>
<description>Number of seconds a thread can be idle before being
terminated.</description>
</property>
@@ -1528,7 +1535,7 @@
<property>
<name>fs.s3a.multipart.purge.age</name>
- <value>86400</value>
+ <value>24h</value>
<description>Minimum age in seconds of multipart uploads to purge
on startup if "fs.s3a.multipart.purge" is true
</description>
@@ -1893,18 +1900,15 @@
<property>
<name>fs.s3a.connection.request.timeout</name>
- <value>0</value>
+ <value>0s</value>
<description>
Time out on HTTP requests to the AWS service; 0 means no timeout.
- Measured in seconds; the usual time suffixes are all supported
Important: this is the maximum duration of any AWS service call,
including upload and copy operations. If non-zero, it must be larger
than the time to upload multi-megabyte blocks to S3 from the client,
and to rename many-GB files. Use with care.
- Values that are larger than Integer.MAX_VALUE milliseconds are
- converged to Integer.MAX_VALUE milliseconds
</description>
</property>
diff --git a/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/AWSApiCallTimeoutException.java b/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/AWSApiCallTimeoutException.java
new file mode 100644
index 000000000000..c5a72f018521
--- /dev/null
+++ b/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/AWSApiCallTimeoutException.java
@@ -0,0 +1,41 @@
+/*
+ * 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 org.apache.hadoop.net.ConnectTimeoutException;
+
+/**
+ * IOException equivalent of an {@code ApiCallTimeoutException}.
+ * Declared as a subclass of {@link ConnectTimeoutException} to allow
+ * for existing code to catch it.
+ */
+public class AWSApiCallTimeoutException extends ConnectTimeoutException {
+
+ /**
+ * Constructor.
+ * @param operation operation in progress
+ * @param cause cause.
+ */
+ public AWSApiCallTimeoutException(
+ final String operation,
+ final Exception cause) {
+ super(operation);
+ initCause(cause);
+ }
+}
diff --git a/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/Constants.java b/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/Constants.java
index 16012871ff48..e602c1b3dc36 100644
--- a/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/Constants.java
+++ b/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/Constants.java
@@ -23,6 +23,7 @@ import org.apache.hadoop.classification.InterfaceStability;
import org.apache.hadoop.fs.Options;
import org.apache.hadoop.security.ssl.DelegatingSSLSocketFactory;
+import java.time.Duration;
import java.util.concurrent.TimeUnit;
/**
@@ -33,7 +34,14 @@ import java.util.concurrent.TimeUnit;
* as deprecated and simply ignored.
*
* All S3Guard related constants are marked as Deprecated and either ignored (ddb config)
- * or rejected (setting the metastore to anything other than the null store)
+ * or rejected (setting the metastore to anything other than the null store).
+ * <p>
+ * Timeout default values are declared as integers or long values in milliseconds and
+ * occasionally seconds.
+ * There are now {@code Duration} constants for these default values; the original
+ * fields are retained for compatibility, and derive their value from the Duration equivalent.
+ * <p>
+ * New timeout/duration constants do not get the equivalent integer/long fields.
*/
@InterfaceAudience.Public
@InterfaceStability.Evolving
@@ -144,25 +152,47 @@ public final class Constants {
SimpleAWSCredentialsProvider.NAME;
- // the maximum number of tasks cached if all threads are already uploading
+ /**
+ * The maximum number of tasks queued (other than prefetcher tasks) if all threads are
+ * busy: {@value}.
+ */
public static final String MAX_TOTAL_TASKS = "fs.s3a.max.total.tasks";
+ /**
+ * Default value for {@link #MAX_TOTAL_TASKS}: {@value}.
+ */
public static final int DEFAULT_MAX_TOTAL_TASKS = 32;
- // number of simultaneous connections to s3
+ /**
+ * Number of simultaneous connections to S3: {@value}.
+ */
public static final String MAXIMUM_CONNECTIONS = "fs.s3a.connection.maximum";
- public static final int DEFAULT_MAXIMUM_CONNECTIONS = 96;
+
+ /**
+ * Default value for {@link #MAXIMUM_CONNECTIONS}: {@value}.
+ * Future releases are likely to increase this value.
+ * Keep in sync with the value in {@code core-default.xml}
+ */
+ public static final int DEFAULT_MAXIMUM_CONNECTIONS = 200;
/**
* Configuration option to configure expiration time of
- * s3 http connection from the connection pool in milliseconds: {@value}.
+ * S3 http connection from the connection pool: {@value}.
*/
public static final String CONNECTION_TTL = "fs.s3a.connection.ttl";
/**
- * Default value for {@code CONNECTION_TTL}: {@value}.
+ * Default duration for {@link #CONNECTION_TTL}: 5 minutes.
+ */
+ public static final Duration DEFAULT_CONNECTION_TTL_DURATION =
+ Duration.ofMinutes(5);
+
+ /**
+ * Default value in millis for {@link #CONNECTION_TTL}: 5 minutes.
+ * @deprecated use {@link #DEFAULT_CONNECTION_TTL_DURATION}
*/
- public static final long DEFAULT_CONNECTION_TTL = 5 * 60_000;
+ public static final long DEFAULT_CONNECTION_TTL =
+ DEFAULT_CONNECTION_TTL_DURATION.toMillis();
// connect to s3 over ssl?
public static final String SECURE_CONNECTIONS =
@@ -264,19 +294,111 @@ public final class Constants {
public static final boolean EXPERIMENTAL_AWS_INTERNAL_THROTTLING_DEFAULT =
true;
- // milliseconds until we give up trying to establish a connection to s3
+ /**
+ * This is the minimum operation duration unless programmatically set.
+ * It ensures that even if a configuration has mistaken a millisecond
+ * option for seconds, a viable duration will actually be used.
+ * Value: 15s.
+ */
+ public static final Duration MINIMUM_NETWORK_OPERATION_DURATION = Duration.ofSeconds(15);
+
+ /**
+ * Milliseconds until a connection is established: {@value}.
+ */
public static final String ESTABLISH_TIMEOUT =
"fs.s3a.connection.establish.timeout";
- public static final int DEFAULT_ESTABLISH_TIMEOUT = 5000;
- // milliseconds until we give up on a connection to s3
+ /**
+ * Default TCP/(and TLS?) establish timeout: 30 seconds.
+ */
+ public static final Duration DEFAULT_ESTABLISH_TIMEOUT_DURATION = Duration.ofSeconds(30);
+
+ /**
+ * Default establish timeout in millis: 30 seconds.
+ * @deprecated use {@link #DEFAULT_ESTABLISH_TIMEOUT_DURATION}
+ */
+ public static final int DEFAULT_ESTABLISH_TIMEOUT =
+ (int)DEFAULT_ESTABLISH_TIMEOUT_DURATION.toMillis();
+
+ /**
+ * Milliseconds until we give up on a connection to s3: {@value}.
+ */
public static final String SOCKET_TIMEOUT = "fs.s3a.connection.timeout";
- public static final int DEFAULT_SOCKET_TIMEOUT = 200000;
- // milliseconds until a request is timed-out
+ /**
+ * Default socket timeout: 200 seconds.
+ */
+ public static final Duration DEFAULT_SOCKET_TIMEOUT_DURATION = Duration.ofSeconds(200);
+
+ /**
+ * Default socket timeout: {@link #DEFAULT_SOCKET_TIMEOUT_DURATION}.
+ * @deprecated use {@link #DEFAULT_SOCKET_TIMEOUT_DURATION}
+ */
+ public static final int DEFAULT_SOCKET_TIMEOUT = (int)DEFAULT_SOCKET_TIMEOUT_DURATION.toMillis();
+
+ /**
+ * Time until a request is timed-out: {@value}.
+ * If zero, there is no timeout.
+ */
public static final String REQUEST_TIMEOUT =
"fs.s3a.connection.request.timeout";
- public static final int DEFAULT_REQUEST_TIMEOUT = 0;
+
+ /**
+ * Default duration of a request before it is timed out: Zero.
+ */
+ public static final Duration DEFAULT_REQUEST_TIMEOUT_DURATION = Duration.ZERO;
+
+ /**
+ * Default duration of a request before it is timed out: Zero.
+ * @deprecated use {@link #DEFAULT_REQUEST_TIMEOUT_DURATION}
+ */
+ public static final int DEFAULT_REQUEST_TIMEOUT =
+ (int)DEFAULT_REQUEST_TIMEOUT_DURATION.toMillis();
+
+ /**
+ * Acquisition timeout for connections from the pool:
+ * {@value}.
+ * Default unit is milliseconds for consistency with other options.
+ */
+ public static final String CONNECTION_ACQUISITION_TIMEOUT =
+ "fs.s3a.connection.acquisition.timeout";
+
+ /**
+ * Default acquisition timeout: 60 seconds.
+ */
+ public static final Duration DEFAULT_CONNECTION_ACQUISITION_TIMEOUT_DURATION =
+ Duration.ofSeconds(60);
+
+ /**
+ * Should TCP Keepalive be enabled on the socket?
+ * This adds some network IO, but finds failures faster.
+ * {@value}.
+ */
+ public static final String CONNECTION_KEEPALIVE =
+ "fs.s3a.connection.keepalive";
+
+ /**
+ * Default value of {@link #CONNECTION_KEEPALIVE}: {@value}.
+ */
+ public static final boolean DEFAULT_CONNECTION_KEEPALIVE = false;
+
+ /**
+ * Maximum idle time for connections in the pool: {@value}.
+ * <p>
+ * Too low: overhead of creating connections.
+ * Too high, risk of stale connections and inability to use the
+ * adaptive load balancing of the S3 front end.
+ * <p>
+ * Default unit is milliseconds for consistency with other options.
+ */
+ public static final String CONNECTION_IDLE_TIME =
+ "fs.s3a.connection.idle.time";
+
+ /**
+ * Default idle time: 60 seconds.
+ */
+ public static final Duration DEFAULT_CONNECTION_IDLE_TIME_DURATION =
+ Duration.ofSeconds(60);
// socket send buffer to be used in Amazon client
public static final String SOCKET_SEND_BUFFER = "fs.s3a.socket.send.buffer";
@@ -290,13 +412,34 @@ public final class Constants {
public static final String MAX_PAGING_KEYS = "fs.s3a.paging.maximum";
public static final int DEFAULT_MAX_PAGING_KEYS = 5000;
- // the maximum number of threads to allow in the pool used by TransferManager
+ /**
+ * The maximum number of threads to allow in the pool used by S3A.
+ * Value: {@value}.
+ */
public static final String MAX_THREADS = "fs.s3a.threads.max";
- public static final int DEFAULT_MAX_THREADS = 10;
- // the time an idle thread waits before terminating
+ /**
+ * Default value of {@link #MAX_THREADS}: {@value}.
+ */
+ public static final int DEFAULT_MAX_THREADS = 96;
+
+ /**
+ * The time an idle thread waits before terminating: {@value}.
+ * This is in SECONDS unless the optional unit is given.
+ */
public static final String KEEPALIVE_TIME = "fs.s3a.threads.keepalivetime";
- public static final int DEFAULT_KEEPALIVE_TIME = 60;
+
+ /**
+ * Default value of {@link #KEEPALIVE_TIME}: 60s.
+ */
+ public static final Duration DEFAULT_KEEPALIVE_TIME_DURATION = Duration.ofSeconds(60);
+
+ /**
+ * Default value of {@link #KEEPALIVE_TIME}: 60s.
+ * @deprecated use {@link #DEFAULT_KEEPALIVE_TIME_DURATION}
+ */
+ public static final int DEFAULT_KEEPALIVE_TIME =
+ (int)DEFAULT_KEEPALIVE_TIME_DURATION.getSeconds();
// size of each of or multipart pieces in bytes
public static final String MULTIPART_SIZE = "fs.s3a.multipart.size";
@@ -501,10 +644,17 @@ public final class Constants {
"fs.s3a.multipart.purge";
public static final boolean DEFAULT_PURGE_EXISTING_MULTIPART = false;
- // purge any multipart uploads older than this number of seconds
+ /**
+ * purge any multipart uploads older than this number of seconds.
+ */
public static final String PURGE_EXISTING_MULTIPART_AGE =
"fs.s3a.multipart.purge.age";
- public static final long DEFAULT_PURGE_EXISTING_MULTIPART_AGE = 86400;
+
+ /**
+ * Default Age.
+ */
+ public static final long DEFAULT_PURGE_EXISTING_MULTIPART_AGE =
+ Duration.ofDays(1).getSeconds();
/**
* s3 server-side encryption, see
@@ -1201,8 +1351,7 @@ public final class Constants {
* Default value for create performance in an S3A FS.
* Value {@value}.
*/
- public static final boolean FS_S3A_CREATE_PERFORMANCE_DEFAULT = true;
-
+ public static final boolean FS_S3A_CREATE_PERFORMANCE_DEFAULT = false;
/**
* Capability to indicate that the FS has been instantiated with
diff --git a/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/S3AFileSystem.java b/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/S3AFileSystem.java
index c7511adf8ab1..cdcb0c9cb148 100644
--- a/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/S3AFileSystem.java
+++ b/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/S3AFileSystem.java
@@ -28,6 +28,7 @@ import java.net.URI;
import java.nio.file.AccessDeniedException;
import java.text.DateFormat;
import java.text.SimpleDateFormat;
+import java.time.Duration;
import java.time.Instant;
import java.util.ArrayList;
import java.util.Collection;
@@ -122,6 +123,7 @@ import org.apache.hadoop.fs.s3a.impl.AWSCannedACL;
import org.apache.hadoop.fs.s3a.impl.AWSHeaders;
import org.apache.hadoop.fs.s3a.impl.BulkDeleteRetryHandler;
import org.apache.hadoop.fs.s3a.impl.ChangeDetectionPolicy;
+import org.apache.hadoop.fs.s3a.impl.ConfigurationHelper;
import org.apache.hadoop.fs.s3a.impl.ContextAccessors;
import org.apache.hadoop.fs.s3a.impl.CopyFromLocalOperation;
import org.apache.hadoop.fs.s3a.impl.CreateFileBuilder;
@@ -821,8 +823,13 @@ public class S3AFileSystem extends FileSystem implements StreamCapabilities,
}
int totalTasks = intOption(conf,
MAX_TOTAL_TASKS, DEFAULT_MAX_TOTAL_TASKS, 1);
- long keepAliveTime = longOption(conf, KEEPALIVE_TIME,
- DEFAULT_KEEPALIVE_TIME, 0);
+ // keepalive time takes a time suffix; default unit is seconds
+ long keepAliveTime = ConfigurationHelper.getDuration(conf,
+ KEEPALIVE_TIME,
+ Duration.ofSeconds(DEFAULT_KEEPALIVE_TIME),
+ TimeUnit.SECONDS,
+ Duration.ZERO).getSeconds();
+
int numPrefetchThreads = this.prefetchEnabled ? this.prefetchBlockCount : 0;
int activeTasksForBoundedThreadPool = maxThreads;
@@ -1218,12 +1225,15 @@ public class S3AFileSystem extends FileSystem implements StreamCapabilities,
private void initMultipartUploads(Configuration conf) throws IOException {
boolean purgeExistingMultipart = conf.getBoolean(PURGE_EXISTING_MULTIPART,
DEFAULT_PURGE_EXISTING_MULTIPART);
- long purgeExistingMultipartAge = longOption(conf,
- PURGE_EXISTING_MULTIPART_AGE, DEFAULT_PURGE_EXISTING_MULTIPART_AGE, 0);
if (purgeExistingMultipart) {
try {
- abortOutstandingMultipartUploads(purgeExistingMultipartAge);
+ Duration purgeDuration = ConfigurationHelper.getDuration(conf,
+ PURGE_EXISTING_MULTIPART_AGE,
+ Duration.ofSeconds(DEFAULT_PURGE_EXISTING_MULTIPART_AGE),
+ TimeUnit.SECONDS,
+ Duration.ZERO);
+ abortOutstandingMultipartUploads(purgeDuration.getSeconds());
} catch (AccessDeniedException e) {
instrumentation.errorIgnored();
LOG.debug("Failed to purge multipart uploads against {}," +
diff --git a/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/S3ARetryPolicy.java b/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/S3ARetryPolicy.java
index cc30da4fbe04..9438ac22bdb1 100644
--- a/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/S3ARetryPolicy.java
+++ b/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/S3ARetryPolicy.java
@@ -201,6 +201,9 @@ public class S3ARetryPolicy implements RetryPolicy {
// Treated as an immediate failure
policyMap.put(AWSBadRequestException.class, fail);
+ // API call timeout.
+ policyMap.put(AWSApiCallTimeoutException.class, retryAwsClientExceptions);
+
// use specific retry policy for aws client exceptions
// nested IOExceptions will already have been extracted and used
// in this map.
diff --git a/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/S3AUtils.java b/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/S3AUtils.java
index 155bd8a6105a..94fa2d35fc6f 100644
--- a/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/S3AUtils.java
+++ b/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/S3AUtils.java
@@ -20,6 +20,8 @@ package org.apache.hadoop.fs.s3a;
import software.amazon.awssdk.awscore.exception.AwsServiceException;
import software.amazon.awssdk.core.exception.AbortedException;
+import software.amazon.awssdk.core.exception.ApiCallAttemptTimeoutException;
+import software.amazon.awssdk.core.exception.ApiCallTimeoutException;
import software.amazon.awssdk.core.exception.SdkException;
import software.amazon.awssdk.core.retry.RetryUtils;
import software.amazon.awssdk.services.s3.model.S3Exception;
@@ -169,9 +171,23 @@ public final class S3AUtils {
operation,
StringUtils.isNotEmpty(path)? (" on " + path) : "",
exception);
+
+ // timeout issues
+ // ApiCallAttemptTimeoutException: a single HTTP request attempt failed.
+ // ApiCallTimeoutException: a request with any configured retries failed.
+ // The ApiCallTimeoutException exception should be the only one seen in
+ // the S3A code, but for due diligence both are handled and mapped to
+ // our own AWSApiCallTimeoutException.
+ if (exception instanceof ApiCallTimeoutException
+ || exception instanceof ApiCallAttemptTimeoutException) {
+ // An API call to an AWS service timed out.
+ // This is a subclass of ConnectTimeoutException so
+ // all retry logic for that exception is handled without
+ // having to look down the stack for a
+ return new AWSApiCallTimeoutException(message, exception);
+ }
if (!(exception instanceof AwsServiceException)) {
// exceptions raised client-side: connectivity, auth, network problems...
-
Exception innerCause = containsInterruptedException(exception);
if (innerCause != null) {
// interrupted IO, or a socket exception underneath that class
@@ -291,6 +307,11 @@ public final class S3AUtils {
ioe = new AWSServiceThrottledException(message, ase);
break;
+ // gateway timeout
+ case SC_504_GATEWAY_TIMEOUT:
+ ioe = new AWSApiCallTimeoutException(message, ase);
+ break;
+
// internal error
case SC_500_INTERNAL_SERVER_ERROR:
ioe = new AWSStatus500Exception(message, ase);
diff --git a/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/impl/AWSClientConfig.java b/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/impl/AWSClientConfig.java
index 3400daff509b..263562fe8a70 100644
--- a/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/impl/AWSClientConfig.java
+++ b/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/impl/AWSClientConfig.java
@@ -28,29 +28,37 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import software.amazon.awssdk.core.client.config.ClientOverrideConfiguration;
import software.amazon.awssdk.core.client.config.SdkAdvancedClientOption;
+import software.amazon.awssdk.core.retry.RetryMode;
import software.amazon.awssdk.core.retry.RetryPolicy;
import software.amazon.awssdk.http.apache.ApacheHttpClient;
import software.amazon.awssdk.http.apache.ProxyConfiguration;
import software.amazon.awssdk.http.nio.netty.NettyNioAsyncHttpClient;
+import org.apache.hadoop.classification.VisibleForTesting;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.s3a.S3AUtils;
import org.apache.hadoop.fs.s3a.auth.SignerFactory;
import org.apache.hadoop.util.VersionInfo;
-import org.apache.http.client.utils.URIBuilder;
+import static org.apache.hadoop.fs.s3a.Constants.CONNECTION_ACQUISITION_TIMEOUT;
import static org.apache.hadoop.fs.s3a.Constants.AWS_SERVICE_IDENTIFIER_S3;
import static org.apache.hadoop.fs.s3a.Constants.AWS_SERVICE_IDENTIFIER_STS;
+import static org.apache.hadoop.fs.s3a.Constants.CONNECTION_IDLE_TIME;
+import static org.apache.hadoop.fs.s3a.Constants.CONNECTION_KEEPALIVE;
import static org.apache.hadoop.fs.s3a.Constants.CONNECTION_TTL;
-import static org.apache.hadoop.fs.s3a.Constants.DEFAULT_CONNECTION_TTL;
-import static org.apache.hadoop.fs.s3a.Constants.DEFAULT_ESTABLISH_TIMEOUT;
+import static org.apache.hadoop.fs.s3a.Constants.DEFAULT_CONNECTION_ACQUISITION_TIMEOUT_DURATION;
+import static org.apache.hadoop.fs.s3a.Constants.DEFAULT_CONNECTION_IDLE_TIME_DURATION;
+import static org.apache.hadoop.fs.s3a.Constants.DEFAULT_CONNECTION_KEEPALIVE;
+import static org.apache.hadoop.fs.s3a.Constants.DEFAULT_CONNECTION_TTL_DURATION;
+import static org.apache.hadoop.fs.s3a.Constants.DEFAULT_ESTABLISH_TIMEOUT_DURATION;
import static org.apache.hadoop.fs.s3a.Constants.DEFAULT_MAXIMUM_CONNECTIONS;
import static org.apache.hadoop.fs.s3a.Constants.DEFAULT_MAX_ERROR_RETRIES;
-import static org.apache.hadoop.fs.s3a.Constants.DEFAULT_REQUEST_TIMEOUT;
-import static org.apache.hadoop.fs.s3a.Constants.DEFAULT_SOCKET_TIMEOUT;
+import static org.apache.hadoop.fs.s3a.Constants.DEFAULT_REQUEST_TIMEOUT_DURATION;
+import static org.apache.hadoop.fs.s3a.Constants.DEFAULT_SOCKET_TIMEOUT_DURATION;
import static org.apache.hadoop.fs.s3a.Constants.ESTABLISH_TIMEOUT;
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.MINIMUM_NETWORK_OPERATION_DURATION;
import static org.apache.hadoop.fs.s3a.Constants.PROXY_DOMAIN;
import static org.apache.hadoop.fs.s3a.Constants.PROXY_HOST;
import static org.apache.hadoop.fs.s3a.Constants.PROXY_PASSWORD;
@@ -64,19 +72,40 @@ import static org.apache.hadoop.fs.s3a.Constants.SIGNING_ALGORITHM_S3;
import static org.apache.hadoop.fs.s3a.Constants.SIGNING_ALGORITHM_STS;
import static org.apache.hadoop.fs.s3a.Constants.SOCKET_TIMEOUT;
import static org.apache.hadoop.fs.s3a.Constants.USER_AGENT_PREFIX;
-import static org.apache.hadoop.fs.s3a.S3AUtils.longOption;
+import static org.apache.hadoop.fs.s3a.impl.ConfigurationHelper.enforceMinimumDuration;
+import static org.apache.hadoop.fs.s3a.impl.ConfigurationHelper.getDuration;
+import static org.apache.hadoop.util.Preconditions.checkArgument;
+
/**
* Methods for configuring the S3 client.
* These methods are used when creating and configuring
- * {@link software.amazon.awssdk.services.s3.S3Client} which communicates with the S3 service.
+ * the HTTP clients which communicate with the S3 service.
+ * <p>
+ * See {@code software.amazon.awssdk.http.SdkHttpConfigurationOption}
+ * for the default values.
*/
public final class AWSClientConfig {
+
private static final Logger LOG = LoggerFactory.getLogger(AWSClientConfig.class);
+ /**
+ * The minimum operation duration.
+ */
+ private static Duration minimumOperationDuration = MINIMUM_NETWORK_OPERATION_DURATION;
+
+
private AWSClientConfig() {
}
+ /**
+ * Create the config for a given service...the service identifier is used
+ * to determine signature implementation.
+ * @param conf configuration
+ * @param awsServiceIdentifier service
+ * @return the builder inited with signer, timeouts and UA.
+ * @throws IOException failure.
+ */
public static ClientOverrideConfiguration.Builder createClientConfigBuilder(Configuration conf,
String awsServiceIdentifier) throws IOException {
ClientOverrideConfiguration.Builder overrideConfigBuilder =
@@ -99,7 +128,10 @@ public final class AWSClientConfig {
}
/**
- * Configures the http client.
+ * Create and configure the http client-based connector with timeouts for:
+ * connection acquisition, max idle, timeout, TTL, socket and keepalive.
+ * SSL channel mode is set up via
+ * {@link NetworkBinding#bindSSLChannelMode(Configuration, ApacheHttpClient.Builder)}.
*
* @param conf The Hadoop configuration
* @return Http client builder
@@ -107,24 +139,17 @@ public final class AWSClientConfig {
*/
public static ApacheHttpClient.Builder createHttpClientBuilder(Configuration conf)
throws IOException {
+ final ConnectionSettings conn = createConnectionSettings(conf);
ApacheHttpClient.Builder httpClientBuilder =
- ApacheHttpClient.builder();
-
- httpClientBuilder.maxConnections(S3AUtils.intOption(conf, MAXIMUM_CONNECTIONS,
- DEFAULT_MAXIMUM_CONNECTIONS, 1));
-
- int connectionEstablishTimeout =
- S3AUtils.intOption(conf, ESTABLISH_TIMEOUT, DEFAULT_ESTABLISH_TIMEOUT, 0);
- int socketTimeout = S3AUtils.intOption(conf, SOCKET_TIMEOUT, DEFAULT_SOCKET_TIMEOUT, 0);
-
- httpClientBuilder.connectionTimeout(Duration.ofMillis(connectionEstablishTimeout));
- httpClientBuilder.socketTimeout(Duration.ofMillis(socketTimeout));
-
- // set the connection TTL irrespective of whether the connection is in use or not.
- // this can balance requests over different S3 servers, and avoid failed
- // connections. See HADOOP-18845.
- long ttl = longOption(conf, CONNECTION_TTL, DEFAULT_CONNECTION_TTL, -1);
- httpClientBuilder.connectionTimeToLive(Duration.ofMillis(ttl));
+ ApacheHttpClient.builder()
+ .connectionAcquisitionTimeout(conn.getAcquisitionTimeout())
+ .connectionMaxIdleTime(conn.getMaxIdleTime())
+ .connectionTimeout(conn.getEstablishTimeout())
+ .connectionTimeToLive(conn.getConnectionTTL())
+ .maxConnections(conn.getMaxConnections())
+ .socketTimeout(conn.getSocketTimeout())
+ .tcpKeepAlive(conn.isKeepAlive())
+ .useIdleConnectionReaper(true); // true by default in the SDK
NetworkBinding.bindSSLChannelMode(conf, httpClientBuilder);
@@ -132,31 +157,26 @@ public final class AWSClientConfig {
}
/**
- * Configures the async http client.
- *
+ * Create and configure the async http client with timeouts for:
+ * connection acquisition, max idle, timeout, TTL, socket and keepalive.
+ * This is netty based and does not allow for the SSL channel mode to be set.
* @param conf The Hadoop configuration
- * @return Http client builder
+ * @return Async Http client builder
*/
public static NettyNioAsyncHttpClient.Builder createAsyncHttpClientBuilder(Configuration conf) {
- NettyNioAsyncHttpClient.Builder httpClientBuilder =
- NettyNioAsyncHttpClient.builder();
+ final ConnectionSettings conn = createConnectionSettings(conf);
- httpClientBuilder.maxConcurrency(S3AUtils.intOption(conf, MAXIMUM_CONNECTIONS,
- DEFAULT_MAXIMUM_CONNECTIONS, 1));
-
- int connectionEstablishTimeout =
- S3AUtils.intOption(conf, ESTABLISH_TIMEOUT, DEFAULT_ESTABLISH_TIMEOUT, 0);
- int socketTimeout = S3AUtils.intOption(conf, SOCKET_TIMEOUT, DEFAULT_SOCKET_TIMEOUT, 0);
-
- httpClientBuilder.connectionTimeout(Duration.ofMillis(connectionEstablishTimeout));
- httpClientBuilder.readTimeout(Duration.ofMillis(socketTimeout));
- httpClientBuilder.writeTimeout(Duration.ofMillis(socketTimeout));
-
- // set the connection TTL irrespective of whether the connection is in use or not.
- // this can balance requests over different S3 servers, and avoid failed
- // connections. See HADOOP-18845.
- long ttl = longOption(conf, CONNECTION_TTL, DEFAULT_CONNECTION_TTL, -1);
- httpClientBuilder.connectionTimeToLive(Duration.ofMillis(ttl));
+ NettyNioAsyncHttpClient.Builder httpClientBuilder =
+ NettyNioAsyncHttpClient.builder()
+ .connectionAcquisitionTimeout(conn.getAcquisitionTimeout())
+ .connectionMaxIdleTime(conn.getMaxIdleTime())
+ .connectionTimeout(conn.getEstablishTimeout())
+ .connectionTimeToLive(conn.getConnectionTTL())
+ .maxConcurrency(conn.getMaxConnections())
+ .readTimeout(conn.getSocketTimeout())
+ .tcpKeepAlive(conn.isKeepAlive())
+ .useIdleConnectionReaper(true) // true by default in the SDK
+ .writeTimeout(conn.getSocketTimeout());
// TODO: Don't think you can set a socket factory for the netty client.
// NetworkBinding.bindSSLChannelMode(conf, awsConf);
@@ -166,13 +186,19 @@ public final class AWSClientConfig {
/**
* Configures the retry policy.
+ * Retry policy is {@code RetryMode.ADAPTIVE}, which
+ * "dynamically limits the rate of AWS requests to maximize success rate",
+ * possibly at the expense of latency.
+ * Based on the ABFS experience, it is better to limit the rate requests are
+ * made rather than have to resort to exponential backoff after failures come
+ * in -especially as that backoff is per http connection.
*
* @param conf The Hadoop configuration
* @return Retry policy builder
*/
public static RetryPolicy.Builder createRetryPolicyBuilder(Configuration conf) {
- RetryPolicy.Builder retryPolicyBuilder = RetryPolicy.builder();
+ RetryPolicy.Builder retryPolicyBuilder = RetryPolicy.builder(RetryMode.ADAPTIVE);
retryPolicyBuilder.numRetries(S3AUtils.intOption(conf, MAX_ERROR_RETRIES,
DEFAULT_MAX_ERROR_RETRIES, 0));
@@ -236,11 +262,14 @@ public final class AWSClientConfig {
}
return proxyConfigBuilder.build();
+
}
/**
* Configures the proxy for the async http client.
- *
+ * <p>
+ * Although this is netty specific, it is part of the AWS SDK public API
+ * and not any shaded netty internal class.
* @param conf The Hadoop configuration
* @param bucket Optional bucket to use to look up per-bucket proxy secrets
* @return Proxy configuration
@@ -248,7 +277,7 @@ public final class AWSClientConfig {
*/
public static software.amazon.awssdk.http.nio.netty.ProxyConfiguration
createAsyncProxyConfiguration(Configuration conf,
- String bucket) throws IOException {
+ String bucket) throws IOException {
software.amazon.awssdk.http.nio.netty.ProxyConfiguration.Builder proxyConfigBuilder =
software.amazon.awssdk.http.nio.netty.ProxyConfiguration.builder();
@@ -307,7 +336,7 @@ public final class AWSClientConfig {
return proxyConfigBuilder.build();
}
- /***
+ /**
* Builds a URI, throws an IllegalArgumentException in case of errors.
*
* @param host proxy host
@@ -316,7 +345,7 @@ public final class AWSClientConfig {
*/
private static URI buildURI(String scheme, String host, int port) {
try {
- return new URIBuilder().setScheme(scheme).setHost(host).setPort(port).build();
+ return new URI(scheme, null, host, port, null, null, null);
} catch (URISyntaxException e) {
String msg =
"Proxy error: incorrect " + PROXY_HOST + " or " + PROXY_PORT;
@@ -346,6 +375,13 @@ public final class AWSClientConfig {
clientConfig.putAdvancedOption(SdkAdvancedClientOption.USER_AGENT_PREFIX, userAgent);
}
+ /**
+ * Initializes the signer override for the given service.
+ * @param conf hadoop configuration
+ * @param clientConfig client configuration to update
+ * @param awsServiceIdentifier service name
+ * @throws IOException failure.
+ */
private static void initSigner(Configuration conf,
ClientOverrideConfiguration.Builder clientConfig, String awsServiceIdentifier)
throws IOException {
@@ -371,24 +407,219 @@ public final class AWSClientConfig {
}
/**
- * Configures request timeout.
+ * Configures request timeout in the client configuration.
+ * This is independent of the timeouts set in the sync and async HTTP clients;
+ * the same method
*
* @param conf Hadoop configuration
* @param clientConfig AWS SDK configuration to update
*/
private static void initRequestTimeout(Configuration conf,
ClientOverrideConfiguration.Builder clientConfig) {
- long requestTimeoutMillis = conf.getTimeDuration(REQUEST_TIMEOUT,
- DEFAULT_REQUEST_TIMEOUT, TimeUnit.SECONDS, TimeUnit.MILLISECONDS);
+ // Get the connection settings
+ final Duration callTimeout = createApiConnectionSettings(conf).getApiCallTimeout();
- if (requestTimeoutMillis > Integer.MAX_VALUE) {
- LOG.debug("Request timeout is too high({} ms). Setting to {} ms instead",
- requestTimeoutMillis, Integer.MAX_VALUE);
- requestTimeoutMillis = Integer.MAX_VALUE;
+ if (callTimeout.toMillis() > 0) {
+ clientConfig.apiCallAttemptTimeout(callTimeout);
+ clientConfig.apiCallTimeout(callTimeout);
}
+ }
- if(requestTimeoutMillis > 0) {
- clientConfig.apiCallAttemptTimeout(Duration.ofMillis(requestTimeoutMillis));
+ /**
+ * Reset the minimum operation duration to the default.
+ * For test use only; Logs at INFO.
+ * <p>
+ * This MUST be called in test teardown in any test suite which
+ * called {@link #setMinimumOperationDuration(Duration)}.
+ */
+ @VisibleForTesting
+ public static void resetMinimumOperationDuration() {
+ setMinimumOperationDuration(MINIMUM_NETWORK_OPERATION_DURATION);
+ }
+
+ /**
+ * Set the minimum operation duration.
+ * This is for testing and will log at info; does require a non-negative duration.
+ * <p>
+ * Test suites must call {@link #resetMinimumOperationDuration()} in their teardown
+ * to avoid breaking subsequent tests in the same process.
+ * @param duration non-negative duration
+ * @throws IllegalArgumentException if the duration is negative.
+ */
+ @VisibleForTesting
+ public static void setMinimumOperationDuration(Duration duration) {
+ LOG.info("Setting minimum operation duration to {}ms", duration.toMillis());
+ checkArgument(duration.compareTo(Duration.ZERO) >= 0,
+ "Duration must be positive: %sms", duration.toMillis());
+ minimumOperationDuration = duration;
+ }
+
+ /**
+ * Get the current minimum operation duration.
+ * @return current duration.
+ */
+ public static Duration getMinimumOperationDuration() {
+ return minimumOperationDuration;
+ }
+
+ /**
+ * Settings for the AWS client, rather than the http client.
+ */
+ static class ClientSettings {
+ private final Duration apiCallTimeout;
+
+ private ClientSettings(final Duration apiCallTimeout) {
+ this.apiCallTimeout = apiCallTimeout;
+ }
+
+ Duration getApiCallTimeout() {
+ return apiCallTimeout;
+ }
+
+ @Override
+ public String toString() {
+ return "ClientSettings{" +
+ "apiCallTimeout=" + apiCallTimeout +
+ '}';
}
}
+
+ /**
+ * All the connection settings, wrapped as a class for use by
+ * both the sync and async client.
+ */
+ static class ConnectionSettings {
+ private final int maxConnections;
+ private final boolean keepAlive;
+ private final Duration acquisitionTimeout;
+ private final Duration connectionTTL;
+ private final Duration establishTimeout;
+ private final Duration maxIdleTime;
+ private final Duration socketTimeout;
+
+ private ConnectionSettings(
+ final int maxConnections,
+ final boolean keepAlive,
+ final Duration acquisitionTimeout,
+ final Duration connectionTTL,
+ final Duration establishTimeout,
+ final Duration maxIdleTime,
+ final Duration socketTimeout) {
+ this.maxConnections = maxConnections;
+ this.keepAlive = keepAlive;
+ this.acquisitionTimeout = acquisitionTimeout;
+ this.connectionTTL = connectionTTL;
+ this.establishTimeout = establishTimeout;
+ this.maxIdleTime = maxIdleTime;
+ this.socketTimeout = socketTimeout;
+ }
+
+ int getMaxConnections() {
+ return maxConnections;
+ }
+
+ boolean isKeepAlive() {
+ return keepAlive;
+ }
+
+ Duration getAcquisitionTimeout() {
+ return acquisitionTimeout;
+ }
+
+ Duration getConnectionTTL() {
+ return connectionTTL;
+ }
+
+ Duration getEstablishTimeout() {
+ return establishTimeout;
+ }
+
+ Duration getMaxIdleTime() {
+ return maxIdleTime;
+ }
+
+ Duration getSocketTimeout() {
+ return socketTimeout;
+ }
+
+ @Override
+ public String toString() {
+ return "ConnectionSettings{" +
+ "maxConnections=" + maxConnections +
+ ", keepAlive=" + keepAlive +
+ ", acquisitionTimeout=" + acquisitionTimeout +
+ ", connectionTTL=" + connectionTTL +
+ ", establishTimeout=" + establishTimeout +
+ ", maxIdleTime=" + maxIdleTime +
+ ", socketTimeout=" + socketTimeout +
+ '}';
+ }
+ }
+
+ /**
+ * Build a client settings object.
+ * @param conf configuration to evaluate
+ * @return connection settings.
+ */
+ static ClientSettings createApiConnectionSettings(Configuration conf) {
+
+ Duration apiCallTimeout = getDuration(conf, REQUEST_TIMEOUT,
+ DEFAULT_REQUEST_TIMEOUT_DURATION, TimeUnit.MILLISECONDS, Duration.ZERO);
+
+ // if the API call timeout is set, it must be at least the minimum duration
+ if (apiCallTimeout.compareTo(Duration.ZERO) > 0) {
+ apiCallTimeout = enforceMinimumDuration(REQUEST_TIMEOUT,
+ apiCallTimeout, minimumOperationDuration);
+ }
+ return new ClientSettings(apiCallTimeout);
+ }
+
+ /**
+ * Build the HTTP connection settings object from the configuration.
+ * All settings are calculated, including the api call timeout.
+ * @param conf configuration to evaluate
+ * @return connection settings.
+ */
+ static ConnectionSettings createConnectionSettings(Configuration conf) {
+
+ int maxConnections = S3AUtils.intOption(conf, MAXIMUM_CONNECTIONS,
+ DEFAULT_MAXIMUM_CONNECTIONS, 1);
+
+ final boolean keepAlive = conf.getBoolean(CONNECTION_KEEPALIVE,
+ DEFAULT_CONNECTION_KEEPALIVE);
+
+ // time to acquire a connection from the pool
+ Duration acquisitionTimeout = getDuration(conf, CONNECTION_ACQUISITION_TIMEOUT,
+ DEFAULT_CONNECTION_ACQUISITION_TIMEOUT_DURATION, TimeUnit.MILLISECONDS,
+ minimumOperationDuration);
+
+ // set the connection TTL irrespective of whether the connection is in use or not.
+ // this can balance requests over different S3 servers, and avoid failed
+ // connections. See HADOOP-18845.
+ Duration connectionTTL = getDuration(conf, CONNECTION_TTL,
+ DEFAULT_CONNECTION_TTL_DURATION, TimeUnit.MILLISECONDS,
+ null);
+
+ Duration establishTimeout = getDuration(conf, ESTABLISH_TIMEOUT,
+ DEFAULT_ESTABLISH_TIMEOUT_DURATION, TimeUnit.MILLISECONDS,
+ minimumOperationDuration);
+
+ // limit on the time a connection can be idle in the pool
+ Duration maxIdleTime = getDuration(conf, CONNECTION_IDLE_TIME,
+ DEFAULT_CONNECTION_IDLE_TIME_DURATION, TimeUnit.MILLISECONDS, Duration.ZERO);
+
+ Duration socketTimeout = getDuration(conf, SOCKET_TIMEOUT,
+ DEFAULT_SOCKET_TIMEOUT_DURATION, TimeUnit.MILLISECONDS,
+ minimumOperationDuration);
+
+ return new ConnectionSettings(
+ maxConnections,
+ keepAlive,
+ acquisitionTimeout,
+ connectionTTL,
+ establishTimeout,
+ maxIdleTime,
+ socketTimeout);
+ }
+
}
diff --git a/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/impl/ConfigurationHelper.java b/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/impl/ConfigurationHelper.java
new file mode 100644
index 000000000000..8acd20a82155
--- /dev/null
+++ b/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/impl/ConfigurationHelper.java
@@ -0,0 +1,132 @@
+/*
+ * 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.impl;
+
+import java.time.Duration;
+import java.util.concurrent.TimeUnit;
+import javax.annotation.Nullable;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.hadoop.classification.InterfaceAudience;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.store.LogExactlyOnce;
+
+/**
+ * Helper class for configuration; where methods related to extracting
+ * configuration should go instead of {@code S3AUtils}.
+ */
+@InterfaceAudience.Private
+public final class ConfigurationHelper {
+
+ private static final Logger LOG = LoggerFactory.getLogger(ConfigurationHelper.class);
+
+ /** Log to warn of range issues on a timeout. */
+ private static final LogExactlyOnce DURATION_WARN_LOG = new LogExactlyOnce(LOG);
+
+ private ConfigurationHelper() {
+ }
+
+ /**
+ * Get a duration. This may be negative; callers must check or set the minimum to zero.
+ * If the config option is greater than {@code Integer.MAX_VALUE} milliseconds,
+ * it is set to that max.
+ * If {@code minimumDuration} is set, and the value is less than that, then
+ * the minimum is used.
+ * Logs the value for diagnostics.
+ * @param conf config
+ * @param name option name
+ * @param defaultDuration default duration
+ * @param defaultUnit default unit on the config option if not declared.
+ * @param minimumDuration optional minimum duration;
+ * @return duration. may be negative.
+ */
+ public static Duration getDuration(
+ final Configuration conf,
+ final String name,
+ final Duration defaultDuration,
+ final TimeUnit defaultUnit,
+ @Nullable final Duration minimumDuration) {
+ long timeMillis = conf.getTimeDuration(name,
+ defaultDuration.toMillis(), defaultUnit, TimeUnit.MILLISECONDS);
+
+ if (timeMillis > Integer.MAX_VALUE) {
+ DURATION_WARN_LOG.warn("Option {} is too high({} ms). Setting to {} ms instead",
+ name, timeMillis, Integer.MAX_VALUE);
+ LOG.debug("Option {} is too high({} ms). Setting to {} ms instead",
+ name, timeMillis, Integer.MAX_VALUE);
+ timeMillis = Integer.MAX_VALUE;
+ }
+
+ Duration duration = enforceMinimumDuration(name, Duration.ofMillis(timeMillis),
+ minimumDuration);
+ LOG.debug("Duration of {} = {}", name, duration);
+ return duration;
+ }
+
+ /**
+ * Set a duration as a time in seconds, with the suffix "s" added.
+ * @param conf configuration to update.
+ * @param name option name
+ * @param duration duration
+ */
+ public static void setDurationAsSeconds(
+ final Configuration conf,
+ final String name,
+ final Duration duration) {
+ conf.set(name, duration.getSeconds() + "s");
+ }
+
+ /**
+ * Set a duration as a time in milliseconds, with the suffix "ms" added.
+ * @param conf configuration to update.
+ * @param name option name
+ * @param duration duration
+ */
+ public static void setDurationAsMillis(
+ final Configuration conf,
+ final String name,
+ final Duration duration) {
+ conf.set(name, duration.toMillis()+ "ms");
+ }
+
+ /**
+ * Enforce a minimum duration of a configuration option, if the supplied
+ * value is non-null.
+ * @param name option name
+ * @param duration duration to check
+ * @param minimumDuration minimum duration; may be null
+ * @return a duration which, if the minimum duration is set, is at least that value.
+ */
+ public static Duration enforceMinimumDuration(
+ final String name,
+ final Duration duration,
+ @Nullable final Duration minimumDuration) {
+
+ if (minimumDuration != null && duration.compareTo(minimumDuration) < 0) {
+ String message = String.format("Option %s is too low (%,d ms). Setting to %,d ms instead",
+ name, duration.toMillis(), minimumDuration.toMillis());
+ DURATION_WARN_LOG.warn(message);
+ LOG.debug(message);
+ return minimumDuration;
+ }
+ return duration;
+ }
+}
diff --git a/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/impl/InternalConstants.java b/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/impl/InternalConstants.java
index 3409bd3cc79e..8b29313faf2d 100644
--- a/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/impl/InternalConstants.java
+++ b/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/impl/InternalConstants.java
@@ -172,6 +172,9 @@ public final class InternalConstants {
/** 503 status code: Service Unavailable. on AWS S3: throttle response. */
public static final int SC_503_SERVICE_UNAVAILABLE = 503;
+ /** 504 Gateway Timeout. AWS SDK considers retryable. */
+ public static final int SC_504_GATEWAY_TIMEOUT = 504;
+
/** Name of the log for throttling events. Value: {@value}. */
public static final String THROTTLE_LOG_NAME =
"org.apache.hadoop.fs.s3a.throttled";
diff --git a/hadoop-tools/hadoop-aws/src/site/markdown/tools/hadoop-aws/index.md b/hadoop-tools/hadoop-aws/src/site/markdown/tools/hadoop-aws/index.md
index 076ea0fd5b8a..ed9cfddf0fe2 100644
--- a/hadoop-tools/hadoop-aws/src/site/markdown/tools/hadoop-aws/index.md
+++ b/hadoop-tools/hadoop-aws/src/site/markdown/tools/hadoop-aws/index.md
@@ -703,8 +703,12 @@ to allow different buckets to override the shared settings. This is commonly
used to change the endpoint, encryption and authentication mechanisms of buckets.
and various minor options.
-Here are the S3A properties for use in production; some testing-related
-options are covered in [Testing](./testing.md).
+Here are some the S3A properties for use in production.
+
+* See [Performance](./performance.html) for performance related settings including
+ thread and network pool options.
+* Testing-related options are covered in [Testing](./testing.md).
+
```xml
@@ -830,16 +834,6 @@ options are covered in [Testing](./testing.md).
</description>
</property>
-<property>
- <name>fs.s3a.connection.maximum</name>
- <value>96</value>
- <description>Controls the maximum number of simultaneous connections to S3.
- This must be bigger than the value of fs.s3a.threads.max so as to stop
- threads being blocked waiting for new HTTPS connections.
- Why not equal? The AWS SDK transfer manager also uses these connections.
- </description>
-</property>
-
<property>
<name>fs.s3a.connection.ssl.enabled</name>
<value>true</value>
@@ -908,18 +902,6 @@ options are covered in [Testing](./testing.md).
</description>
</property>
-<property>
- <name>fs.s3a.connection.establish.timeout</name>
- <value>5000</value>
- <description>Socket connection setup timeout in milliseconds.</description>
-</property>
-
-<property>
- <name>fs.s3a.connection.timeout</name>
- <value>200000</value>
- <description>Socket connection timeout in milliseconds.</description>
-</property>
-
<property>
<name>fs.s3a.socket.send.buffer</name>
<value>8192</value>
@@ -939,43 +921,6 @@ options are covered in [Testing](./testing.md).
directory listings at a time.</description>
</property>
-<property>
- <name>fs.s3a.threads.max</name>
- <value>64</value>
- <description>The total number of threads available in the filesystem for data
- uploads *or any other queued filesystem operation*.</description>
-</property>
-
-<property>
- <name>fs.s3a.threads.keepalivetime</name>
- <value>60</value>
- <description>Number of seconds a thread can be idle before being
- terminated.</description>
-</property>
-
-<property>
- <name>fs.s3a.max.total.tasks</name>
- <value>32</value>
- <description>The number of operations which can be queued for execution.
- This is in addition to the number of active threads in fs.s3a.threads.max.
- </description>
-</property>
-
-<property>
- <name>fs.s3a.executor.capacity</name>
- <value>16</value>
- <description>The maximum number of submitted tasks which is a single
- operation (e.g. rename(), delete()) may submit simultaneously for
- execution -excluding the IO-heavy block uploads, whose capacity
- is set in "fs.s3a.fast.upload.active.blocks"
-
- All tasks are submitted to the shared thread pool whose size is
- set in "fs.s3a.threads.max"; the value of capacity should be less than that
- of the thread pool itself, as the goal is to stop a single operation
- from overloading that thread pool.
- </description>
-</property>
-
<property>
<name>fs.s3a.multipart.size</name>
<value>64M</value>
@@ -2232,43 +2177,33 @@ from VMs running on EC2.
</description>
</property>
-<property>
- <name>fs.s3a.threads.max</name>
- <value>10</value>
- <description>The total number of threads available in the filesystem for data
- uploads *or any other queued filesystem operation*.</description>
-</property>
-
-<property>
- <name>fs.s3a.max.total.tasks</name>
- <value>5</value>
- <description>The number of operations which can be queued for execution</description>
-</property>
-
-<property>
- <name>fs.s3a.threads.keepalivetime</name>
- <value>60</value>
- <description>Number of seconds a thread can be idle before being
- terminated.</description>
-</property>
```
### <a name="multipart_purge"></a>Cleaning up after partial Upload Failures
-There are two mechanisms for cleaning up after leftover multipart
+There are four mechanisms for cleaning up after leftover multipart
uploads:
+- AWS Lifecycle rules. This is the simplest and SHOULD be used unless there
+ are are good reasons, such as the store not supporting lifecycle rules.
- Hadoop s3guard CLI commands for listing and deleting uploads by their
age. Documented in the [S3Guard](./s3guard.html) section.
+- Setting `fs.s3a.directory.operations.purge.uploads` to `true` for automatic
+ scan and delete during directory rename and delete
- The configuration parameter `fs.s3a.multipart.purge`, covered below.
If a large stream write operation is interrupted, there may be
intermediate partitions uploaded to S3 —data which will be billed for.
+If an S3A committer job is halted partway through, again, there may be
+many incomplete multipart uploads in the output directory.
These charges can be reduced by enabling `fs.s3a.multipart.purge`,
-and setting a purge time in seconds, such as 86400 seconds —24 hours.
+and setting a purge time in seconds, such as 24 hours.
When an S3A FileSystem instance is instantiated with the purge time greater
than zero, it will, on startup, delete all outstanding partition requests
-older than this time.
+older than this time. However, this makes filesystem instantiate slow, especially
+against very large buckets, as a full scan is made.
+
+Consider avoiding this in future.
```xml
<property>
@@ -2280,7 +2215,7 @@ older than this time.
<property>
<name>fs.s3a.multipart.purge.age</name>
- <value>86400</value>
+ <value>24h</value>
<description>Minimum age in seconds of multipart uploads to purge</description>
</property>
```
diff --git a/hadoop-tools/hadoop-aws/src/site/markdown/tools/hadoop-aws/performance.md b/hadoop-tools/hadoop-aws/src/site/markdown/tools/hadoop-aws/performance.md
index 88e6e8a0b211..e1d334bbfc43 100644
--- a/hadoop-tools/hadoop-aws/src/site/markdown/tools/hadoop-aws/performance.md
+++ b/hadoop-tools/hadoop-aws/src/site/markdown/tools/hadoop-aws/performance.md
@@ -196,40 +196,93 @@ Fix: Use one of the dedicated [S3A Committers](committers.md).
## <a name="tuning"></a> Options to Tune
-### <a name="pooling"></a> Thread and connection pool sizes.
+### <a name="pooling"></a> Thread and connection pool settings.
Each S3A client interacting with a single bucket, as a single user, has its
-own dedicated pool of open HTTP 1.1 connections alongside a pool of threads used
-for upload and copy operations.
+own dedicated pool of open HTTP connections alongside a pool of threads used
+for background/parallel operations in addition to the worker threads of the
+actual application.
+
The default pool sizes are intended to strike a balance between performance
and memory/thread use.
You can have a larger pool of (reused) HTTP connections and threads
-for parallel IO (especially uploads) by setting the properties
+for parallel IO (especially uploads, prefetching and vector reads) by setting the appropriate
+properties. Note: S3A Connectors have their own thread pools for job commit, but
+everything uses the same HTTP connection pool.
+| Property | Default | Meaning |
+|--------------------------------|---------|------------------------------------------------------------------|
+| `fs.s3a.threads.max` | `96` | Threads in the thread pool |
+| `fs.s3a.threads.keepalivetime` | `60s` | Expiry time for idle threads in the thread pool |
+| `fs.s3a.executor.capacity` | `16` | Maximum threads for any single operation |
+| `fs.s3a.max.total.tasks` | `16` | Extra tasks which can be queued excluding prefetching operations |
-| property | meaning | default |
-|----------|---------|---------|
-| `fs.s3a.threads.max`| Threads in the AWS transfer manager| 10 |
-| `fs.s3a.connection.maximum`| Maximum number of HTTP connections | 10|
-We recommend using larger values for processes which perform
-a lot of IO: `DistCp`, Spark Workers and similar.
+Network timeout options can be tuned to make the client fail faster *or* retry more.
+The choice is yours. Generally recovery is better, but sometimes fail-fast is more useful.
-```xml
-<property>
- <name>fs.s3a.threads.max</name>
- <value>20</value>
-</property>
-<property>
- <name>fs.s3a.connection.maximum</name>
- <value>20</value>
-</property>
-```
-Be aware, however, that processes which perform many parallel queries
-may consume large amounts of resources if each query is working with
-a different set of s3 buckets, or are acting on behalf of different users.
+| Property | Default | V2 | Meaning |
+|-----------------------------------------|---------|:----|-------------------------------------------------------|
+| `fs.s3a.connection.maximum` | `200` | | Connection pool size |
+| `fs.s3a.connection.keepalive` | `false` | `*` | Use TCP keepalive on open channels |
+| `fs.s3a.connection.acquisition.timeout` | `60s` | `*` | Timeout for waiting for a connection from the pool. |
+| `fs.s3a.connection.establish.timeout` | `30s` | | Time to establish the TCP/TLS connection |
+| `fs.s3a.connection.idle.time` | `60s` | `*` | Maximum time for idle HTTP connections in the pool |
+| `fs.s3a.connection.request.timeout` | `0` | | If greater than zero, maximum duration of any request |
+| `fs.s3a.connection.timeout` | `200s` | | Timeout for socket problems on a TCP channel |
+| `fs.s3a.connection.ttl` | `5m` | | Lifetime of HTTP connections from the pool |
+
+
+Units:
+1. The default unit for all these options except for `fs.s3a.threads.keepalivetime` is milliseconds, unless a time suffix is declared.
+2. Versions of Hadoop built with the AWS V1 SDK *only* support milliseconds rather than suffix values.
+ If configurations are intended to apply across hadoop releases, you MUST use milliseconds without a suffix.
+3. `fs.s3a.threads.keepalivetime` has a default unit of seconds on all hadoop releases.
+4. Options flagged as "V2" are new with the AWS V2 SDK; they are ignored on V1 releases.
+
+---
+
+There are some hard tuning decisions related to pool size and expiry.
+As servers add more cores and services add many more worker threads, a larger pool size is more and more important:
+the default values in `core-default.xml` have been slowly increased over time but should be treated as
+"the best", simply what is considered a good starting case.
+With Vectored IO adding multiple GET requests per Spark/Hive worker thread,
+and stream prefetching performing background block prefetch, larger pool and thread sizes are even more important.
+
+In large hive deployments, thread and connection pools of thousands have been known to have been set.
+
+Small pool: small value in `fs.s3a.connection.maximum`.
+* Keeps network/memory cost of having many S3A instances in the same process low.
+* But: limit on how many connections can be open at at a time.
+
+* Large Pool. More HTTP connections can be created and kept, but cost of keeping network connections increases
+unless idle time is reduced through `fs.s3a.connection.idle.time`.
+
+If exceptions are raised with about timeouts acquiring connections from the pool, this can be a symptom of
+* Heavy load. Increase pool size and acquisition timeout `fs.s3a.connection.acquisition.timeout`
+* Process failing to close open input streams from the S3 store.
+ Fix: Find uses of `open()`/`openFile()` and make sure that the streams are being `close()d`
+
+*Retirement of HTTP Connections.*
+
+Connections are retired from the pool by `fs.s3a.connection.idle.time`, the maximum time for idle connections,
+and `fs.s3a.connection.ttl`, the maximum life of any connection in the pool, even if it repeatedly reused.
+
+Limiting idle time saves on network connections, at the cost of requiring new connections on subsequent S3 operations.
+
+Limiting connection TTL is useful to spread across load balancers and recover from some network
+connection problems, including those caused by proxies.
+
+*Request timeout*: `fs.s3a.connection.request.timeout`
+
+If set, this sets an upper limit on any non-streaming API call (i.e. everything but `GET`).
+
+A timeout is good to detect and recover from failures.
+However, it also sets a limit on the duration of a POST/PUT of data
+-which, if after a timeout, will only be repeated, ultimately to failure.
+
### For large data uploads, tune the block size: `fs.s3a.block.size`
@@ -327,18 +380,6 @@ efficient in terms of HTTP connection use, and reduce the IOP rate against
the S3 bucket/shard.
```xml
-<property>
- <name>fs.s3a.threads.max</name>
- <value>20</value>
-</property>
-
-<property>
- <name>fs.s3a.connection.maximum</name>
- <value>30</value>
- <descriptiom>
- Make greater than both fs.s3a.threads.max and -numListstatusThreads
- </descriptiom>
-</property>
<property>
<name>fs.s3a.experimental.input.fadvise</name>
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 c4d9ca7bcc04..b26ca6889bd1 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
@@ -24,6 +24,7 @@ import static org.apache.hadoop.fs.s3a.S3ATestUtils.verifyExceptionClass;
import static org.apache.hadoop.fs.s3a.S3AUtils.*;
import static org.apache.hadoop.fs.s3a.audit.AuditIntegration.maybeTranslateAuditException;
import static org.apache.hadoop.fs.s3a.impl.InternalConstants.*;
+import static org.apache.hadoop.test.LambdaTestUtils.verifyCause;
import static org.junit.Assert.*;
import java.io.EOFException;
@@ -37,6 +38,8 @@ import java.util.function.Consumer;
import org.assertj.core.api.Assertions;
import software.amazon.awssdk.awscore.exception.AwsErrorDetails;
import software.amazon.awssdk.awscore.exception.AwsServiceException;
+import software.amazon.awssdk.core.exception.ApiCallAttemptTimeoutException;
+import software.amazon.awssdk.core.exception.ApiCallTimeoutException;
import software.amazon.awssdk.core.exception.SdkClientException;
import software.amazon.awssdk.core.exception.SdkException;
import software.amazon.awssdk.http.SdkHttpResponse;
@@ -44,11 +47,12 @@ import software.amazon.awssdk.services.s3.model.S3Exception;
import org.junit.Test;
+import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.s3a.api.UnsupportedRequestException;
import org.apache.hadoop.fs.s3a.audit.AuditFailureException;
import org.apache.hadoop.fs.s3a.audit.AuditOperationRejectedException;
import org.apache.hadoop.fs.s3a.impl.ErrorTranslation;
-
+import org.apache.hadoop.io.retry.RetryPolicy;
import static org.apache.hadoop.test.GenericTestUtils.assertExceptionContains;
@@ -330,4 +334,51 @@ public class TestS3AExceptionTranslation {
.isNull();
}
+ /**
+ * 504 gateway timeout is translated to a {@link AWSApiCallTimeoutException}.
+ */
+ @Test
+ public void test504ToTimeout() throws Throwable {
+ AWSApiCallTimeoutException ex =
+ verifyExceptionClass(AWSApiCallTimeoutException.class,
+ translateException("test", "/", createS3Exception(504)));
+ verifyCause(S3Exception.class, ex);
+ }
+
+ /**
+ * SDK ApiCallTimeoutException is translated to a
+ * {@link AWSApiCallTimeoutException}.
+ */
+ @Test
+ public void testApiCallTimeoutExceptionToTimeout() throws Throwable {
+ AWSApiCallTimeoutException ex =
+ verifyExceptionClass(AWSApiCallTimeoutException.class,
+ translateException("test", "/",
+ ApiCallTimeoutException.builder()
+ .message("timeout")
+ .build()));
+ verifyCause(ApiCallTimeoutException.class, ex);
+ }
+
+ /**
+ * SDK ApiCallAttemptTimeoutException is translated to a
+ * {@link AWSApiCallTimeoutException}.
+ */
+ @Test
+ public void testApiCallAttemptTimeoutExceptionToTimeout() throws Throwable {
+ AWSApiCallTimeoutException ex =
+ verifyExceptionClass(AWSApiCallTimeoutException.class,
+ translateException("test", "/",
+ ApiCallAttemptTimeoutException.builder()
+ .message("timeout")
+ .build()));
+ verifyCause(ApiCallAttemptTimeoutException.class, ex);
+
+ // and confirm these timeouts are retried.
+ final S3ARetryPolicy retryPolicy = new S3ARetryPolicy(new Configuration(false));
+ Assertions.assertThat(retryPolicy.shouldRetry(ex, 0, 0, true).action)
+ .describedAs("retry policy for exception %s", ex)
+ .isEqualTo(RetryPolicy.RetryAction.RetryDecision.RETRY);
+ }
+
}
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
new file mode 100644
index 000000000000..2da70e6ef459
--- /dev/null
+++ b/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/impl/ITestConnectionTimeouts.java
@@ -0,0 +1,149 @@
+/*
+ * 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.impl;
+
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.List;
+
+import org.junit.Test;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FSDataInputStream;
+import org.apache.hadoop.fs.FileStatus;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.FutureDataInputStreamBuilder;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.fs.contract.ContractTestUtils;
+import org.apache.hadoop.fs.s3a.AbstractS3ATestBase;
+import org.apache.hadoop.fs.s3a.S3AFileSystem;
+import org.apache.hadoop.io.IOUtils;
+import org.apache.hadoop.net.ConnectTimeoutException;
+
+import static org.apache.hadoop.fs.Options.OpenFileOptions.FS_OPTION_OPENFILE_READ_POLICY;
+import static org.apache.hadoop.fs.Options.OpenFileOptions.FS_OPTION_OPENFILE_READ_POLICY_WHOLE_FILE;
+import static org.apache.hadoop.fs.contract.ContractTestUtils.dataset;
+import static org.apache.hadoop.fs.s3a.Constants.CONNECTION_ACQUISITION_TIMEOUT;
+import static org.apache.hadoop.fs.s3a.Constants.CONNECTION_IDLE_TIME;
+import static org.apache.hadoop.fs.s3a.Constants.CONNECTION_TTL;
+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.MAXIMUM_CONNECTIONS;
+import static org.apache.hadoop.fs.s3a.Constants.MAX_ERROR_RETRIES;
+import static org.apache.hadoop.fs.s3a.Constants.PREFETCH_ENABLED_KEY;
+import static org.apache.hadoop.fs.s3a.Constants.REQUEST_TIMEOUT;
+import static org.apache.hadoop.fs.s3a.Constants.RETRY_LIMIT;
+import static org.apache.hadoop.fs.s3a.Constants.SOCKET_TIMEOUT;
+import static org.apache.hadoop.fs.s3a.S3ATestUtils.removeBaseAndBucketOverrides;
+import static org.apache.hadoop.fs.s3a.impl.ConfigurationHelper.setDurationAsMillis;
+import static org.apache.hadoop.test.LambdaTestUtils.intercept;
+
+/**
+ * Test for timeout generation/handling, especially those related to connection
+ * pools.
+ * This test has proven a bit brittle to parallel test runs, so test cases should
+ * create their own FS instances.
+ * The likely cause is actually -Dprefetch test runs as these return connections to
+ * the pool.
+ * However, it is also important to have a non-brittle FS for creating the test file
+ * and teardow, again, this makes for a flaky test..
+ */
+public class ITestConnectionTimeouts extends AbstractS3ATestBase {
+
+ /**
+ * How big a file to create?
+ */
+ public static final int FILE_SIZE = 1024;
+
+ /**
+ * Create a configuration for an FS which has timeouts set to very low values
+ * and no retries.
+ * @return a configuration to use for the brittle FS.
+ */
+ private Configuration timingOutConfiguration() {
+ Configuration conf = new Configuration(getConfiguration());
+ removeBaseAndBucketOverrides(conf,
+ CONNECTION_TTL,
+ CONNECTION_ACQUISITION_TIMEOUT,
+ CONNECTION_IDLE_TIME,
+ ESTABLISH_TIMEOUT,
+ MAX_ERROR_RETRIES,
+ MAXIMUM_CONNECTIONS,
+ PREFETCH_ENABLED_KEY,
+ REQUEST_TIMEOUT,
+ SOCKET_TIMEOUT,
+ FS_S3A_CREATE_PERFORMANCE
+ );
+
+ // only one connection is allowed, and the establish timeout is low
+ conf.setInt(MAXIMUM_CONNECTIONS, 1);
+ conf.setInt(MAX_ERROR_RETRIES, 0);
+ // needed to ensure that streams are kept open.
+ // without this the tests is unreliable in batch runs.
+ conf.setBoolean(PREFETCH_ENABLED_KEY, false);
+ conf.setInt(RETRY_LIMIT, 0);
+ conf.setBoolean(FS_S3A_CREATE_PERFORMANCE, true);
+ final Duration ms10 = Duration.ofMillis(10);
+ setDurationAsMillis(conf, CONNECTION_ACQUISITION_TIMEOUT, ms10);
+ setDurationAsMillis(conf, ESTABLISH_TIMEOUT, ms10);
+ return conf;
+ }
+
+ @Override
+ public void teardown() throws Exception {
+ AWSClientConfig.resetMinimumOperationDuration();
+ super.teardown();
+ }
+
+ /**
+ * Use up the connection pool and expect the failure to be handled.
+ */
+ @Test
+ public void testGeneratePoolTimeouts() throws Throwable {
+ byte[] data = dataset(FILE_SIZE, '0', 10);
+ AWSClientConfig.setMinimumOperationDuration(Duration.ZERO);
+ Configuration conf = timingOutConfiguration();
+ Path path = methodPath();
+ int streamsToCreate = 100;
+ List<FSDataInputStream> streams = new ArrayList<>(streamsToCreate);
+ final S3AFileSystem fs = getFileSystem();
+ // create the test file using the good fs, to avoid connection timeouts
+ // during setup.
+ ContractTestUtils.createFile(fs, path, true, data);
+ final FileStatus st = fs.getFileStatus(path);
+ try (FileSystem brittleFS = FileSystem.newInstance(fs.getUri(), conf)) {
+ intercept(ConnectTimeoutException.class, () -> {
+ for (int i = 0; i < streamsToCreate; i++) {
+ FutureDataInputStreamBuilder b = brittleFS.openFile(path);
+ b.withFileStatus(st);
+ b.opt(FS_OPTION_OPENFILE_READ_POLICY, FS_OPTION_OPENFILE_READ_POLICY_WHOLE_FILE);
+
+ final FSDataInputStream in = b.build().get();
+ streams.add(in);
+ // kick off the read so forcing a GET.
+ in.read();
+ }
+ });
+ } finally {
+ streams.forEach(s -> {
+ IOUtils.cleanupWithLogger(LOG, s);
+ });
+ }
+ }
+}
diff --git a/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/impl/TestAwsClientConfig.java b/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/impl/TestAwsClientConfig.java
new file mode 100644
index 000000000000..2a9e05755ce6
--- /dev/null
+++ b/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/impl/TestAwsClientConfig.java
@@ -0,0 +1,188 @@
+/*
+ * 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.impl;
+
+import java.time.Duration;
+import java.util.Arrays;
+
+import org.assertj.core.api.Assertions;
+import org.junit.After;
+import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.test.AbstractHadoopTestBase;
+
+import static org.apache.hadoop.fs.s3a.Constants.CONNECTION_ACQUISITION_TIMEOUT;
+import static org.apache.hadoop.fs.s3a.Constants.CONNECTION_IDLE_TIME;
+import static org.apache.hadoop.fs.s3a.Constants.CONNECTION_KEEPALIVE;
+import static org.apache.hadoop.fs.s3a.Constants.CONNECTION_TTL;
+import static org.apache.hadoop.fs.s3a.Constants.DEFAULT_CONNECTION_ACQUISITION_TIMEOUT_DURATION;
+import static org.apache.hadoop.fs.s3a.Constants.DEFAULT_CONNECTION_IDLE_TIME_DURATION;
+import static org.apache.hadoop.fs.s3a.Constants.DEFAULT_CONNECTION_KEEPALIVE;
+import static org.apache.hadoop.fs.s3a.Constants.DEFAULT_CONNECTION_TTL_DURATION;
+import static org.apache.hadoop.fs.s3a.Constants.DEFAULT_ESTABLISH_TIMEOUT_DURATION;
+import static org.apache.hadoop.fs.s3a.Constants.DEFAULT_MAXIMUM_CONNECTIONS;
+import static org.apache.hadoop.fs.s3a.Constants.DEFAULT_SOCKET_TIMEOUT_DURATION;
+import static org.apache.hadoop.fs.s3a.Constants.ESTABLISH_TIMEOUT;
+import static org.apache.hadoop.fs.s3a.Constants.MAXIMUM_CONNECTIONS;
+import static org.apache.hadoop.fs.s3a.Constants.MINIMUM_NETWORK_OPERATION_DURATION;
+import static org.apache.hadoop.fs.s3a.Constants.REQUEST_TIMEOUT;
+import static org.apache.hadoop.fs.s3a.Constants.SOCKET_TIMEOUT;
+import static org.apache.hadoop.fs.s3a.impl.AWSClientConfig.createApiConnectionSettings;
+import static org.apache.hadoop.fs.s3a.impl.AWSClientConfig.createConnectionSettings;
+import static org.apache.hadoop.fs.s3a.impl.ConfigurationHelper.enforceMinimumDuration;
+
+/**
+ * Unit tests for {@link AWSClientConfig}.
+ * These may play with the config timeout settings, so reset the timeouts
+ * during teardown.
+ * For isolation from any site settings, the tests create configurations
+ * without loading of defaut/site XML files.
+ */
+public class TestAwsClientConfig extends AbstractHadoopTestBase {
+
+ private static final Logger LOG = LoggerFactory.getLogger(TestAwsClientConfig.class);
+
+ @After
+ public void teardown() throws Exception {
+ AWSClientConfig.resetMinimumOperationDuration();
+ }
+
+ /**
+ * Create a new empty configuration.
+ * @return configuration.
+ */
+ private Configuration conf() {
+ return new Configuration(false);
+ }
+
+ /**
+ * Innermost duration enforcement, which is not applied if
+ * the minimum value is null.
+ */
+ @Test
+ public void testEnforceMinDuration() {
+ final Duration s10 = Duration.ofSeconds(10);
+ final Duration s1 = Duration.ofSeconds(1);
+
+ Assertions.assertThat(enforceMinimumDuration("key", s1, s10))
+ .describedAs("10s")
+ .isEqualTo(s10);
+
+ // and a null check
+ Assertions.assertThat(enforceMinimumDuration("key",
+ s1, null))
+ .describedAs("10s")
+ .isEqualTo(s1);
+ }
+
+ /**
+ * When loading a connection settings from an empty configuration, the
+ * correct default values are loaded.
+ */
+ @Test
+ public void testLoadUnsetValues() {
+ final AWSClientConfig.ConnectionSettings conn = createConnectionSettings(conf());
+ assertDuration(CONNECTION_ACQUISITION_TIMEOUT, DEFAULT_CONNECTION_ACQUISITION_TIMEOUT_DURATION,
+ conn.getAcquisitionTimeout());
+ assertDuration(CONNECTION_TTL, DEFAULT_CONNECTION_TTL_DURATION,
+ conn.getConnectionTTL());
+ assertDuration(CONNECTION_IDLE_TIME, DEFAULT_CONNECTION_IDLE_TIME_DURATION,
+ conn.getMaxIdleTime());
+ assertDuration(ESTABLISH_TIMEOUT, DEFAULT_ESTABLISH_TIMEOUT_DURATION,
+ conn.getEstablishTimeout());
+ assertDuration(SOCKET_TIMEOUT, DEFAULT_SOCKET_TIMEOUT_DURATION,
+ conn.getSocketTimeout());
+ Assertions.assertThat(conn.getMaxConnections())
+ .describedAs(MAXIMUM_CONNECTIONS)
+ .isEqualTo(DEFAULT_MAXIMUM_CONNECTIONS);
+ Assertions.assertThat(conn.isKeepAlive())
+ .describedAs(CONNECTION_KEEPALIVE)
+ .isEqualTo(DEFAULT_CONNECTION_KEEPALIVE);
+ }
+
+ /**
+ * If we set a minimum duration that is bigger than the configured value,
+ * the minimum value wins.
+ * Some options have a minimum value of zero.
+ */
+ @Test
+ public void testMinimumDurationWins() {
+
+ final Configuration conf = conf();
+ setOptionsToValue("1s", conf,
+ CONNECTION_ACQUISITION_TIMEOUT,
+ CONNECTION_TTL,
+ CONNECTION_IDLE_TIME,
+ ESTABLISH_TIMEOUT,
+ SOCKET_TIMEOUT);
+ final AWSClientConfig.ConnectionSettings conn = createConnectionSettings(conf);
+ LOG.info("Connection settings: {}", conn);
+ assertDuration(CONNECTION_ACQUISITION_TIMEOUT, MINIMUM_NETWORK_OPERATION_DURATION,
+ conn.getAcquisitionTimeout());
+
+ assertDuration(ESTABLISH_TIMEOUT, MINIMUM_NETWORK_OPERATION_DURATION,
+ conn.getEstablishTimeout());
+ assertDuration(SOCKET_TIMEOUT, MINIMUM_NETWORK_OPERATION_DURATION,
+ conn.getSocketTimeout());
+
+ // those options with a minimum of zero
+ final Duration s1 = Duration.ofSeconds(1);
+ assertDuration(CONNECTION_TTL, s1, conn.getConnectionTTL());
+ assertDuration(CONNECTION_IDLE_TIME, s1, conn.getMaxIdleTime());
+ }
+
+ /**
+ * Assert that a a duration has the expected value.
+ * @param name option name for assertion text
+ * @param expected expected duration
+ * @param actual actual duration
+ */
+ private void assertDuration(String name, Duration expected, Duration actual) {
+ Assertions.assertThat(actual)
+ .describedAs("Duration of %s", name)
+ .isEqualTo(expected);
+ }
+
+ /**
+ * Test {@link AWSClientConfig#createApiConnectionSettings(Configuration)}.
+ */
+ @Test
+ public void testCreateApiConnectionSettings() {
+ final Configuration conf = conf();
+ conf.set(REQUEST_TIMEOUT, "1h");
+ final AWSClientConfig.ClientSettings settings =
+ createApiConnectionSettings(conf);
+ Assertions.assertThat(settings.getApiCallTimeout())
+ .describedAs("%s in %s", REQUEST_TIMEOUT, settings)
+ .isEqualTo(Duration.ofHours(1));
+ }
+
+ /**
+ * Set a list of keys to the same value.
+ * @param value value to set
+ * @param conf configuration to patch
+ * @param keys keys
+ */
+ private void setOptionsToValue(String value, Configuration conf, String... keys) {
+ Arrays.stream(keys).forEach(key -> conf.set(key, value));
+ }
+}
diff --git a/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/performance/ITestUnbufferDraining.java b/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/performance/ITestUnbufferDraining.java
index fb75560a8075..b77ca97c7ddf 100644
--- a/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/performance/ITestUnbufferDraining.java
+++ b/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/performance/ITestUnbufferDraining.java
@@ -19,6 +19,7 @@
package org.apache.hadoop.fs.s3a.performance;
import java.io.IOException;
+import java.time.Duration;
import org.assertj.core.api.Assertions;
import org.junit.Test;
@@ -34,6 +35,7 @@ import org.apache.hadoop.fs.contract.ContractTestUtils;
import org.apache.hadoop.fs.s3a.S3AFileSystem;
import org.apache.hadoop.fs.s3a.S3AInputPolicy;
import org.apache.hadoop.fs.s3a.S3AInputStream;
+import org.apache.hadoop.fs.s3a.impl.AWSClientConfig;
import org.apache.hadoop.fs.statistics.IOStatistics;
import org.apache.hadoop.io.IOUtils;
@@ -51,6 +53,7 @@ import static org.apache.hadoop.fs.s3a.Constants.REQUEST_TIMEOUT;
import static org.apache.hadoop.fs.s3a.Constants.RETRY_LIMIT;
import static org.apache.hadoop.fs.s3a.Constants.SOCKET_TIMEOUT;
import static org.apache.hadoop.fs.s3a.S3ATestUtils.removeBaseAndBucketOverrides;
+import static org.apache.hadoop.fs.s3a.impl.ConfigurationHelper.setDurationAsSeconds;
import static org.apache.hadoop.fs.statistics.IOStatisticAssertions.verifyStatisticCounterValue;
import static org.apache.hadoop.fs.statistics.IOStatisticsSupport.retrieveIOStatistics;
import static org.apache.hadoop.fs.statistics.StreamStatisticNames.STREAM_READ_ABORTED;
@@ -119,16 +122,23 @@ public class ITestUnbufferDraining extends AbstractS3ACostTest {
// now create a new FS with minimal http capacity and recovery
// a separate one is used to avoid test teardown suffering
// from the lack of http connections and short timeouts.
- Configuration conf = getConfiguration();
- // kick off async drain for any data
- conf.setInt(ASYNC_DRAIN_THRESHOLD, 1);
- conf.setInt(MAXIMUM_CONNECTIONS, 2);
- conf.setInt(MAX_ERROR_RETRIES, 1);
- conf.setInt(ESTABLISH_TIMEOUT, 1000);
- conf.setInt(READAHEAD_RANGE, READAHEAD);
- conf.setInt(RETRY_LIMIT, 1);
-
- brittleFS = FileSystem.newInstance(getFileSystem().getUri(), conf);
+ try {
+ // allow small durations.
+ AWSClientConfig.setMinimumOperationDuration(Duration.ZERO);
+ Configuration conf = getConfiguration();
+ // kick off async drain for any data
+ conf.setInt(ASYNC_DRAIN_THRESHOLD, 1);
+ conf.setInt(MAXIMUM_CONNECTIONS, 2);
+ conf.setInt(MAX_ERROR_RETRIES, 1);
+ conf.setInt(READAHEAD_RANGE, READAHEAD);
+ conf.setInt(RETRY_LIMIT, 1);
+ setDurationAsSeconds(conf, ESTABLISH_TIMEOUT,
+ Duration.ofSeconds(1));
+
+ brittleFS = FileSystem.newInstance(getFileSystem().getUri(), conf);
+ } finally {
+ AWSClientConfig.resetMinimumOperationDuration();
+ }
}
@Override
---------------------------------------------------------------------
To unsubscribe, e-mail: common-commits-unsubscribe@hadoop.apache.org
For additional commands, e-mail: common-commits-help@hadoop.apache.org
(hadoop) 03/05: HADOOP-18925. S3A: option to enable/disable CopyFromLocalOperation (#6163)
Posted by ah...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
ahmar pushed a commit to branch feature-HADOOP-18073-sdk-v2-upgrade-3.4
in repository https://gitbox.apache.org/repos/asf/hadoop.git
commit b327fcac96d1fdb68fb09b9271deb26f2b0f324b
Author: Steve Loughran <st...@cloudera.com>
AuthorDate: Mon Nov 6 16:00:56 2023 +0000
HADOOP-18925. S3A: option to enable/disable CopyFromLocalOperation (#6163)
Add a new option:
fs.s3a.optimized.copy.from.local.enabled
This will enable (default) or disable the
optimized CopyFromLocalOperation upload operation
when copyFromLocalFile() is invoked.
When false the superclass implementation is used; duration
statistics are still collected, though audit span entries
in logs will be for the individual fs operations, not the
overall operation.
Contributed by Steve Loughran
---
.../java/org/apache/hadoop/fs/s3a/Constants.java | 13 +++
.../org/apache/hadoop/fs/s3a/S3AFileSystem.java | 93 +++++++++++++++-------
.../tools/hadoop-aws/troubleshooting_s3a.md | 10 +++
.../hadoop/fs/s3a/ITestS3ACopyFromLocalFile.java | 51 ++++++++++++
4 files changed, 138 insertions(+), 29 deletions(-)
diff --git a/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/Constants.java b/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/Constants.java
index e602c1b3dc36..140ea53ceb6f 100644
--- a/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/Constants.java
+++ b/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/Constants.java
@@ -1496,4 +1496,17 @@ public final class Constants {
*/
public static final boolean DIRECTORY_OPERATIONS_PURGE_UPLOADS_DEFAULT = false;
+
+ /**
+ * Is the higher performance copy from local file to S3 enabled?
+ * This switch allows for it to be disabled if there are problems.
+ * Value: {@value}.
+ */
+ public static final String OPTIMIZED_COPY_FROM_LOCAL = "fs.s3a.optimized.copy.from.local.enabled";
+
+ /**
+ * Default value for {@link #OPTIMIZED_COPY_FROM_LOCAL}.
+ * Value: {@value}.
+ */
+ public static final boolean OPTIMIZED_COPY_FROM_LOCAL_DEFAULT = true;
}
diff --git a/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/S3AFileSystem.java b/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/S3AFileSystem.java
index cdcb0c9cb148..3e995fb1ddb4 100644
--- a/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/S3AFileSystem.java
+++ b/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/S3AFileSystem.java
@@ -464,6 +464,12 @@ public class S3AFileSystem extends FileSystem implements StreamCapabilities,
*/
private String scheme = FS_S3A;
+ /**
+ * Flag to indicate that the higher performance copyFromLocalFile implementation
+ * should be used.
+ */
+ private boolean optimizedCopyFromLocal;
+
/** Add any deprecated keys. */
@SuppressWarnings("deprecation")
private static void addDeprecatedKeys() {
@@ -700,6 +706,9 @@ public class S3AFileSystem extends FileSystem implements StreamCapabilities,
AWS_S3_VECTOR_ACTIVE_RANGE_READS, DEFAULT_AWS_S3_VECTOR_ACTIVE_RANGE_READS, 1);
vectoredIOContext = populateVectoredIOContext(conf);
scheme = (this.uri != null && this.uri.getScheme() != null) ? this.uri.getScheme() : FS_S3A;
+ optimizedCopyFromLocal = conf.getBoolean(OPTIMIZED_COPY_FROM_LOCAL,
+ OPTIMIZED_COPY_FROM_LOCAL_DEFAULT);
+ LOG.debug("Using optimized copyFromLocal implementation: {}", optimizedCopyFromLocal);
} catch (SdkException e) {
// amazon client exception: stop all services then throw the translation
cleanupWithLogger(LOG, span);
@@ -4028,9 +4037,9 @@ public class S3AFileSystem extends FileSystem implements StreamCapabilities,
* the given dst name.
*
* This version doesn't need to create a temporary file to calculate the md5.
- * Sadly this doesn't seem to be used by the shell cp :(
+ * If {@link Constants#OPTIMIZED_COPY_FROM_LOCAL} is set to false,
+ * the superclass implementation is used.
*
- * delSrc indicates if the source should be removed
* @param delSrc whether to delete the src
* @param overwrite whether to overwrite an existing file
* @param src path
@@ -4038,35 +4047,59 @@ public class S3AFileSystem extends FileSystem implements StreamCapabilities,
* @throws IOException IO problem
* @throws FileAlreadyExistsException the destination file exists and
* overwrite==false
- * @throws SdkException failure in the AWS SDK
*/
@Override
@AuditEntryPoint
public void copyFromLocalFile(boolean delSrc, boolean overwrite, Path src,
Path dst) throws IOException {
checkNotClosed();
- LOG.debug("Copying local file from {} to {}", src, dst);
- trackDurationAndSpan(INVOCATION_COPY_FROM_LOCAL_FILE, dst,
- () -> new CopyFromLocalOperation(
- createStoreContext(),
- src,
- dst,
- delSrc,
- overwrite,
- createCopyFromLocalCallbacks()).execute());
+ LOG.debug("Copying local file from {} to {} (delSrc={} overwrite={}",
+ src, dst, delSrc, overwrite);
+ if (optimizedCopyFromLocal) {
+ trackDurationAndSpan(INVOCATION_COPY_FROM_LOCAL_FILE, dst, () ->
+ new CopyFromLocalOperation(
+ createStoreContext(),
+ src,
+ dst,
+ delSrc,
+ overwrite,
+ createCopyFromLocalCallbacks(getActiveAuditSpan()))
+ .execute());
+ } else {
+ // call the superclass, but still count statistics.
+ // there is no overall span here, as each FS API call will
+ // be in its own span.
+ LOG.debug("Using base copyFromLocalFile implementation");
+ trackDurationAndSpan(INVOCATION_COPY_FROM_LOCAL_FILE, dst, () -> {
+ super.copyFromLocalFile(delSrc, overwrite, src, dst);
+ return null;
+ });
+ }
}
+ /**
+ * Create the CopyFromLocalCallbacks;
+ * protected to assist in mocking.
+ * @param span audit span.
+ * @return the callbacks
+ * @throws IOException failure to get the local fs.
+ */
protected CopyFromLocalOperation.CopyFromLocalOperationCallbacks
- createCopyFromLocalCallbacks() throws IOException {
+ createCopyFromLocalCallbacks(final AuditSpanS3A span) throws IOException {
LocalFileSystem local = getLocal(getConf());
- return new CopyFromLocalCallbacksImpl(local);
+ return new CopyFromLocalCallbacksImpl(span, local);
}
protected final class CopyFromLocalCallbacksImpl implements
CopyFromLocalOperation.CopyFromLocalOperationCallbacks {
+
+ /** Span to use for all operations. */
+ private final AuditSpanS3A span;
private final LocalFileSystem local;
- private CopyFromLocalCallbacksImpl(LocalFileSystem local) {
+ private CopyFromLocalCallbacksImpl(final AuditSpanS3A span,
+ LocalFileSystem local) {
+ this.span = span;
this.local = local;
}
@@ -4088,20 +4121,18 @@ public class S3AFileSystem extends FileSystem implements StreamCapabilities,
@Override
public void copyLocalFileFromTo(File file, Path from, Path to) throws IOException {
- trackDurationAndSpan(
- OBJECT_PUT_REQUESTS,
- to,
- () -> {
- final String key = pathToKey(to);
- Progressable progress = null;
- PutObjectRequest.Builder putObjectRequestBuilder =
- newPutObjectRequestBuilder(key, file.length(), false);
- S3AFileSystem.this.invoker.retry("putObject(" + "" + ")", to.toString(), true,
- () -> executePut(putObjectRequestBuilder.build(), progress, putOptionsForPath(to),
- file));
-
- return null;
- });
+ // the duration of the put is measured, but the active span is the
+ // constructor-supplied one -this ensures all audit log events are grouped correctly
+ span.activate();
+ trackDuration(getDurationTrackerFactory(), OBJECT_PUT_REQUESTS.getSymbol(), () -> {
+ final String key = pathToKey(to);
+ PutObjectRequest.Builder putObjectRequestBuilder =
+ newPutObjectRequestBuilder(key, file.length(), false);
+ final String dest = to.toString();
+ S3AFileSystem.this.invoker.retry("putObject(" + dest + ")", dest, true, () ->
+ executePut(putObjectRequestBuilder.build(), null, putOptionsForPath(to), file));
+ return null;
+ });
}
@Override
@@ -5403,6 +5434,10 @@ public class S3AFileSystem extends FileSystem implements StreamCapabilities,
case FS_S3A_CREATE_PERFORMANCE_ENABLED:
return performanceCreation;
+ // is the optimized copy from local enabled.
+ case OPTIMIZED_COPY_FROM_LOCAL:
+ return optimizedCopyFromLocal;
+
default:
return super.hasPathCapability(p, cap);
}
diff --git a/hadoop-tools/hadoop-aws/src/site/markdown/tools/hadoop-aws/troubleshooting_s3a.md b/hadoop-tools/hadoop-aws/src/site/markdown/tools/hadoop-aws/troubleshooting_s3a.md
index 8c57db1faf4c..be359336463a 100644
--- a/hadoop-tools/hadoop-aws/src/site/markdown/tools/hadoop-aws/troubleshooting_s3a.md
+++ b/hadoop-tools/hadoop-aws/src/site/markdown/tools/hadoop-aws/troubleshooting_s3a.md
@@ -1544,3 +1544,13 @@ software.amazon.awssdk.core.exception.SdkClientException: Unable to execute HTTP
When this happens, try to set `fs.s3a.connection.request.timeout` to a larger value or disable it
completely by setting it to `0`.
+
+### <a name="debug-switches"></a> Debugging Switches
+
+There are some switches which can be set to enable/disable features and assist
+in isolating problems and at least make them "go away".
+
+
+| Key | Default | Action |
+|------|---------|----------|
+| `fs.s3a.optimized.copy.from.local.enabled` | `true` | [HADOOP-18925](https://issues.apache.org/jira/browse/HADOOP-18925) enable/disable CopyFromLocalOperation. Also a path capability. |
diff --git a/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/ITestS3ACopyFromLocalFile.java b/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/ITestS3ACopyFromLocalFile.java
index dfac771dd78d..f9600de6d20c 100644
--- a/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/ITestS3ACopyFromLocalFile.java
+++ b/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/ITestS3ACopyFromLocalFile.java
@@ -19,6 +19,8 @@
package org.apache.hadoop.fs.s3a;
import java.io.File;
+import java.util.Arrays;
+import java.util.Collection;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.contract.AbstractContractCopyFromLocalTest;
@@ -26,18 +28,67 @@ import org.apache.hadoop.fs.contract.AbstractFSContract;
import org.apache.hadoop.fs.contract.s3a.S3AContract;
import org.apache.hadoop.fs.Path;
+
+import org.assertj.core.api.Assertions;
import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+import static org.apache.hadoop.fs.s3a.Constants.OPTIMIZED_COPY_FROM_LOCAL;
+import static org.apache.hadoop.fs.s3a.S3ATestUtils.disableFilesystemCaching;
+import static org.apache.hadoop.fs.s3a.S3ATestUtils.getTestBucketName;
+import static org.apache.hadoop.fs.s3a.S3ATestUtils.removeBaseAndBucketOverrides;
import static org.apache.hadoop.test.LambdaTestUtils.intercept;
+/**
+ * Test copying files from the local filesystem to S3A.
+ * Parameterized on whether or not the optimized
+ * copyFromLocalFile is enabled.
+ */
+@RunWith(Parameterized.class)
public class ITestS3ACopyFromLocalFile extends
AbstractContractCopyFromLocalTest {
+ /**
+ * Parameterization.
+ */
+ @Parameterized.Parameters(name = "enabled={0}")
+ public static Collection<Object[]> params() {
+ return Arrays.asList(new Object[][]{
+ {true},
+ {false},
+ });
+ }
+ private final boolean enabled;
+
+ public ITestS3ACopyFromLocalFile(final boolean enabled) {
+ this.enabled = enabled;
+ }
+
+ @Override
+ protected Configuration createConfiguration() {
+ final Configuration conf = super.createConfiguration();
+
+ removeBaseAndBucketOverrides(getTestBucketName(conf), conf,
+ OPTIMIZED_COPY_FROM_LOCAL);
+ conf.setBoolean(OPTIMIZED_COPY_FROM_LOCAL, enabled);
+ disableFilesystemCaching(conf);
+ return conf;
+ }
@Override
protected AbstractFSContract createContract(Configuration conf) {
return new S3AContract(conf);
}
+ @Test
+ public void testOptionPropagation() throws Throwable {
+ Assertions.assertThat(getFileSystem().hasPathCapability(new Path("/"),
+ OPTIMIZED_COPY_FROM_LOCAL))
+ .describedAs("path capability of %s", OPTIMIZED_COPY_FROM_LOCAL)
+ .isEqualTo(enabled);
+
+ }
+
@Test
public void testLocalFilesOnly() throws Throwable {
describe("Copying into other file systems must fail");
---------------------------------------------------------------------
To unsubscribe, e-mail: common-commits-unsubscribe@hadoop.apache.org
For additional commands, e-mail: common-commits-help@hadoop.apache.org
(hadoop) 04/05: S3A: Upgrade AWS SDK version to 2.21.33 for Amazon S3 Express One Zone support (#6306)
Posted by ah...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
ahmar pushed a commit to branch feature-HADOOP-18073-sdk-v2-upgrade-3.4
in repository https://gitbox.apache.org/repos/asf/hadoop.git
commit 8f4fd04722106ac6570441956907dda9813e3a55
Author: ahmarsuhail <ah...@amazon.co.uk>
AuthorDate: Wed Nov 29 13:16:19 2023 +0000
S3A: Upgrade AWS SDK version to 2.21.33 for Amazon S3 Express One Zone support (#6306)
With this upgrade, it is possible to connect to an Amazon S3 Express One Zone bucket.
Some tests from the S3A test suite will currently fail against a one zone bucket, as one zone buckets
do not support some S3 standard features (eg: SSE-KMS), and certain operations behave slightly
differently (eg: listMPU will return a directory that has incomplete MPUs).
Contributed by Ahmar Suhail
---
LICENSE-binary | 2 +-
hadoop-project/pom.xml | 2 +-
2 files changed, 2 insertions(+), 2 deletions(-)
diff --git a/LICENSE-binary b/LICENSE-binary
index 2df6cf730b7a..3279137509ef 100644
--- a/LICENSE-binary
+++ b/LICENSE-binary
@@ -364,7 +364,7 @@ org.objenesis:objenesis:2.6
org.xerial.snappy:snappy-java:1.1.10.4
org.yaml:snakeyaml:2.0
org.wildfly.openssl:wildfly-openssl:1.1.3.Final
-software.amazon.awssdk:bundle:jar:2.20.160
+software.amazon.awssdk:bundle:jar:2.21.33
--------------------------------------------------------------------------------
diff --git a/hadoop-project/pom.xml b/hadoop-project/pom.xml
index 1c913c9a6fc0..a3fd8b824e5b 100644
--- a/hadoop-project/pom.xml
+++ b/hadoop-project/pom.xml
@@ -192,7 +192,7 @@
<make-maven-plugin.version>1.0-beta-1</make-maven-plugin.version>
<surefire.fork.timeout>900</surefire.fork.timeout>
<aws-java-sdk.version>1.12.565</aws-java-sdk.version>
- <aws-java-sdk-v2.version>2.20.160</aws-java-sdk-v2.version>
+ <aws-java-sdk-v2.version>2.21.33</aws-java-sdk-v2.version>
<aws.eventstream.version>1.0.1</aws.eventstream.version>
<hsqldb.version>2.7.1</hsqldb.version>
<frontend-maven-plugin.version>1.11.2</frontend-maven-plugin.version>
---------------------------------------------------------------------
To unsubscribe, e-mail: common-commits-unsubscribe@hadoop.apache.org
For additional commands, e-mail: common-commits-help@hadoop.apache.org