HADOOP-16085: use object version or etags to protect against inconsistent read after replace/overwrite

hadoop-common-project/hadoop-common/src/main/resources/core-default.xml

@@ -1904,15 +1904,15 @@
   <name>fs.s3a.change.detection.mode</name>
   <value>server</value>
   <description>
-    Determines how change detection is applied to alert to S3 objects
-    rewritten while being read. Value 'server' indicates to apply the attribute
-    constraint directly on GetObject requests to S3. Value 'client' means to do a
-    client-side comparison of the attribute value returned in the response.  Value
-    'server' would not work with third-party S3 implementations that do not
-    support these constraints on GetObject. Values 'server' and 'client' generate
-    RemoteObjectChangedException when a mismatch is detected.  Value 'warn' works
-    like 'client' but generates only a warning.  Value 'none' will ignore change
-    detection completely.
+    Determines how change detection is applied to alert to inconsistent S3
+    objects read during or after an overwrite. Value 'server' indicates to apply
+    the attribute constraint directly on GetObject requests to S3. Value 'client'
+    means to do a client-side comparison of the attribute value returned in the
+    response.  Value 'server' would not work with third-party S3 implementations
+    that do not support these constraints on GetObject. Values 'server' and
+    'client' generate RemoteObjectChangedException when a mismatch is detected.
+    Value 'warn' works like 'client' but generates only a warning.  Value 'none'
+    will ignore change detection completely.
   </description>
 </property>
 

hadoop-tools/hadoop-aws/pom.xml

@@ -406,6 +406,11 @@
       <artifactId>hadoop-common</artifactId>
       <scope>provided</scope>
     </dependency>
+    <dependency>
+      <groupId>org.apache.httpcomponents</groupId>
+      <artifactId>httpcore</artifactId>
+      <scope>provided</scope>
+    </dependency>
     <dependency>
       <groupId>org.apache.hadoop</groupId>
       <artifactId>hadoop-common</artifactId>

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

@@ -197,6 +197,33 @@ public void retry(String action,
         });
   }
 
+  /**
+   * Execute a void operation with retry processing when doRetry=true, else
+   * just once.
+   * @param doRetry true if retries should be performed
+   * @param action action to execute (used in error messages)
+   * @param path path of work (used in error messages)
+   * @param idempotent does the operation have semantics
+   * which mean that it can be retried even if was already executed?
+   * @param retrying callback on retries
+   * @param operation operation to execute
+   * @throws IOException any IOE raised, or translated exception
+   */
+  @Retries.RetryTranslated
+  public void maybeRetry(boolean doRetry,
+      String action,
+      String path,
+      boolean idempotent,
+      Retried retrying,
+      VoidOperation operation)
+      throws IOException {
+    maybeRetry(doRetry, action, path, idempotent, retrying,
+        () -> {
+          operation.execute();
+          return null;
+        });
+  }
+
   /**
    * Execute a void operation with  the default retry callback invoked.
    * @param action action to execute (used in error messages)
@@ -215,6 +242,28 @@ public void retry(String action,
     retry(action, path, idempotent, retryCallback, operation);
   }
 
+  /**
+   * Execute a void operation with the default retry callback invoked when
+   * doRetry=true, else just once.
+   * @param doRetry true if retries should be performed
+   * @param action action to execute (used in error messages)
+   * @param path path of work (used in error messages)
+   * @param idempotent does the operation have semantics
+   * which mean that it can be retried even if was already executed?
+   * @param operation operation to execute
+   * @throws IOException any IOE raised, or translated exception
+   */
+  @Retries.RetryTranslated
+  public void maybeRetry(
+      boolean doRetry,
+      String action,
+      String path,
+      boolean idempotent,
+      VoidOperation operation)
+      throws IOException {
+    maybeRetry(doRetry, action, path, idempotent, retryCallback, operation);
+  }
+
   /**
    * Execute a function with the default retry callback invoked.
    * @param action action to execute (used in error messages)
@@ -265,6 +314,41 @@ public <T> T retry(
         () -> once(action, path, operation));
   }
 
+  /**
+   * Execute a function with retry processing when doRetry=true, else just once.
+   * Uses {@link #once(String, String, Operation)} as the inner
+   * invocation mechanism before retry logic is performed.
+   * @param <T> type of return value
+   * @param doRetry true if retries should be performed
+   * @param action action to execute (used in error messages)
+   * @param path path of work (used in error messages)
+   * @param idempotent does the operation have semantics
+   * which mean that it can be retried even if was already executed?
+   * @param retrying callback on retries
+   * @param operation operation to execute
+   * @return the result of the call
+   * @throws IOException any IOE raised, or translated exception
+   */
+  @Retries.RetryTranslated
+  public <T> T maybeRetry(
+      boolean doRetry,
+      String action,
+      @Nullable String path,
+      boolean idempotent,
+      Retried retrying,
+      Operation<T> operation)
+      throws IOException {
+    if (doRetry) {
+      return retryUntranslated(
+          toDescription(action, path),
+          idempotent,
+          retrying,
+          () -> once(action, path, operation));
+    } else {
+      return once(action, path, operation);
+    }
+  }
+
   /**
    * Execute a function with retry processing and no translation.
    * and the default retry callback.