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

Contributed by Ben Roling.

S3Guard will now track the etag of uploaded files and, if an S3
bucket is versioned, the object version.

You can then control how to react to a mismatch between the data
in the DynamoDB table and that in the store: warn, fail, or, when
using versions, return the original value.

This adds two new columns to the table: etag and version.
This is transparent to older S3A clients -but when such clients
add/update data to the S3Guard table, they will not add these values.
As a result, the etag/version checks will not work with files uploaded by older clients.

For a consistent experience, upgrade all clients to use the latest hadoop version.

Author: Ben Roling

hadoop-tools/hadoop-aws/pom.xml

@@ -405,6 +405,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.

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

@@ -32,6 +32,9 @@
 @InterfaceStability.Unstable
 public class RemoteFileChangedException extends PathIOException {
 
+  public static final String PRECONDITIONS_FAILED =
+      "Constraints of request were unsatisfiable";
+
   /**
    * Constructs a RemoteFileChangedException.
    *
@@ -46,4 +49,21 @@ public RemoteFileChangedException(String path,
     super(path, message);
     setOperation(operation);
   }
+
+  /**
+   * Constructs a RemoteFileChangedException.
+   *
+   * @param path the path accessed when the change was detected
+   * @param operation the operation (e.g. open, re-open) performed when the
+   * change was detected
+   * @param message a message providing more details about the condition
+   * @param cause inner cause.
+   */
+  public RemoteFileChangedException(String path,
+      String operation,
+      String message,
+      Throwable cause) {
+    super(path, message, cause);
+    setOperation(operation);
+  }
 }

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

@@ -32,6 +32,8 @@
 @InterfaceStability.Evolving
 public class S3AFileStatus extends FileStatus {
   private Tristate isEmptyDirectory;
+  private String eTag;
+  private String versionId;
 
   /**
    * Create a directory status.
@@ -69,15 +71,17 @@ public S3AFileStatus(Tristate isemptydir,
    * @param path path
    * @param blockSize block size
    * @param owner owner
+   * @param eTag eTag of the S3 object if available, else null
+   * @param versionId versionId of the S3 object if available, else null
    */
   public S3AFileStatus(long length, long modification_time, Path path,
-      long blockSize, String owner) {
-    super(length, false, 1, blockSize, modification_time, 0,
-        null, null, null, null,
+      long blockSize, String owner, String eTag, String versionId) {
+    super(length, false, 1, blockSize, modification_time,
+        0, null, owner, owner, null,
         path, false, true, false);
     isEmptyDirectory = Tristate.FALSE;
-    setOwner(owner);
-    setGroup(owner);
+    this.eTag = eTag;
+    this.versionId = versionId;
   }
 
   /**
@@ -86,16 +90,19 @@ public S3AFileStatus(long length, long modification_time, Path path,
    * @param source FileStatus to convert to S3AFileStatus
    * @param isEmptyDirectory TRUE/FALSE if known to be / not be an empty
    *     directory, UNKNOWN if that information was not computed.
+   * @param eTag eTag of the S3 object if available, else null
+   * @param versionId versionId of the S3 object if available, else null
    * @return a new S3AFileStatus
    */
   public static S3AFileStatus fromFileStatus(FileStatus source,
-      Tristate isEmptyDirectory) {
+      Tristate isEmptyDirectory, String eTag, String versionId) {
     if (source.isDirectory()) {
       return new S3AFileStatus(isEmptyDirectory, source.getPath(),
           source.getOwner());
     } else {
       return new S3AFileStatus(source.getLen(), source.getModificationTime(),
-          source.getPath(), source.getBlockSize(), source.getOwner());
+          source.getPath(), source.getBlockSize(), source.getOwner(),
+          eTag, versionId);
     }
   }
 
@@ -109,6 +116,28 @@ public Tristate isEmptyDirectory() {
     return isEmptyDirectory;
   }
 
+  /**
+   * Update isEmptyDirectory attribute.
+   * @param isEmptyDirectory new isEmptyDirectory value
+   */
+  public void setIsEmptyDirectory(Tristate isEmptyDirectory) {
+    this.isEmptyDirectory = isEmptyDirectory;
+  }
+
+  /**
+   * @return the S3 object eTag when available, else null.
+   */
+  public String getETag() {
+    return eTag;
+  }
+
+  /**
+   * @return the S3 object versionId when available, else null.
+   */
+  public String getVersionId() {
+    return versionId;
+  }
+
   /** Compare if this object is equal to another object.
    * @param   o the object to be compared.
    * @return  true if two file status has the same path name; false if not.
@@ -150,8 +179,10 @@ public long getModificationTime(){
 
   @Override
   public String toString() {
-    return super.toString() +
-        String.format(" isEmptyDirectory=%s", isEmptyDirectory().name());
+    return super.toString()
+        + String.format(" isEmptyDirectory=%s", isEmptyDirectory().name()
+        + String.format(" eTag=%s", eTag)
+        + String.format(" versionId=%s", versionId));
   }
 
 }

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

@@ -0,0 +1,63 @@
+/**
+ * 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
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * 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;
+
+import org.apache.hadoop.fs.BlockLocation;
+import org.apache.hadoop.fs.LocatedFileStatus;
+
+import static com.google.common.base.Preconditions.checkNotNull;
+
+/**
+ * {@link LocatedFileStatus} extended to also carry ETag and object version ID.
+ */
+public class S3ALocatedFileStatus extends LocatedFileStatus {
+
+  private static final long serialVersionUID = 3597192103662929338L;
+
+  private final String eTag;
+  private final String versionId;
+
+  public S3ALocatedFileStatus(S3AFileStatus status, BlockLocation[] locations,
+      String eTag, String versionId) {
+    super(checkNotNull(status), locations);
+    this.eTag = eTag;
+    this.versionId = versionId;
+  }
+
+  public String getETag() {
+    return eTag;
+  }
+
+  public String getVersionId() {
+    return versionId;
+  }
+
+  // equals() and hashCode() overridden to avoid FindBugs warning.
+  // Base implementation is equality on Path only, which is still appropriate.
+
+  @Override
+  public boolean equals(Object o) {
+    return super.equals(o);
+  }
+
+  @Override
+  public int hashCode() {
+    return super.hashCode();
+  }
+}

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

@@ -531,16 +531,20 @@ public static String stringify(AmazonS3Exception e) {
    * @param summary summary from AWS
    * @param blockSize block size to declare.
    * @param owner owner of the file
+   * @param eTag S3 object eTag or null if unavailable
+   * @param versionId S3 object versionId or null if unavailable
    * @return a status entry
    */
   public static S3AFileStatus createFileStatus(Path keyPath,
       S3ObjectSummary summary,
       long blockSize,
-      String owner) {
+      String owner,
+      String eTag,
+      String versionId) {
     long size = summary.getSize();
     return createFileStatus(keyPath,
         objectRepresentsDirectory(summary.getKey(), size),
-        size, summary.getLastModified(), blockSize, owner);
+        size, summary.getLastModified(), blockSize, owner, eTag, versionId);
   }
 
   /**
@@ -553,22 +557,27 @@ public static S3AFileStatus createFileStatus(Path keyPath,
    * @param size file length
    * @param blockSize block size for file status
    * @param owner Hadoop username
+   * @param eTag S3 object eTag or null if unavailable
+   * @param versionId S3 object versionId or null if unavailable
    * @return a status entry
    */
   public static S3AFileStatus createUploadFileStatus(Path keyPath,
-      boolean isDir, long size, long blockSize, String owner) {
+      boolean isDir, long size, long blockSize, String owner,
+      String eTag, String versionId) {
     Date date = isDir ? null : new Date();
-    return createFileStatus(keyPath, isDir, size, date, blockSize, owner);
+    return createFileStatus(keyPath, isDir, size, date, blockSize, owner,
+        eTag, versionId);
   }
 
   /* Date 'modified' is ignored when isDir is true. */
   private static S3AFileStatus createFileStatus(Path keyPath, boolean isDir,
-      long size, Date modified, long blockSize, String owner) {
+      long size, Date modified, long blockSize, String owner,
+      String eTag, String versionId) {
     if (isDir) {
       return new S3AFileStatus(Tristate.UNKNOWN, keyPath, owner);
     } else {
       return new S3AFileStatus(size, dateToLong(modified), keyPath, blockSize,
-          owner);
+          owner, eTag, versionId);
     }
   }
 

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

@@ -42,6 +42,7 @@ public S3GuardExistsRetryPolicy(Configuration conf) {
   protected Map<Class<? extends Exception>, RetryPolicy> createExceptionMap() {
     Map<Class<? extends Exception>, RetryPolicy> b = super.createExceptionMap();
     b.put(FileNotFoundException.class, retryIdempotentCalls);
+    b.put(RemoteFileChangedException.class, retryIdempotentCalls);
     return b;
   }
 }