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

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

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.

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);
     }

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(),

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;
+    }
+  }
 }

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;
+      }
+    }
+  }
 }

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

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;
+      }
+    }
+  }
 }

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);
+  }
+
+
 }

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));

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)

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;
   }
 

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()) {

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;
   }