HADOOP-19256. review of conditional write

+ hadoop common and s3a create file options

wired up to s3a create file builder, and nowhere else.

Change-Id: Ieaafa9ecdfd68306deb41ecf40c416374a310859

Author: steveloughran

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/Options.java

@@ -710,4 +710,55 @@ private OpenFileOptions() {
     public static final String FS_OPTION_OPENFILE_EC_POLICY =
         FS_OPTION_OPENFILE + "ec.policy";
   }
+
+  /**
+   * The standard {@code createFile()} options.
+   * <p>
+   * If an option is not supported during file creation and it is considered
+   * part of a commit protocol, then, when supplied in a must() option,
+   * it MUST be rejected.
+   */
+  @InterfaceAudience.Public
+  @InterfaceStability.Evolving
+  public interface CreateFileOptionKeys {
+
+    /**
+     * {code createFile()} option to write a file iff there is nothing at the destination.
+     * This may happen during create() or in the close.
+     * <p>
+     * Explicitly set {@link #FS_OPTION_CREATE_IN_CLOSE} if you want to force the end of file
+     * creation.
+     *
+     * Value {@value}.
+     * <p>
+     * This can be set in the builder.
+     * <p>
+     * It should be exported as a path capability for all stores where
+     * the feature is available *and* enabled.
+     */
+    String FS_OPTION_CREATE_CONDITIONAL_OVERWRITE = "fs.option.create.conditional.overwrite";
+
+    /**
+     * Overwrite a file only if there is an Etag match. This option takes a string.
+     * Value {@value}.
+     */
+    String FS_OPTION_CREATE_CONDITIONAL_OVERWRITE_ETAG =
+        "fs.option.create.conditional.overwrite.etag";
+
+    /**
+     * String to define the content filetype.
+     * Value {@value}.
+     */
+    String FS_OPTION_CREATE_CONTENT_TYPE = "fs.option.create.content.type";
+
+    /**
+     * A flag which requires the filesystem to create files/objects in close(),
+     * rather than create/createFile.
+     * <p>
+     * Object stores with this behavior should also export it as a path capability.
+     *
+     * Value {@value}.
+     */
+    String FS_OPTION_CREATE_IN_CLOSE = "fs.option.create.in.close";
+  }
 }

hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/Constants.java

@@ -1502,13 +1502,6 @@ private Constants() {
    */
   public static final String FS_S3A_CREATE_PERFORMANCE = "fs.s3a.create.performance";
 
-  /**
-   * Flag for commit if none match.
-   * This can be set in the {code createFile()} builder.
-   * Value {@value}.
-   */
-  public static final String FS_S3A_CONDITIONAL_FILE_CREATE = "fs.s3a.conditional.file.create";
-
   /**
    * Default value for create performance in an S3A FS.
    * Value {@value}.
@@ -1528,6 +1521,23 @@ private Constants() {
    */
   public static final String FS_S3A_PERFORMANCE_FLAGS =
       "fs.s3a.performance.flags";
+
+
+  /**
+   * Is the create overwrite feature enabled or not?
+   * A configuration option and a path status probe.
+   * Value {@value}.
+   */
+  public static final String FS_S3A_CREATE_OVERWRITE_SUPPORTED = "fs.s3a.create.overwrite.supported";
+
+  /**
+   * Create a multipart file, always: {@value}.
+   * <p>
+   * This is inefficient and will not work on a store which doesn't support that feature,
+   * so is primarily for testing.
+   */
+  public static final String FS_S3A_CREATE_MULTIPART = "fs.s3a.create.multipart";
+
   /**
    * 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

@@ -223,6 +223,10 @@
 import static org.apache.hadoop.fs.CommonConfigurationKeysPublic.IO_FILE_BUFFER_SIZE_DEFAULT;
 import static org.apache.hadoop.fs.CommonConfigurationKeysPublic.IO_FILE_BUFFER_SIZE_KEY;
 import static org.apache.hadoop.fs.CommonPathCapabilities.DIRECTORY_LISTING_INCONSISTENT;
+import static org.apache.hadoop.fs.Options.CreateFileOptionKeys.FS_OPTION_CREATE_CONDITIONAL_OVERWRITE;
+import static org.apache.hadoop.fs.Options.CreateFileOptionKeys.FS_OPTION_CREATE_CONDITIONAL_OVERWRITE_ETAG;
+import static org.apache.hadoop.fs.Options.CreateFileOptionKeys.FS_OPTION_CREATE_CONTENT_TYPE;
+import static org.apache.hadoop.fs.Options.CreateFileOptionKeys.FS_OPTION_CREATE_IN_CLOSE;
 import static org.apache.hadoop.fs.impl.FlagSet.buildFlagSet;
 import static org.apache.hadoop.fs.impl.PathCapabilitiesSupport.validatePathCapabilityArgs;
 import static org.apache.hadoop.fs.s3a.Constants.*;
@@ -2171,9 +2175,9 @@ private FSDataOutputStream innerCreateFile(
     String destKey = putTracker.getDestKey();
 
     // put options are derived from the option builder.
-    boolean conditionalCreate = options.isConditionalCreate();
+    boolean conditionalCreate = options.isConditionalOverwrite();
     final PutObjectOptions putOptions =
-        new PutObjectOptions(null, options.getHeaders(), false, null);
+        new PutObjectOptions(null, options.getHeaders(), conditionalCreate, null);
 
 
     validateOutputStreamConfiguration(path, getConf());
@@ -3248,7 +3252,8 @@ private DeleteObjectsResponse deleteObjects(DeleteObjectsRequest deleteRequest)
   public PutObjectRequest.Builder newPutObjectRequestBuilder(String key,
       long length,
       boolean isDirectoryMarker) {
-    return requestFactory.newPutObjectRequestBuilder(key, null, length, isDirectoryMarker);
+    return requestFactory.newPutObjectRequestBuilder(key, null, length, isDirectoryMarker,
+        PutObjectOptions.defaultOptions());
   }
 
   /**
@@ -5416,11 +5421,19 @@ public boolean hasPathCapability(final Path path, final String capability)
     case STORE_CAPABILITY_DIRECTORY_MARKER_MULTIPART_UPLOAD_ENABLED:
       return isMultipartUploadEnabled();
 
-    // create file options
+    // create file options which are always true
+
+    case FS_OPTION_CREATE_IN_CLOSE:
+    case FS_OPTION_CREATE_CONTENT_TYPE:
     case FS_S3A_CREATE_PERFORMANCE:
     case FS_S3A_CREATE_HEADER:
       return true;
 
+    case FS_OPTION_CREATE_CONDITIONAL_OVERWRITE:
+    case FS_OPTION_CREATE_CONDITIONAL_OVERWRITE_ETAG:
+      // TODO HADOOP-19256. conditional on enablement
+      return true;
+
     // is the FS configured for create file performance
     case FS_S3A_CREATE_PERFORMANCE_ENABLED:
       return performanceFlags.enabled(PerformanceFlagEnum.Create);

hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/WriteOperationHelper.java

@@ -245,7 +245,7 @@ public PutObjectRequest createPutObjectRequest(String destKey,
     activateAuditSpan();
 
     return getRequestFactory()
-        .newPutObjectRequestBuilder(destKey, options, length, false)
+        .newPutObjectRequestBuilder(destKey, options, length, false, options)
         .build();
   }
 

hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/api/RequestFactory.java

@@ -118,12 +118,14 @@ CopyObjectRequest.Builder newCopyObjectRequestBuilder(String srcKey,
    * @param options options for the request
    * @param length length of object to be uploaded
    * @param isDirectoryMarker true if object to be uploaded is a directory marker
+   * @param putOptions
    * @return the request builder
    */
   PutObjectRequest.Builder newPutObjectRequestBuilder(String key,
       PutObjectOptions options,
       long length,
-      boolean isDirectoryMarker);
+      boolean isDirectoryMarker,
+      PutObjectOptions putOptions);
 
   /**
    * Create a {@link PutObjectRequest} request for creating