getUploadsAborted() {
+ return uploadsAborted;
+ }
+
/**
* Delete a file or directory tree.
*
@@ -236,6 +260,17 @@ protected void deleteDirectoryTree(final Path path,
try (DurationInfo ignored =
new DurationInfo(LOG, false, "deleting %s", dirKey)) {
+ final CompletableFuture abortUploads;
+ if (dirOperationsPurgeUploads) {
+ final StoreContext sc = getStoreContext();
+ final String key = sc.pathToKey(path) + "/";
+ LOG.debug("All uploads under {} will be deleted", key);
+ abortUploads = submit(sc.getExecutor(), sc.getActiveAuditSpan(), () ->
+ callbacks.abortMultipartUploadsUnderPrefix(key));
+ } else {
+ abortUploads = null;
+ }
+
// init the lists of keys and paths to delete
resetDeleteList();
deleteFuture = null;
@@ -257,10 +292,10 @@ protected void deleteDirectoryTree(final Path path,
LOG.debug("Deleting final batch of listed files");
submitNextBatch();
maybeAwaitCompletion(deleteFuture);
-
+ uploadsAborted = waitForCompletionIgnoringExceptions(abortUploads);
}
- LOG.debug("Delete \"{}\" completed; deleted {} objects", path,
- filesDeleted);
+ LOG.debug("Delete \"{}\" completed; deleted {} objects and aborted {} uploads", path,
+ filesDeleted, uploadsAborted.orElse(0L));
}
/**
@@ -313,7 +348,8 @@ private void submitNextBatch()
throws IOException {
// delete a single page of keys and the metadata.
// block for any previous batch.
- maybeAwaitCompletion(deleteFuture);
+ maybeAwaitCompletion(deleteFuture).ifPresent(count ->
+ LOG.debug("Deleted {} uploads", count));
// delete the current page of keys and paths
deleteFuture = submitDelete(keys);
diff --git a/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/impl/OperationCallbacks.java b/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/impl/OperationCallbacks.java
index e0d9c7c6aada70..9c88870633a358 100644
--- a/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/impl/OperationCallbacks.java
+++ b/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/impl/OperationCallbacks.java
@@ -164,4 +164,16 @@ RemoteIterator listObjects(
Path path,
String key)
throws IOException;
+
+ /**
+ * Abort multipart uploads under a path; paged.
+ * @param prefix prefix for uploads to abort
+ * @return a count of aborts
+ * @throws IOException trouble; FileNotFoundExceptions are swallowed.
+ */
+ @Retries.RetryTranslated
+ default long abortMultipartUploadsUnderPrefix(String prefix)
+ throws IOException {
+ return 0;
+ }
}
diff --git a/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/impl/RenameOperation.java b/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/impl/RenameOperation.java
index 4bb15f74965a9b..288b3c0aae585e 100644
--- a/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/impl/RenameOperation.java
+++ b/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/impl/RenameOperation.java
@@ -22,6 +22,7 @@
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
+import java.util.Optional;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.atomic.AtomicLong;
@@ -44,6 +45,7 @@
import org.apache.hadoop.util.OperationDuration;
import static org.apache.hadoop.fs.s3a.S3AUtils.translateException;
+import static org.apache.hadoop.fs.s3a.impl.CallableSupplier.waitForCompletionIgnoringExceptions;
import static org.apache.hadoop.fs.store.audit.AuditingFunctions.callableWithinAuditSpan;
import static org.apache.hadoop.fs.s3a.impl.CallableSupplier.submit;
import static org.apache.hadoop.fs.s3a.impl.CallableSupplier.waitForCompletion;
@@ -124,9 +126,18 @@ public class RenameOperation extends ExecutingStoreOperation {
private final List keysToDelete =
new ArrayList<>();
+ /**
+ * Do directory operations purge pending uploads?
+ */
+ private final boolean dirOperationsPurgeUploads;
+
+ /**
+ * Count of uploads aborted.
+ */
+ private Optional uploadsAborted = Optional.empty();
+
/**
* Initiate the rename.
- *
* @param storeContext store context
* @param sourcePath source path
* @param sourceKey key of source
@@ -136,6 +147,7 @@ public class RenameOperation extends ExecutingStoreOperation {
* @param destStatus destination status.
* @param callbacks callback provider
* @param pageSize size of delete requests
+ * @param dirOperationsPurgeUploads Do directory operations purge pending uploads?
*/
public RenameOperation(
final StoreContext storeContext,
@@ -146,7 +158,8 @@ public RenameOperation(
final String destKey,
final S3AFileStatus destStatus,
final OperationCallbacks callbacks,
- final int pageSize) {
+ final int pageSize,
+ final boolean dirOperationsPurgeUploads) {
super(storeContext);
this.sourcePath = sourcePath;
this.sourceKey = sourceKey;
@@ -159,6 +172,16 @@ public RenameOperation(
&& pageSize <= InternalConstants.MAX_ENTRIES_TO_DELETE,
"page size out of range: %s", pageSize);
this.pageSize = pageSize;
+ this.dirOperationsPurgeUploads = dirOperationsPurgeUploads;
+ }
+
+ /**
+ * Get the count of uploads aborted.
+ * Non-empty iff enabled, and the operations completed without errors.
+ * @return count of aborted uploads.
+ */
+ public Optional getUploadsAborted() {
+ return uploadsAborted;
}
/**
@@ -341,6 +364,16 @@ protected void recursiveDirectoryRename() throws IOException {
throw new RenameFailedException(srcKey, dstKey,
"cannot rename a directory to a subdirectory of itself ");
}
+ // start the async dir cleanup
+ final CompletableFuture abortUploads;
+ if (dirOperationsPurgeUploads) {
+ final String key = srcKey;
+ LOG.debug("All uploads under {} will be deleted", key);
+ abortUploads = submit(getStoreContext().getExecutor(), () ->
+ callbacks.abortMultipartUploadsUnderPrefix(key));
+ } else {
+ abortUploads = null;
+ }
if (destStatus != null
&& destStatus.isEmptyDirectory() == Tristate.TRUE) {
@@ -422,6 +455,8 @@ protected void recursiveDirectoryRename() throws IOException {
// have been deleted.
completeActiveCopiesAndDeleteSources("final copy and delete");
+ // and if uploads were being aborted, wait for that to finish
+ uploadsAborted = waitForCompletionIgnoringExceptions(abortUploads);
}
/**
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 22fc630dad1f5a..ea1ea908486e25 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
@@ -47,8 +47,8 @@
import org.apache.hadoop.fs.FilterFileSystem;
import org.apache.hadoop.fs.Options;
import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.fs.RemoteIterator;
import org.apache.hadoop.fs.s3a.Constants;
-import org.apache.hadoop.fs.s3a.MultipartUtils;
import org.apache.hadoop.fs.s3a.S3AFileSystem;
import org.apache.hadoop.fs.s3a.WriteOperationHelper;
import org.apache.hadoop.fs.s3a.auth.RolePolicies;
@@ -683,7 +683,7 @@ private void promptBeforeAbort(PrintStream out) throws IOException {
private void processUploads(PrintStream out) throws IOException {
final S3AFileSystem fs = getFilesystem();
- MultipartUtils.UploadIterator uploads = fs.listUploads(prefix);
+ RemoteIterator uploads = fs.listUploads(prefix);
// create a span so that the write operation helper
// is within one
AuditSpan span =
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 a7ea7b2e590248..0216e46014c7ea 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
@@ -39,11 +39,12 @@ The features which may be unavailable include:
* Optional Bucket Probes at startup (`fs.s3a.bucket.probe = 0`).
This is now the default -do not change it.
* 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
-### Connecting to a third party object store over HTTPS
+## 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`.
@@ -89,6 +90,57 @@ then these must be set, either in XML or (preferred) in a JCEKS file.
If per-bucket settings are used here, then third-party stores and credentials may be used alongside an AWS store.
+
+
+## Other issues
+
+### Coping without bucket lifecycle rules
+
+Not all third-party stores support bucket lifecycle rules to clean up buckets
+of incomplete uploads.
+
+This can be addressed in two ways
+* Command line: `hadoop s3guard uploads -abort -force \`.
+* With `fs.s3a.multipart.purge` and a purge age set in `fs.s3a.multipart.purge.age`
+* In rename/delete `fs.s3a.directory.operations.purge.uploads = true`.
+
+#### S3Guard uploads command
+
+This can be executed on a schedule, or manually
+
+```
+hadoop s3guard uploads -abort -force s3a://bucket/
+```
+
+Consult the [S3Guard documentation](s3guard.html) for the full set of parameters.
+
+#### In startup: `fs.s3a.multipart.purge`
+
+This lists all uploads in a bucket when a filesystem is created and deletes
+all of those above a certain age.
+
+This can hurt performance on a large bucket, as the purge scans the entire tree,
+and is executed whenever a filesystem is created -which can happen many times during
+hive, spark, distcp jobs.
+
+For this reason, this option may be deleted in future, however it has long been
+available in the S3A client and so guaranteed to work across versions.
+
+#### During rename and delete: `fs.s3a.directory.operations.purge.uploads`
+
+When `fs.s3a.directory.operations.purge.uploads` is set, when a directory is renamed
+or deleted, then in parallel with the delete an attempt is made to list
+all pending uploads.
+If there are any, they are aborted (sequentially).
+
+* This is disabled by default: it adds overhead and extra cost.
+* Because it only applies to the directories being processed, directories which
+ are not renamed or deleted will retain all incomplete uploads.
+* There is no age checking: all uploads will be aborted.
+* If any other process is writing to the same directory tree, their operations
+will be cancelled.
+
+
# Troubleshooting
The most common problem when talking to third-party stores are
@@ -412,4 +464,5 @@ It is also a way to regression test foundational S3A third-party store compatibi
```
_Note_ If anyone is set up to test this reguarly, please let the hadoop developer team know if regressions do surface,
-as it is not a common test configuration.
\ No newline at end of file
+as it is not a common test configuration.
+[]
\ No newline at end of file
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 d2a858f615ef6a..7a2a10879dd8ee 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 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 6669e8426af0a1..013ec901d0a77e 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.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 void testCopyToLocalWithUseRawLocalFileSystemOption()
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 dae6312d480982..0e4a8eda5b297e 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 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 @@
/**
* 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 ITestS3AFileOperationCost(
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 void testCostOfGlobStatus() throws Throwable {
// 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 void testCostOfGlobStatusNoSymlinkResolution() throws Throwable {
// 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 7ce7b8385cec4b..56827043c9b825 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.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 void testRenameNonExistentPath() throws Exception {
() -> 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/ITestS3AMultipartUtils.java b/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/ITestS3AMultipartUtils.java
index 263a857e033003..e0559b7c49edc6 100644
--- a/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/ITestS3AMultipartUtils.java
+++ b/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/ITestS3AMultipartUtils.java
@@ -24,6 +24,7 @@
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.fs.RemoteIterator;
import org.apache.hadoop.fs.store.audit.AuditSpan;
import java.io.IOException;
@@ -76,7 +77,7 @@ public void testListMultipartUploads() throws Exception {
// 2. Verify all uploads are found listing by prefix
describe("Verifying upload list by prefix");
- MultipartUtils.UploadIterator uploads = fs.listUploads(getPartPrefix(fs));
+ RemoteIterator uploads = fs.listUploads(getPartPrefix(fs));
assertUploadsPresent(uploads, keySet);
// 3. Verify all uploads are found listing without prefix
@@ -97,7 +98,7 @@ public void testListMultipartUploads() throws Exception {
* @param ourUploads set up uploads that should be present
* @throws IOException on I/O error
*/
- private void assertUploadsPresent(MultipartUtils.UploadIterator list,
+ private void assertUploadsPresent(RemoteIterator list,
Set ourUploads) throws IOException {
// Don't modify passed-in set, use copy.
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 3e343a9ea85f23..3f6870be46b2ad 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
@@ -23,6 +23,7 @@
import software.amazon.awssdk.services.s3.model.UploadPartRequest;
import software.amazon.awssdk.services.s3.model.UploadPartResponse;
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;
@@ -96,7 +97,7 @@ public static void clearAnyUploads(S3AFileSystem fs, Path path) {
String key = fs.pathToKey(path);
AuditSpan span = null;
try {
- MultipartUtils.UploadIterator uploads = fs.listUploads(key);
+ RemoteIterator uploads = fs.listUploads(key);
span = fs.createSpan("multipart", path.toString(), null);
final WriteOperationHelper helper
= fs.getWriteOperationHelper();
@@ -118,7 +119,7 @@ public static void clearAnyUploads(S3AFileSystem fs, Path path) {
public static void assertNoUploadsAt(S3AFileSystem fs, Path path) throws
Exception {
String key = fs.pathToKey(path);
- MultipartUtils.UploadIterator uploads = fs.listUploads(key);
+ RemoteIterator uploads = fs.listUploads(key);
while (uploads.hasNext()) {
MultipartUpload upload = uploads.next();
Assert.fail("Found unexpected upload " + upload.key() + " " +
@@ -130,7 +131,7 @@ public static void assertNoUploadsAt(S3AFileSystem fs, Path path) throws
public static int countUploadsAt(S3AFileSystem fs, Path path) throws
IOException {
String key = fs.pathToKey(path);
- MultipartUtils.UploadIterator uploads = fs.listUploads(key);
+ RemoteIterator uploads = fs.listUploads(key);
int count = 0;
while (uploads.hasNext()) {
MultipartUpload upload = uploads.next();
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 3382c300b93151..aa38186c650325 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 static boolean isBulkDeleteEnabled(FileSystem fs) {
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/auth/ITestAssumeRole.java b/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/auth/ITestAssumeRole.java
index 5534bb77c0ddb6..12234301b50d8f 100644
--- a/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/auth/ITestAssumeRole.java
+++ b/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/auth/ITestAssumeRole.java
@@ -27,6 +27,7 @@
import java.util.stream.IntStream;
import software.amazon.awssdk.auth.credentials.AwsCredentials;
+import software.amazon.awssdk.services.s3.model.MultipartUpload;
import software.amazon.awssdk.services.sts.model.StsException;
import com.fasterxml.jackson.core.JsonProcessingException;
import org.assertj.core.api.Assertions;
@@ -40,10 +41,10 @@
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.LocatedFileStatus;
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.AWSBadRequestException;
import org.apache.hadoop.fs.s3a.AbstractS3ATestBase;
-import org.apache.hadoop.fs.s3a.MultipartUtils;
import org.apache.hadoop.fs.s3a.S3AFileSystem;
import org.apache.hadoop.fs.s3a.S3ATestConstants;
import org.apache.hadoop.fs.s3a.SimpleAWSCredentialsProvider;
@@ -463,7 +464,7 @@ public void testReadOnlyOperations() throws Throwable {
// list multipart uploads.
// This is part of the read policy.
int counter = 0;
- MultipartUtils.UploadIterator iterator = roleFS.listUploads("/");
+ RemoteIterator iterator = roleFS.listUploads("/");
while (iterator.hasNext()) {
counter++;
iterator.next();
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
new file mode 100644
index 00000000000000..9e070273759893
--- /dev/null
+++ b/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/impl/ITestUploadPurgeOnDirectoryOperations.java
@@ -0,0 +1,163 @@
+/*
+ * 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.nio.charset.StandardCharsets;
+
+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.Path;
+import org.apache.hadoop.fs.RemoteIterator;
+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.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.
+ */
+public class ITestUploadPurgeOnDirectoryOperations extends AbstractS3ACostTest {
+
+ @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();
+ assertHasPathCapabilities(fs, new Path("/"),
+ DIRECTORY_OPERATIONS_PURGE_UPLOADS);
+ clearAnyUploads(fs, methodPath());
+ }
+
+ @Test
+ public void testDeleteWithPendingUpload() throws Throwable {
+
+ final S3AFileSystem fs = getFileSystem();
+ final Path dir = methodPath();
+
+ // create a magic file.
+ createMagicFile(fs, dir);
+
+ // and there's a pending upload
+ assertUploadCount(dir, 1);
+
+ // delete the dir, with a cost of 1 abort, 1 list.
+ verifyMetrics(() -> fs.delete(dir, true),
+ with(OBJECT_MULTIPART_UPLOAD_ABORTED, 1), // abort
+ with(OBJECT_MULTIPART_UPLOAD_LIST, 1), // HTTP request inside iterator
+ with(MULTIPART_UPLOAD_LIST, 0)); // api list call
+
+
+ // and the pending upload is gone
+ assertUploadCount(dir, 0);
+ }
+
+ @Test
+ public void testRenameWithPendingUpload() throws Throwable {
+
+ final S3AFileSystem fs = getFileSystem();
+ final Path base = methodPath();
+ final Path dir = new Path(base, "src");
+ final Path dest = new Path(base, "dest");
+
+ // create a magic file.
+ createMagicFile(fs, dir);
+
+ // and there's a pending upload
+ assertUploadCount(dir, 1);
+
+ // rename the dir, with a cost of 1 abort, 1 list.
+ verifyMetrics(() -> fs.rename(dir, dest),
+ with(OBJECT_MULTIPART_UPLOAD_ABORTED, 1), // abort
+ with(OBJECT_MULTIPART_UPLOAD_LIST, 1), // HTTP request inside iterator
+ with(MULTIPART_UPLOAD_LIST, 0)); // api list call
+
+ // and there isn't
+ 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.
+ * @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 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/ITestXAttrCost.java b/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/impl/ITestXAttrCost.java
index f6fe68b8e6d337..71fdb7aeaa62cf 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.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_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 @@ private void logXAttrs(final Map xAttrs) {
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 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/AbstractS3ACostTest.java b/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/performance/AbstractS3ACostTest.java
index 48378ce75dc9cd..e37717bfa1e340 100644
--- a/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/performance/AbstractS3ACostTest.java
+++ b/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/performance/AbstractS3ACostTest.java
@@ -91,6 +91,13 @@ protected AbstractS3ACostTest(
this.keepMarkers = keepMarkers;
}
+ /**
+ * Constructor with markers kept.
+ */
+ public AbstractS3ACostTest() {
+ this(true);
+ }
+
@Override
public Configuration createConfiguration() {
Configuration conf = super.createConfiguration();
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 39530d97bf7941..2d128cffc5af00 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.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.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 @@
* 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 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 void testCreateNoOverwrite() throws Throwable {
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 void testCreateOverwrite() throws Throwable {
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 void testCreateNoOverwriteFileExists() throws Throwable {
// 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 void testCreateBuilderSequence() throws Throwable {
// 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 void testCreateFileRecursive() throws Throwable {
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 void testCreateFileNonRecursive() throws Throwable {
verifyMetrics(() ->
build(fs.createFile(methodPath()).overwrite(true)),
- always(CREATE_FILE_OVERWRITE));
+ always(expected(CREATE_FILE_OVERWRITE)));
}
@@ -196,7 +274,7 @@ public void testCreateNonRecursive() throws Throwable {
.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 cd4ee44406676f..8dbf5baaa6206d 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.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 @@
*
* Similarly: JUnit assertions over AssertJ.
*
- * 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 @@ protected Configuration createConfiguration() {
// 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 fae2df973af7f3..97f51fe2c8dcd2 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.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.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 ITestS3ADeleteCost(final String name,
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/s3guard/ITestS3GuardTool.java b/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/s3guard/ITestS3GuardTool.java
index 844230e8bea9f1..28bc2a246af1a7 100644
--- a/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/s3guard/ITestS3GuardTool.java
+++ b/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/s3guard/ITestS3GuardTool.java
@@ -97,22 +97,22 @@ public void testStoreInfo() throws Throwable {
LOG.info("Exec output=\n{}", output);
}
- private final static String UPLOAD_PREFIX = "test-upload-prefix";
private final static String UPLOAD_NAME = "test-upload";
@Test
public void testUploads() throws Throwable {
S3AFileSystem fs = getFileSystem();
- Path path = path(UPLOAD_PREFIX + "/" + UPLOAD_NAME);
+ Path path = methodPath();
+ Path file = new Path(path, UPLOAD_NAME);
describe("Cleaning up any leftover uploads from previous runs.");
- final String key = fs.pathToKey(path);
+ final String key = fs.pathToKey(file);
try {
// 1. Make sure key doesn't already exist
clearAnyUploads(fs, path);
// 2. Confirm no uploads are listed via API
- assertNoUploadsAt(fs, path.getParent());
+ assertNoUploadsAt(fs, path);
// 3. Confirm no uploads are listed via CLI
describe("Confirming CLI lists nothing.");
@@ -127,8 +127,6 @@ public void testUploads() throws Throwable {
// 6. Confirm part exists via CLI, direct path and parent path
describe("Confirming CLI lists one part");
assertNumUploads(path, 1);
- assertNumUploads(path.getParent(), 1);
- // 7. Use CLI to delete part, assert it worked
describe("Deleting part via CLI");
assertNumDeleted(fs, path, 1);
@@ -150,22 +148,23 @@ public void testUploads() throws Throwable {
@Test
public void testUploadListByAge() throws Throwable {
S3AFileSystem fs = getFileSystem();
- Path path = path(UPLOAD_PREFIX + "/" + UPLOAD_NAME);
+ Path path = methodPath();
+ Path file = new Path(path, UPLOAD_NAME);
describe("Cleaning up any leftover uploads from previous runs.");
+
// 1. Make sure key doesn't already exist
clearAnyUploads(fs, path);
// 2. Create a upload part
describe("Uploading single part.");
- final String key = fs.pathToKey(path);
+ final String key = fs.pathToKey(file);
createPartUpload(fs, key, 128, 1);
//try (AuditSpan span = fs.startOperation("multipart", key, null)) {
try {
- // 3. Confirm it exists via API.. may want to wrap with
- // LambdaTestUtils.eventually() ?
+ // 3. Confirm it exists via API
assertEquals("Should be one upload", 1, countUploadsAt(fs, path));
// 4. Confirm part does appear in listing with long age filter
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 0c2473a5f61daa..759a3bf129eef1 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 @@ protected Configuration createConfiguration() {
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;
}
diff --git a/hadoop-tools/hadoop-sls/src/main/java/org/apache/hadoop/yarn/sls/web/SLSWebApp.java b/hadoop-tools/hadoop-sls/src/main/java/org/apache/hadoop/yarn/sls/web/SLSWebApp.java
index a2974615c4c671..ce36854bca2fde 100644
--- a/hadoop-tools/hadoop-sls/src/main/java/org/apache/hadoop/yarn/sls/web/SLSWebApp.java
+++ b/hadoop-tools/hadoop-sls/src/main/java/org/apache/hadoop/yarn/sls/web/SLSWebApp.java
@@ -136,6 +136,7 @@ public void start() throws Exception {
String webRootDir = getClass().getClassLoader().getResource("html").
toExternalForm();
staticHandler.setResourceBase(webRootDir);
+ staticHandler.start();
Handler handler = new AbstractHandler() {
@Override
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/conf/YarnConfiguration.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/conf/YarnConfiguration.java
index 90a8978a228b2d..d195c8cdae6818 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/conf/YarnConfiguration.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/conf/YarnConfiguration.java
@@ -4369,6 +4369,22 @@ public static boolean isAclEnabled(Configuration conf) {
public static final long DEFAULT_ROUTER_USER_CLIENT_THREAD_POOL_KEEP_ALIVE_TIME =
TimeUnit.SECONDS.toMillis(0); // 0s
+ /**
+ * This method configures the policy for core threads regarding termination
+ * when no tasks arrive within the keep-alive time.
+ * When set to false, core threads are never terminated due to a lack of tasks.
+ * When set to true, the same keep-alive policy
+ * that applies to non-core threads also applies to core threads.
+ * To prevent constant thread replacement,
+ * ensure that the keep-alive time is greater than zero when setting it to true.
+ * It's advisable to call this method before the pool becomes actively used.
+ */
+ public static final String ROUTER_USER_CLIENT_THREAD_POOL_ALLOW_CORE_THREAD_TIMEOUT =
+ ROUTER_PREFIX + "interceptor.user-thread-pool.allow-core-thread-time-out";
+
+ public static final boolean DEFAULT_ROUTER_USER_CLIENT_THREAD_POOL_ALLOW_CORE_THREAD_TIMEOUT =
+ false;
+
/** The address of the Router web application. */
public static final String ROUTER_WEBAPP_ADDRESS =
ROUTER_WEBAPP_PREFIX + "address";
@@ -4549,6 +4565,11 @@ public static boolean isAclEnabled(Configuration conf) {
public static final String DEFAULT_GPG_WEBAPP_HTTPS_ADDRESS =
"0.0.0.0:" + DEFAULT_GPG_WEBAPP_HTTPS_PORT;
+ public static final String GPG_WEBAPP_CONNECT_TIMEOUT = GPG_WEBAPP_PREFIX + "connect-timeout";
+ public static final long DEFAULT_GPG_WEBAPP_CONNECT_TIMEOUT = TimeUnit.SECONDS.toMillis(30);
+ public static final String GPG_WEBAPP_READ_TIMEOUT = GPG_WEBAPP_PREFIX + "read-timeout";
+ public static final long DEFAULT_GPG_WEBAPP_READ_TIMEOUT = TimeUnit.SECONDS.toMillis(30);
+
/**
* Connection and Read timeout from the Router to RM.
*/
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/main/java/org/apache/hadoop/yarn/client/cli/RouterCLI.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/main/java/org/apache/hadoop/yarn/client/cli/RouterCLI.java
index 0aa02c8124a3a7..2da584f9d61bb0 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/main/java/org/apache/hadoop/yarn/client/cli/RouterCLI.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/main/java/org/apache/hadoop/yarn/client/cli/RouterCLI.java
@@ -23,7 +23,10 @@
import org.apache.commons.cli.Option;
import org.apache.commons.cli.Options;
import org.apache.commons.cli.ParseException;
+import org.apache.commons.collections.CollectionUtils;
+import org.apache.commons.collections.MapUtils;
import org.apache.commons.lang3.StringUtils;
+import org.apache.hadoop.classification.VisibleForTesting;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.ha.HAAdmin.UsageInfo;
@@ -67,6 +70,7 @@
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
+import java.util.LinkedHashMap;
import java.util.stream.Collectors;
import static org.apache.hadoop.yarn.server.api.protocolrecords.FederationQueueWeight.checkHeadRoomAlphaValid;
@@ -74,29 +78,8 @@
public class RouterCLI extends Configured implements Tool {
-
private static final Logger LOG = LoggerFactory.getLogger(RouterCLI.class);
- protected final static Map ADMIN_USAGE =
- ImmutableMap.builder()
- // Command1: deregisterSubCluster
- .put("-deregisterSubCluster", new UsageInfo(
- "[-sc|--subClusterId [subCluster Id]]",
- "Deregister SubCluster, If the interval between the heartbeat time of the subCluster " +
- "and the current time exceeds the timeout period, " +
- "set the state of the subCluster to SC_LOST."))
- // Command2: policy
- .put("-policy", new UsageInfo(
- "[-s|--save [queue;router weight;amrm weight;headroomalpha]] " +
- "[-bs|--batch-save [--format xml] [-f|--input-file fileName]]" +
- "[-l|--list [--pageSize][--currentPage][--queue][--queues]]",
- "We provide a set of commands for Policy:" +
- " Include list policies, save policies, batch save policies. " +
- " (Note: The policy type will be directly read from the" +
- " yarn.federation.policy-manager in the local yarn-site.xml.)" +
- " eg. (routeradmin -policy [-s|--save] root.a;SC-1:0.7,SC-2:0.3;SC-1:0.7,SC-2:0.3;1.0)"))
- .build();
-
// Common Constant
private static final String SEMICOLON = ";";
@@ -104,6 +87,7 @@ public class RouterCLI extends Configured implements Tool {
private static final String CMD_EMPTY = "";
private static final int EXIT_SUCCESS = 0;
private static final int EXIT_ERROR = -1;
+ private static final String CMD_HELP = "-help";
// Command1: deregisterSubCluster
private static final String DEREGISTER_SUBCLUSTER_TITLE =
@@ -115,17 +99,46 @@ public class RouterCLI extends Configured implements Tool {
private static final String OPTION_SC = "sc";
private static final String OPTION_SUBCLUSTERID = "subClusterId";
private static final String CMD_DEREGISTERSUBCLUSTER = "-deregisterSubCluster";
- private static final String CMD_HELP = "-help";
+
+ // DeregisterSubCluster Command Parameters
+ protected final static UsageInfo SUBCLUSTER_ID = new UsageInfo("<-sc|--subClusterId>",
+ "'-sc' option allows you to specify the sub-cluster to operate on, " +
+ "while the '--subClusterId' option is the long format of -sc and serves the same purpose.");
+
+ // DeregisterSubCluster Command Examples
+ protected final static String DEREGISTER_SUBCLUSTER_EXAMPLE_1 =
+ "yarn routeradmin -deregisterSubCluster -sc SC-1";
+ protected final static String DEREGISTER_SUBCLUSTER_EXAMPLE_2 =
+ "yarn routeradmin -deregisterSubCluster --subClusterId SC-1";
+
+ // DeregisterSubCluster Command Help Information
+ protected final static String DEREGISTER_SUBCLUSTER_HELP_INFO =
+ "deregister subCluster, If the interval between the heartbeat time of the subCluster and" +
+ "the current time exceeds the timeout period, set the state of the subCluster to SC_LOST.";
+
+ protected final static RouterCmdUsageInfos DEREGISTER_SUBCLUSTER_USAGEINFOS =
+ new RouterCmdUsageInfos()
+ .addUsageInfo(SUBCLUSTER_ID)
+ .addHelpInfo(DEREGISTER_SUBCLUSTER_HELP_INFO)
+ .addExampleDescs(CMD_DEREGISTERSUBCLUSTER, "If we want to deregisterSubCluster SC-1")
+ .addExample(CMD_DEREGISTERSUBCLUSTER, DEREGISTER_SUBCLUSTER_EXAMPLE_1)
+ .addExample(CMD_DEREGISTERSUBCLUSTER, DEREGISTER_SUBCLUSTER_EXAMPLE_2);
// Command2: policy
+
+ private static final String CMD_POLICY = "-policy";
+
// save policy
private static final String OPTION_S = "s";
- private static final String OPTION_BATCH_S = "bs";
private static final String OPTION_SAVE = "save";
+ // batch save policy
+ private static final String OPTION_BATCH_S = "bs";
private static final String OPTION_BATCH_SAVE = "batch-save";
private static final String OPTION_FORMAT = "format";
+ private static final String FORMAT_XML = "xml";
private static final String OPTION_FILE = "f";
private static final String OPTION_INPUT_FILE = "input-file";
+ // list policy
private static final String OPTION_L = "l";
private static final String OPTION_LIST = "list";
private static final String OPTION_PAGE_SIZE = "pageSize";
@@ -133,9 +146,6 @@ public class RouterCLI extends Configured implements Tool {
private static final String OPTION_QUEUE = "queue";
private static final String OPTION_QUEUES = "queues";
- private static final String CMD_POLICY = "-policy";
- private static final String FORMAT_XML = "xml";
- private static final String FORMAT_JSON = "json";
private static final String XML_TAG_SUBCLUSTERIDINFO = "subClusterIdInfo";
private static final String XML_TAG_AMRMPOLICYWEIGHTS = "amrmPolicyWeights";
private static final String XML_TAG_ROUTERPOLICYWEIGHTS = "routerPolicyWeights";
@@ -146,10 +156,85 @@ public class RouterCLI extends Configured implements Tool {
private static final String LIST_POLICIES_TITLE =
"Yarn Federation Queue Policies";
+
// Columns information
private static final List LIST_POLICIES_HEADER = Arrays.asList(
"Queue Name", "AMRM Weight", "Router Weight");
+ // Policy Commands
+ protected final static UsageInfo POLICY_SAVE_USAGE = new UsageInfo(
+ "-s|--save ()",
+ "This command is used to save the policy information of the queue, " +
+ "including queue and weight information.");
+
+ protected final static String POLICY_SAVE_USAGE_EXAMPLE_DESC =
+ "We have two sub-clusters, SC-1 and SC-2. \\" +
+ "We want to configure a weight policy for the 'root.a' queue. \\" +
+ "The Router Weight is set to SC-1 with a weight of 0.7 and SC-2 with a weight of 0.3. \\" +
+ "The AMRM Weight is set SC-1 to 0.6 and SC-2 to 0.4. \\" +
+ "We are using the default value of 0.1 for headroomalpha.";
+
+ protected final static String POLICY_SAVE_USAGE_EXAMPLE_1 =
+ "yarn routeradmin -policy -s root.a;SC-1:0.7,SC-2:0.3;SC-1:0.6,SC-2:0.4;1.0";
+ protected final static String POLICY_SAVE_USAGE_EXAMPLE_2 =
+ "yarn routeradmin -policy --save root.a;SC-1:0.7,SC-2:0.3;SC-1:0.6,SC-2:0.4;1.0";
+
+ protected final static UsageInfo POLICY_BATCH_SAVE_USAGE = new UsageInfo(
+ "-bs|--batch-save (--format ) (-f|--input-file )",
+ "This command can batch load weight information for queues " +
+ "based on the provided `federation-weights.xml` file.");
+
+ protected final static String POLICY_BATCH_SAVE_USAGE_EXAMPLE_DESC =
+ "We have two sub-clusters, SC-1 and SC-2. \\" +
+ "We would like to configure weights for 'root.a' and 'root.b' queues. \\" +
+ "We can set the weights for 'root.a' and 'root.b' in the 'federation-weights.xml' file. \\" +
+ "and then use the batch-save command to save the configurations in bulk.";
+
+ protected final static String POLICY_BATCH_SAVE_USAGE_EXAMPLE_1 =
+ "yarn routeradmin -policy -bs --format xml -f federation-weights.xml";
+ protected final static String POLICY_BATCH_SAVE_USAGE_EXAMPLE_2 =
+ "yarn routeradmin -policy --batch-save --format xml -f federation-weights.xml";
+
+ protected final static UsageInfo POLICY_LIST_USAGE = new UsageInfo(
+ "-l|--list [--pageSize][--currentPage][--queue][--queues]",
+ "This command is used to display the configured queue weight information.");
+
+ protected final static String POLICY_LIST_USAGE_EXAMPLE_DESC =
+ "We can display the list of already configured queue weight information. \\" +
+ "We can use the --queue option to query the weight information for a specific queue \\" +
+ " or use the --queues option to query the weight information for multiple queues. \\";
+
+ protected final static String POLICY_LIST_USAGE_EXAMPLE_1 =
+ "yarn routeradmin -policy -l --pageSize 20 --currentPage 1 --queue root.a";
+
+ protected final static String POLICY_LIST_USAGE_EXAMPLE_2 =
+ "yarn routeradmin -policy -list --pageSize 20 --currentPage 1 --queues root.a,root.b";
+
+ protected final static RouterCmdUsageInfos POLICY_USAGEINFOS = new RouterCmdUsageInfos()
+ // Policy Save
+ .addUsageInfo(POLICY_SAVE_USAGE)
+ .addExampleDescs(POLICY_SAVE_USAGE.args, POLICY_SAVE_USAGE_EXAMPLE_DESC)
+ .addExample(POLICY_SAVE_USAGE.args, POLICY_SAVE_USAGE_EXAMPLE_1)
+ .addExample(POLICY_SAVE_USAGE.args, POLICY_SAVE_USAGE_EXAMPLE_2)
+ // Policy Batch Save
+ .addUsageInfo(POLICY_BATCH_SAVE_USAGE)
+ .addExampleDescs(POLICY_BATCH_SAVE_USAGE.args, POLICY_BATCH_SAVE_USAGE_EXAMPLE_DESC)
+ .addExample(POLICY_BATCH_SAVE_USAGE.args, POLICY_BATCH_SAVE_USAGE_EXAMPLE_1)
+ .addExample(POLICY_BATCH_SAVE_USAGE.args, POLICY_BATCH_SAVE_USAGE_EXAMPLE_2)
+ // Policy List Save
+ .addUsageInfo(POLICY_LIST_USAGE)
+ .addExampleDescs(POLICY_LIST_USAGE.args, POLICY_LIST_USAGE_EXAMPLE_DESC)
+ .addExample(POLICY_LIST_USAGE.args, POLICY_LIST_USAGE_EXAMPLE_1)
+ .addExample(POLICY_LIST_USAGE.args, POLICY_LIST_USAGE_EXAMPLE_2);
+
+ protected final static Map ADMIN_USAGE =
+ ImmutableMap.builder()
+ // Command1: deregisterSubCluster
+ .put(CMD_DEREGISTERSUBCLUSTER, DEREGISTER_SUBCLUSTER_USAGEINFOS)
+ // Command2: policy
+ .put(CMD_POLICY, POLICY_USAGEINFOS)
+ .build();
+
public RouterCLI() {
super();
}
@@ -159,43 +244,66 @@ public RouterCLI(Configuration conf) {
}
private static void buildHelpMsg(String cmd, StringBuilder builder) {
- UsageInfo usageInfo = ADMIN_USAGE.get(cmd);
- if (usageInfo == null) {
+ RouterCmdUsageInfos routerUsageInfo = ADMIN_USAGE.get(cmd);
+
+ if (routerUsageInfo == null) {
return;
}
+ builder.append("[").append(cmd).append("]\n");
- if (usageInfo.args != null) {
- String space = (usageInfo.args == "") ? "" : " ";
- builder.append(" ")
- .append(cmd)
- .append(space)
- .append(usageInfo.args)
- .append(": ")
- .append(usageInfo.help);
- } else {
- builder.append(" ")
- .append(cmd)
- .append(": ")
- .append(usageInfo.help);
+ if (!routerUsageInfo.helpInfos.isEmpty()) {
+ builder.append("\t Description: \n");
+ for (String helpInfo : routerUsageInfo.helpInfos) {
+ builder.append("\t\t").append(helpInfo).append("\n\n");
+ }
}
- }
- private static void buildIndividualUsageMsg(String cmd, StringBuilder builder) {
- UsageInfo usageInfo = ADMIN_USAGE.get(cmd);
- if (usageInfo == null) {
- return;
+ if (!routerUsageInfo.usageInfos.isEmpty()) {
+ builder.append("\t UsageInfos: \n");
+ for (UsageInfo usageInfo : routerUsageInfo.usageInfos) {
+ builder.append("\t\t").append(usageInfo.args)
+ .append(": ")
+ .append("\n\t\t")
+ .append(usageInfo.help).append("\n\n");
+ }
}
- if (usageInfo.args == null) {
- builder.append("Usage: routeradmin [")
- .append(cmd)
- .append("]\n");
- } else {
- String space = (usageInfo.args == "") ? "" : " ";
- builder.append("Usage: routeradmin [")
- .append(cmd)
- .append(space)
- .append(usageInfo.args)
- .append("]\n");
+
+ if (MapUtils.isNotEmpty(routerUsageInfo.examples)) {
+ builder.append("\t Examples: \n");
+ int count = 1;
+ for (Map.Entry> example : routerUsageInfo.examples.entrySet()) {
+
+ String keyCmd = example.getKey();
+ builder.append("\t\t")
+ .append("Cmd:").append(count)
+ .append(". ").append(keyCmd)
+ .append(": \n\n");
+
+ // Print Command Description
+ List exampleDescs = routerUsageInfo.exampleDescs.get(keyCmd);
+ if (CollectionUtils.isNotEmpty(exampleDescs)) {
+ builder.append("\t\t").append("Cmd Requirement Description:\n");
+ for (String value : exampleDescs) {
+ String[] valueDescs = StringUtils.split(value, "\\");
+ for (String valueDesc : valueDescs) {
+ builder.append("\t\t").append(valueDesc).append("\n");
+ }
+ }
+ }
+
+ builder.append("\n");
+
+ // Print Command example
+ List valueExamples = example.getValue();
+ if (CollectionUtils.isNotEmpty(valueExamples)) {
+ builder.append("\t\t").append("Cmd Examples:\n");
+ for (String valueExample : valueExamples) {
+ builder.append("\t\t").append(valueExample).append("\n");
+ }
+ }
+ builder.append("\n");
+ count++;
+ }
}
}
@@ -204,12 +312,7 @@ private static void printHelp() {
summary.append("routeradmin is the command to execute ")
.append("YARN Federation administrative commands.\n")
.append("The full syntax is: \n\n")
- .append("routeradmin\n")
- .append(" [-deregisterSubCluster [-sc|--subClusterId [subCluster Id]]\n")
- .append(" [-policy [-s|--save [queue;router weight;amrm weight;headroomalpha] " +
- "[-bs|--batch-save [--format xml,json] [-f|--input-file fileName]]] " +
- "[-l|--list [--pageSize][--currentPage][--queue][--queues]]\n")
- .append(" [-help [cmd]]").append("\n");
+ .append("routeradmin\n");
StringBuilder helpBuilder = new StringBuilder();
System.out.println(summary);
@@ -235,13 +338,9 @@ protected ResourceManagerAdministrationProtocol createAdminProtocol()
private static void buildUsageMsg(StringBuilder builder) {
builder.append("routeradmin is only used in Yarn Federation Mode.\n");
builder.append("Usage: routeradmin\n");
- for (Map.Entry cmdEntry : ADMIN_USAGE.entrySet()) {
- UsageInfo usageInfo = cmdEntry.getValue();
- builder.append(" ")
- .append(cmdEntry.getKey())
- .append(" ")
- .append(usageInfo.args)
- .append("\n");
+ for (String cmdKey : ADMIN_USAGE.keySet()) {
+ buildHelpMsg(cmdKey, builder);
+ builder.append("\n");
}
builder.append(" -help [cmd]\n");
}
@@ -249,7 +348,7 @@ private static void buildUsageMsg(StringBuilder builder) {
private static void printUsage(String cmd) {
StringBuilder usageBuilder = new StringBuilder();
if (ADMIN_USAGE.containsKey(cmd)) {
- buildIndividualUsageMsg(cmd, usageBuilder);
+ buildHelpMsg(cmd, usageBuilder);
} else {
buildUsageMsg(usageBuilder);
}
@@ -353,7 +452,7 @@ private int handlePolicy(String[] args)
saveOpt.setOptionalArg(true);
Option batchSaveOpt = new Option(OPTION_BATCH_S, OPTION_BATCH_SAVE, false,
"We will save queue policies in bulk, " +
- "where users can provide XML or JSON files containing the policies. " +
+ "where users can provide XML files containing the policies. " +
"This command will parse the file contents and store the results " +
"in the FederationStateStore.");
Option formatOpt = new Option(null, "format", true,
@@ -748,8 +847,59 @@ public int run(String[] args) throws Exception {
return EXIT_SUCCESS;
}
+ public static UsageInfo getPolicyBatchSaveUsage() {
+ return POLICY_BATCH_SAVE_USAGE;
+ }
+
+ static class RouterCmdUsageInfos {
+ private List usageInfos;
+ private List helpInfos;
+ private Map> examples;
+ protected Map> exampleDescs;
+
+ RouterCmdUsageInfos() {
+ this.usageInfos = new ArrayList<>();
+ this.helpInfos = new ArrayList<>();
+ this.examples = new LinkedHashMap<>();
+ this.exampleDescs = new LinkedHashMap<>();
+ }
+
+ public RouterCmdUsageInfos addUsageInfo(UsageInfo usageInfo) {
+ this.usageInfos.add(usageInfo);
+ return this;
+ }
+
+ public RouterCmdUsageInfos addHelpInfo(String helpInfo) {
+ this.helpInfos.add(helpInfo);
+ return this;
+ }
+
+ private RouterCmdUsageInfos addExample(String cmd, String example) {
+ List exampleList = this.examples.getOrDefault(cmd, new ArrayList<>());
+ exampleList.add(example);
+ this.examples.put(cmd, exampleList);
+ return this;
+ }
+
+ private RouterCmdUsageInfos addExampleDescs(String cmd, String exampleDesc) {
+ List exampleDescList = this.exampleDescs.getOrDefault(cmd, new ArrayList<>());
+ exampleDescList.add(exampleDesc);
+ this.exampleDescs.put(cmd, exampleDescList);
+ return this;
+ }
+
+ public Map> getExamples() {
+ return examples;
+ }
+ }
+
public static void main(String[] args) throws Exception {
int result = ToolRunner.run(new RouterCLI(), args);
System.exit(result);
}
+
+ @VisibleForTesting
+ public Map getAdminUsage(){
+ return ADMIN_USAGE;
+ }
}
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/test/java/org/apache/hadoop/yarn/client/cli/TestRouterCLI.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/test/java/org/apache/hadoop/yarn/client/cli/TestRouterCLI.java
index 6ed83826dfa586..a86878dac3f81f 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/test/java/org/apache/hadoop/yarn/client/cli/TestRouterCLI.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/test/java/org/apache/hadoop/yarn/client/cli/TestRouterCLI.java
@@ -40,6 +40,7 @@
import java.util.ArrayList;
import java.util.Date;
import java.util.List;
+import java.util.Map;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
@@ -266,4 +267,29 @@ public void testListPolicies() throws Exception {
String[] args = {"-policy", "-l", "--queue", "root.a"};
assertEquals(0, rmAdminCLI.run(args));
}
+
+ @Test
+ public void testBuildHelpMsg() throws Exception {
+ Map adminUsage = rmAdminCLI.getAdminUsage();
+ assertEquals(2, adminUsage.size());
+
+ RouterCLI.RouterCmdUsageInfos deregisterSubClusterUsageInfos =
+ adminUsage.get("-deregisterSubCluster");
+ assertNotNull(deregisterSubClusterUsageInfos);
+ Map> dsExamplesMap = deregisterSubClusterUsageInfos.getExamples();
+ assertNotNull(dsExamplesMap);
+ assertEquals(1, dsExamplesMap.size());
+ List dsExamples = dsExamplesMap.get("-deregisterSubCluster");
+ assertNotNull(dsExamples);
+ assertEquals(2, dsExamples.size());
+
+ RouterCLI.RouterCmdUsageInfos policyUsageInfos = adminUsage.get("-policy");
+ assertNotNull(policyUsageInfos);
+ Map> policyExamplesMap = policyUsageInfos.getExamples();
+ assertNotNull(policyExamplesMap);
+ assertEquals(3, policyExamplesMap.size());
+ policyExamplesMap.forEach((cmd, cmdExamples) -> {
+ assertEquals(2, cmdExamples.size());
+ });
+ }
}
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/logaggregation/AggregatedLogFormat.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/logaggregation/AggregatedLogFormat.java
index 477a8a293cebf6..26c3e01a45d036 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/logaggregation/AggregatedLogFormat.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/logaggregation/AggregatedLogFormat.java
@@ -32,6 +32,7 @@
import java.io.PrintStream;
import java.io.Writer;
import java.nio.charset.Charset;
+import java.nio.charset.StandardCharsets;
import java.security.PrivilegedExceptionAction;
import java.util.ArrayList;
import java.util.Arrays;
@@ -783,7 +784,10 @@ public static void readAcontainerLogs(DataInputStream valueStream,
OutputStream os = null;
PrintStream ps = null;
try {
- os = new WriterOutputStream(writer, Charset.forName("UTF-8"));
+ os = WriterOutputStream.builder()
+ .setWriter(writer)
+ .setCharset(StandardCharsets.UTF_8)
+ .get();
ps = new PrintStream(os);
while (true) {
try {
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/resources/yarn-default.xml b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/resources/yarn-default.xml
index 9991e841d74b64..0e317712f826c6 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/resources/yarn-default.xml
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/resources/yarn-default.xml
@@ -5132,10 +5132,27 @@
yarn.router.interceptor.user-thread-pool.keep-alive-time
- 0s
+ 30s
This configurable is used to set the keepAliveTime of the thread pool of the interceptor.
- Default is 0s.
+ Default is 30s.
+
+
+
+
+ yarn.router.interceptor.user-thread-pool.allow-core-thread-time-out
+ false
+
+ This method configures the policy for core threads regarding termination
+ when no tasks arrive within the keep-alive time.
+ When set to false, core threads are never terminated due to a lack of tasks.
+ When set to true, the same keep-alive policy
+ that applies to non-core threads also applies to core threads.
+ To prevent constant thread replacement,
+ ensure that the keep-alive time is greater than zero when setting it to true.
+ It's advisable to call this method before the pool becomes actively used.
+ We need to ensure that
+ yarn.router.interceptor.user-thread-pool.keep-alive-time is greater than 0.
@@ -5706,4 +5723,27 @@
50000