HADOOP-19197. S3A: KMS Encryption Context support: follow-up

Followup the main HADOOP-19197 patch to address serialization
and compilation issues

* Recreate serialization ID
* Restore two arg constructor
* Define DEFAULT_S3_ENCRYPTION_CONTEXT to specify what the
  default value is (just "", but being explicit)
* Restore ability to unmarshal old version encryption secrets.
* Tests

This allows for YARN services to load DTs supplied by older releases.

If they marshall the secrets again the fact they were the older version
is lost, they get upgraded. This may complicate any worker node launch
where the DT list is modified before passing to the launched process

Author: raphaelazzolini

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

@@ -786,6 +786,13 @@ private Constants() {
   public static final String S3_ENCRYPTION_CONTEXT =
       "fs.s3a.encryption.context";
 
+  /**
+   * Default S3-SSE encryption context.
+   * value:{@value}
+   */
+  public static final String DEFAULT_S3_ENCRYPTION_CONTEXT =
+      "";
+
   /**
    * Client side encryption (CSE-CUSTOM) with custom cryptographic material manager class name.
    * Custom keyring class name for CSE-KMS.

hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/auth/delegation/EncryptionSecrets.java

@@ -25,12 +25,18 @@
 import java.io.Serializable;
 import java.util.Objects;
 
+import com.google.common.annotations.VisibleForTesting;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
 import org.apache.commons.lang3.StringUtils;
 import org.apache.hadoop.fs.s3a.S3AEncryptionMethods;
 import org.apache.hadoop.io.LongWritable;
 import org.apache.hadoop.io.Text;
 import org.apache.hadoop.io.Writable;
 
+import static org.apache.hadoop.fs.s3a.Constants.DEFAULT_S3_ENCRYPTION_CONTEXT;
+
 /**
  * Encryption options in a form which can serialized or marshalled as a hadoop
  * Writeable.
@@ -52,9 +58,24 @@
  */
 public class EncryptionSecrets implements Writable, Serializable {
 
+  private static final Logger LOG =
+          LoggerFactory.getLogger(EncryptionSecrets.class);
+
   public static final int MAX_SECRET_LENGTH = 2048;
 
-  private static final long serialVersionUID = 1208329045511296375L;
+  /**
+   * Change this after any change to the payload: {@value}.
+   */
+  private static final long serialVersionUID = 8834417969966697162L;
+
+  @VisibleForTesting
+  public static final long SERIAL_VERSION_UID_CURRENT = serialVersionUID;
+
+  /**
+   * Serial version ID prior to {@link #encryptionContext} field being added: {@value}.
+   */
+  @VisibleForTesting
+  public static final long SERIAL_VERSION_UID_1 = 1208329045511296375L;
 
   /**
    * Encryption algorithm to use: must match one in
@@ -70,7 +91,7 @@ public class EncryptionSecrets implements Writable, Serializable {
   /**
    * Encryption context: base64-encoded UTF-8 string.
    */
-  private String encryptionContext = "";
+  private String encryptionContext = DEFAULT_S3_ENCRYPTION_CONTEXT;
 
   /**
    * This field isn't serialized/marshalled; it is rebuilt from the
@@ -86,7 +107,24 @@ public EncryptionSecrets() {
   }
 
   /**
-   * Create a pair of secrets.
+   * Create a tuple of secrets. The encryption context is set to "".
+   * This constructor is used in external implementations of S3A delegation
+   * tokens, sp MUST be retained even if there is no use in our own
+   * production code.
+   * @param encryptionAlgorithm algorithm enumeration.
+   * @param encryptionKey key/key reference.
+   * @throws IOException failure to initialize.
+   * @deprecated use {@link #EncryptionSecrets(S3AEncryptionMethods, String, String)}
+   * which takes an encryption context.
+   */
+  public EncryptionSecrets(final S3AEncryptionMethods encryptionAlgorithm,
+      final String encryptionKey) throws IOException {
+    this(encryptionAlgorithm.getMethod(), encryptionKey,
+        DEFAULT_S3_ENCRYPTION_CONTEXT);
+  }
+
+  /**
+   * Create a 3/tuple of secrets.
    * @param encryptionAlgorithm algorithm enumeration.
    * @param encryptionKey key/key reference.
    * @param encryptionContext  base64-encoded string with the encryption context key-value pairs.
@@ -99,7 +137,7 @@ public EncryptionSecrets(final S3AEncryptionMethods encryptionAlgorithm,
   }
 
   /**
-   * Create a pair of secrets.
+   * Create a 3/tuple of secrets.
    * @param encryptionAlgorithm algorithm name
    * @param encryptionKey key/key reference.
    * @param encryptionContext  base64-encoded string with the encryption context key-value pairs.
@@ -137,13 +175,26 @@ public void write(final DataOutput out) throws IOException {
   public void readFields(final DataInput in) throws IOException {
     final LongWritable version = new LongWritable();
     version.readFields(in);
-    if (version.get() != serialVersionUID) {
+    boolean readContext;
+
+    final long versionId = version.get();
+    if (versionId == SERIAL_VERSION_UID_1) {
+      LOG.info("Unmarshalling Encryption Secrets from older client; "
+          + "setting encryption context to \"\"");
+      readContext = false;
+    } else if (versionId == serialVersionUID) {
+      readContext = true;
+    } else {
       throw new DelegationTokenIOException(
-          "Incompatible EncryptionSecrets version");
+          "Incompatible EncryptionSecrets version: " + versionId);
     }
     encryptionAlgorithm = Text.readString(in, MAX_SECRET_LENGTH);
     encryptionKey = Text.readString(in, MAX_SECRET_LENGTH);
-    encryptionContext = Text.readString(in);
+    if (readContext) {
+      encryptionContext = Text.readString(in);
+    } else {
+      encryptionContext = DEFAULT_S3_ENCRYPTION_CONTEXT;
+    }
     init();
   }
 
@@ -168,14 +219,26 @@ private void init() throws IOException {
         encryptionAlgorithm);
   }
 
+  /**
+   * Get the encryption algorithm
+   * @return the encryption algorithm
+   */
   public String getEncryptionAlgorithm() {
     return encryptionAlgorithm;
   }
 
+  /**
+   * Get the encryption key
+   * @return the encryption key
+   */
   public String getEncryptionKey() {
     return encryptionKey;
   }
 
+  /**
+   * Get the encryption context
+   * @return the encryption context
+   */
   public String getEncryptionContext() {
     return encryptionContext;
   }

hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/commit/files/PendingSet.java

@@ -20,7 +20,6 @@
 
 import javax.annotation.Nullable;
 import java.io.IOException;
-import java.io.ObjectInputStream;
 import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.HashSet;
@@ -152,20 +151,6 @@ public void add(SinglePendingCommit commit) {
     }
   }
 
-  /**
-   * Deserialize via java Serialization API: deserialize the instance
-   * and then call {@link #validate()} to verify that the deserialized
-   * data is valid.
-   * @param inStream input stream
-   * @throws IOException IO problem or validation failure
-   * @throws ClassNotFoundException reflection problems
-   */
-  private void readObject(ObjectInputStream inStream) throws IOException,
-      ClassNotFoundException {
-    inStream.defaultReadObject();
-    validate();
-  }
-
   /**
    * Validate the data: those fields which must be non empty, must be set.
    * @throws ValidationFailure if the data is invalid

hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/commit/files/SinglePendingCommit.java

@@ -20,7 +20,6 @@
 
 import javax.annotation.Nullable;
 import java.io.IOException;
-import java.io.ObjectInputStream;
 import java.io.Serializable;
 import java.net.URI;
 import java.net.URISyntaxException;
@@ -186,21 +185,6 @@ public static SinglePendingCommit load(FileSystem fs,
     return instance;
   }
 
-  /**
-   * Deserialize via java Serialization API: deserialize the instance
-   * and then call {@link #validate()} to verify that the deserialized
-   * data is valid.
-   * @param inStream input stream
-   * @throws IOException IO problem
-   * @throws ClassNotFoundException reflection problems
-   * @throws ValidationFailure validation failure
-   */
-  private void readObject(ObjectInputStream inStream) throws IOException,
-      ClassNotFoundException {
-    inStream.defaultReadObject();
-    validate();
-  }
-
   /**
    * Set the various timestamp fields to the supplied value.
    * @param millis time in milliseconds

hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/commit/files/UploadEtag.java

@@ -25,7 +25,7 @@
 import software.amazon.awssdk.services.s3.model.CompletedPart;
 
 /**
- * Stores ETag and checksum values from {@link  CompletedPart} responses from S3.
+ * Stores ETag and checksum values from {@link CompletedPart} responses from S3.
  * These values need to be stored to be later passed to the
  * {@link software.amazon.awssdk.services.s3.model.CompleteMultipartUploadRequest
  * CompleteMultipartUploadRequest}
@@ -37,8 +37,17 @@ public class UploadEtag implements Serializable {
    */
   private static final long serialVersionUID = 1L;
 
+  /**
+   * CompletedPart's ETag value.
+   */
   private String etag;
+  /**
+   * Algorithm used to calculate the checksum.
+   */
   private String checksumAlgorithm;
+  /**
+   * CompletedPart's checksum value.
+   */
   private String checksum;
 
   public UploadEtag() {

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

@@ -31,6 +31,7 @@
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.s3a.S3AUtils;
 
+import static org.apache.hadoop.fs.s3a.Constants.DEFAULT_S3_ENCRYPTION_CONTEXT;
 import static org.apache.hadoop.fs.s3a.Constants.S3_ENCRYPTION_CONTEXT;
 
 /**
@@ -61,7 +62,7 @@ public static String getS3EncryptionContext(String bucket, Configuration conf)
     }
     if (encryptionContext == null) {
       // no encryption context, return ""
-      return "";
+      return DEFAULT_S3_ENCRYPTION_CONTEXT;
     }
     return encryptionContext;
   }