This is an automated email from the ASF dual-hosted git repository.
stevel pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/hadoop.git
The following commit(s) were added to refs/heads/trunk by this push:
new 120620c1b7f HADOOP-18888. S3A. createS3AsyncClient() always enables
multipart uploads (#6056)
120620c1b7f is described below
commit 120620c1b7f0a11e140a7d80d2bf7be4c5c83c1b
Author: Steve Loughran <ste...@cloudera.com>
AuthorDate: Fri Sep 15 15:45:09 2023 +0100
HADOOP-18888. S3A. createS3AsyncClient() always enables multipart uploads
(#6056)
* The multipart flag fs.s3a.multipart.uploads.enabled is passed to the
async client created
* s3A connector bypasses the transfer manager entirely if disabled or for
small files.
Contributed by Steve Loughran
---
.../hadoop/fs/s3a/DefaultS3ClientFactory.java | 2 +-
.../org/apache/hadoop/fs/s3a/S3AFileSystem.java | 92 +++++++++++++++-------
.../org/apache/hadoop/fs/s3a/S3AInternals.java | 6 ++
.../org/apache/hadoop/fs/s3a/S3ClientFactory.java | 24 ++++++
.../fs/s3a/scale/AbstractSTestS3AHugeFiles.java | 4 +-
.../fs/s3a/scale/ITestS3AHugeFilesNoMultipart.java | 53 +++++--------
6 files changed, 117 insertions(+), 64 deletions(-)
diff --git
a/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/DefaultS3ClientFactory.java
b/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/DefaultS3ClientFactory.java
index 98c72d27662..c85263f1903 100644
---
a/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/DefaultS3ClientFactory.java
+++
b/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/DefaultS3ClientFactory.java
@@ -112,7 +112,7 @@ public class DefaultS3ClientFactory extends Configured
return configureClientBuilder(S3AsyncClient.builder(), parameters, conf,
bucket)
.httpClientBuilder(httpClientBuilder)
.multipartConfiguration(multipartConfiguration)
- .multipartEnabled(true)
+ .multipartEnabled(parameters.isMultipartCopy())
.build();
}
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 e192135b9f3..9307c4c2650 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
@@ -440,6 +440,12 @@ public class S3AFileSystem extends FileSystem implements
StreamCapabilities,
*/
private boolean isMultipartUploadEnabled = DEFAULT_MULTIPART_UPLOAD_ENABLED;
+ /**
+ * Should file copy operations use the S3 transfer manager?
+ * True unless multipart upload is disabled.
+ */
+ private boolean isMultipartCopyEnabled;
+
/**
* A cache of files that should be deleted when the FileSystem is closed
* or the JVM is exited.
@@ -576,6 +582,9 @@ public class S3AFileSystem extends FileSystem implements
StreamCapabilities,
intOption(conf, PREFETCH_BLOCK_COUNT_KEY,
PREFETCH_BLOCK_DEFAULT_COUNT, 1);
this.isMultipartUploadEnabled =
conf.getBoolean(MULTIPART_UPLOADS_ENABLED,
DEFAULT_MULTIPART_UPLOAD_ENABLED);
+ // multipart copy and upload are the same; this just makes it explicit
+ this.isMultipartCopyEnabled = isMultipartUploadEnabled;
+
initThreadPools(conf);
int listVersion = conf.getInt(LIST_VERSION, DEFAULT_LIST_VERSION);
@@ -982,6 +991,7 @@ public class S3AFileSystem extends FileSystem implements
StreamCapabilities,
.withRequesterPays(conf.getBoolean(ALLOW_REQUESTER_PAYS,
DEFAULT_ALLOW_REQUESTER_PAYS))
.withExecutionInterceptors(auditManager.createExecutionInterceptors())
.withMinimumPartSize(partSize)
+ .withMultipartCopyEnabled(isMultipartCopyEnabled)
.withMultipartThreshold(multiPartThreshold)
.withTransferManagerExecutor(unboundedThreadPool)
.withRegion(region);
@@ -1468,6 +1478,11 @@ public class S3AFileSystem extends FileSystem implements
StreamCapabilities,
LOG.debug("Sharing credentials for: {}", purpose);
return credentials.share();
}
+
+ @Override
+ public boolean isMultipartCopyEnabled() {
+ return S3AFileSystem.this.isMultipartUploadEnabled;
+ }
}
/**
@@ -4436,37 +4451,56 @@ public class S3AFileSystem extends FileSystem
implements StreamCapabilities,
e);
}
- return readInvoker.retry(
- action, srcKey,
- true,
- () -> {
- CopyObjectRequest.Builder copyObjectRequestBuilder =
- getRequestFactory().newCopyObjectRequestBuilder(srcKey, dstKey,
srcom);
- changeTracker.maybeApplyConstraint(copyObjectRequestBuilder);
- incrementStatistic(OBJECT_COPY_REQUESTS);
-
- Copy copy = transferManager.copy(
- CopyRequest.builder()
- .copyObjectRequest(copyObjectRequestBuilder.build())
- .build());
+ CopyObjectRequest.Builder copyObjectRequestBuilder =
+ getRequestFactory().newCopyObjectRequestBuilder(srcKey, dstKey, srcom);
+ changeTracker.maybeApplyConstraint(copyObjectRequestBuilder);
+ CopyObjectResponse response;
- try {
- CompletedCopy completedCopy = copy.completionFuture().join();
- CopyObjectResponse result = completedCopy.response();
- changeTracker.processResponse(result);
- incrementWriteOperations();
- instrumentation.filesCopied(1, size);
- return result;
- } catch (CompletionException e) {
- Throwable cause = e.getCause();
- if (cause instanceof SdkException) {
- SdkException awsException = (SdkException)cause;
- changeTracker.processException(awsException, "copy");
- throw awsException;
+ // transfer manager is skipped if disabled or the file is too small to
worry about
+ final boolean useTransferManager = isMultipartCopyEnabled && size >=
multiPartThreshold;
+ if (useTransferManager) {
+ // use transfer manager
+ response = readInvoker.retry(
+ action, srcKey,
+ true,
+ () -> {
+ incrementStatistic(OBJECT_COPY_REQUESTS);
+
+ Copy copy = transferManager.copy(
+ CopyRequest.builder()
+ .copyObjectRequest(copyObjectRequestBuilder.build())
+ .build());
+
+ try {
+ CompletedCopy completedCopy = copy.completionFuture().join();
+ return completedCopy.response();
+ } catch (CompletionException e) {
+ Throwable cause = e.getCause();
+ if (cause instanceof SdkException) {
+ SdkException awsException = (SdkException)cause;
+ changeTracker.processException(awsException, "copy");
+ throw awsException;
+ }
+ throw extractException(action, srcKey, e);
}
- throw extractException(action, srcKey, e);
- }
- });
+ });
+ } else {
+ // single part copy bypasses transfer manager
+ // note, this helps with some mock testing, e.g. HBoss. as there is less
to mock.
+ response = readInvoker.retry(
+ action, srcKey,
+ true,
+ () -> {
+ LOG.debug("copyFile: single part copy {} -> {} of size {}",
srcKey, dstKey, size);
+ incrementStatistic(OBJECT_COPY_REQUESTS);
+ return s3Client.copyObject(copyObjectRequestBuilder.build());
+ });
+ }
+
+ changeTracker.processResponse(response);
+ incrementWriteOperations();
+ instrumentation.filesCopied(1, size);
+ return response;
}
/**
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 23c4d350120..18d6c1af586 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
@@ -115,4 +115,10 @@ public interface S3AInternals {
@AuditEntryPoint
@Retries.RetryTranslated
HeadBucketResponse getBucketMetadata() throws IOException;
+
+ /**
+ * Is multipart copy enabled?
+ * @return true if the transfer manager is used to copy files.
+ */
+ boolean isMultipartCopyEnabled();
}
diff --git
a/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/S3ClientFactory.java
b/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/S3ClientFactory.java
index d4504cd08d7..e2e792ebfb6 100644
---
a/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/S3ClientFactory.java
+++
b/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/S3ClientFactory.java
@@ -156,6 +156,11 @@ public interface S3ClientFactory {
*/
private long multiPartThreshold;
+ /**
+ * Multipart upload enabled.
+ */
+ private boolean multipartCopy = true;
+
/**
* Executor that the transfer manager will use to execute background tasks.
*/
@@ -399,5 +404,24 @@ public interface S3ClientFactory {
public Region getRegion() {
return region;
}
+
+ /**
+ * Set the multipart flag..
+ *
+ * @param value new value
+ * @return the builder
+ */
+ public S3ClientCreationParameters withMultipartCopyEnabled(final boolean
value) {
+ this.multipartCopy = value;
+ return this;
+ }
+
+ /**
+ * Get the multipart flag.
+ * @return multipart flag
+ */
+ public boolean isMultipartCopy() {
+ return multipartCopy;
+ }
}
}
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 1a30c04358b..75c6efbe2ab 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
@@ -355,10 +355,10 @@ public abstract class AbstractSTestS3AHugeFiles extends
S3AScaleTestBase {
/**
* Is this expected to be a multipart upload?
* Assertions will change if not.
- * @return true by default.
+ * @return what the filesystem expects.
*/
protected boolean expectMultipartUpload() {
- return true;
+ return getFileSystem().getS3AInternals().isMultipartCopyEnabled();
}
/**
diff --git
a/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/scale/ITestS3AHugeFilesNoMultipart.java
b/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/scale/ITestS3AHugeFilesNoMultipart.java
index ed300dba01e..e154ab5676f 100644
---
a/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/scale/ITestS3AHugeFilesNoMultipart.java
+++
b/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/scale/ITestS3AHugeFilesNoMultipart.java
@@ -18,13 +18,10 @@
package org.apache.hadoop.fs.s3a.scale;
+import org.assertj.core.api.Assertions;
+
import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.fs.FileSystem;
-import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.s3a.Constants;
-import org.apache.hadoop.fs.s3a.S3AFileSystem;
-import org.apache.hadoop.fs.s3a.S3ATestUtils;
-import org.apache.hadoop.fs.s3a.api.UnsupportedRequestException;
import static
org.apache.hadoop.fs.contract.ContractTestUtils.IO_CHUNK_BUFFER_SIZE;
import static org.apache.hadoop.fs.s3a.Constants.MIN_MULTIPART_THRESHOLD;
@@ -33,7 +30,6 @@ import static
org.apache.hadoop.fs.s3a.Constants.MULTIPART_SIZE;
import static org.apache.hadoop.fs.s3a.Constants.MULTIPART_UPLOADS_ENABLED;
import static org.apache.hadoop.fs.s3a.Constants.REQUEST_TIMEOUT;
import static
org.apache.hadoop.fs.s3a.S3ATestUtils.removeBaseAndBucketOverrides;
-import static org.apache.hadoop.test.LambdaTestUtils.intercept;
/**
* Use a single PUT for the whole upload/rename/delete workflow; include
verification
@@ -41,11 +37,6 @@ import static
org.apache.hadoop.test.LambdaTestUtils.intercept;
*/
public class ITestS3AHugeFilesNoMultipart extends AbstractSTestS3AHugeFiles {
- /**
- * Size to ensure MPUs don't happen in transfer manager.
- */
- public static final String S_1T = "1T";
-
public static final String SINGLE_PUT_REQUEST_TIMEOUT = "1h";
/**
@@ -56,11 +47,23 @@ public class ITestS3AHugeFilesNoMultipart extends
AbstractSTestS3AHugeFiles {
return Constants.FAST_UPLOAD_BUFFER_DISK;
}
+ /**
+ * Multipart upload is always disabled.
+ * @return false
+ */
@Override
protected boolean expectMultipartUpload() {
return false;
}
+ /**
+ * Is multipart copy enabled?
+ * @return true if the transfer manager is used to copy files.
+ */
+ private boolean isMultipartCopyEnabled() {
+ return getFileSystem().getS3AInternals().isMultipartCopyEnabled();
+ }
+
/**
* Create a configuration without multipart upload,
* and a long request timeout to allow for a very slow
@@ -77,35 +80,21 @@ public class ITestS3AHugeFilesNoMultipart extends
AbstractSTestS3AHugeFiles {
MULTIPART_SIZE,
REQUEST_TIMEOUT);
conf.setInt(IO_CHUNK_BUFFER_SIZE, 655360);
- conf.set(MIN_MULTIPART_THRESHOLD, S_1T);
- conf.set(MULTIPART_SIZE, S_1T);
+ conf.setInt(MIN_MULTIPART_THRESHOLD, MULTIPART_MIN_SIZE);
+ conf.setInt(MULTIPART_SIZE, MULTIPART_MIN_SIZE);
conf.setBoolean(MULTIPART_UPLOADS_ENABLED, false);
conf.set(REQUEST_TIMEOUT, SINGLE_PUT_REQUEST_TIMEOUT);
return conf;
}
/**
- * After the file is created, attempt a rename with an FS
- * instance with a small multipart threshold;
- * this MUST be rejected.
+ * Verify multipart copy is disabled.
*/
@Override
public void test_030_postCreationAssertions() throws Throwable {
- assumeHugeFileExists();
- final Path hugefile = getHugefile();
- final Path hugefileRenamed = getHugefileRenamed();
- describe("renaming %s to %s", hugefile, hugefileRenamed);
- S3AFileSystem fs = getFileSystem();
- fs.delete(hugefileRenamed, false);
- // create a new fs with a small multipart threshold; expect rename failure.
- final Configuration conf = new Configuration(fs.getConf());
- conf.setInt(MIN_MULTIPART_THRESHOLD, MULTIPART_MIN_SIZE);
- conf.setInt(MULTIPART_SIZE, MULTIPART_MIN_SIZE);
- S3ATestUtils.disableFilesystemCaching(conf);
-
- try (FileSystem fs2 = FileSystem.get(fs.getUri(), conf)) {
- intercept(UnsupportedRequestException.class, () ->
- fs2.rename(hugefile, hugefileRenamed));
- }
+ super.test_030_postCreationAssertions();
+ Assertions.assertThat(isMultipartCopyEnabled())
+ .describedAs("Multipart copy should be disabled in %s",
getFileSystem())
+ .isFalse();
}
}
---------------------------------------------------------------------
To unsubscribe, e-mail: common-commits-unsubscr...@hadoop.apache.org
For additional commands, e-mail: common-commits-h...@hadoop.apache.org
