[DO NOT MERGE] Backport some fixes and compatibility commits from master to ozone-1.4

hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/OzoneBucket.java

@@ -39,6 +39,7 @@
 import org.apache.hadoop.ozone.OzoneAcl;
 import org.apache.hadoop.ozone.om.exceptions.OMException;
 import org.apache.hadoop.ozone.om.helpers.BasicOmKeyInfo;
+import org.apache.hadoop.ozone.om.helpers.ErrorInfo;
 import org.apache.hadoop.ozone.om.helpers.OmMultipartInfo;
 import org.apache.hadoop.ozone.om.helpers.OmMultipartUploadCompleteInfo;
 import org.apache.hadoop.ozone.om.helpers.OmKeyInfo;
@@ -62,6 +63,7 @@
 import java.util.NoSuchElementException;
 import java.util.stream.Collectors;
 
+import static org.apache.hadoop.ozone.OzoneConsts.ETAG;
 import static org.apache.hadoop.ozone.OzoneConsts.QUOTA_RESET;
 import static org.apache.hadoop.ozone.OzoneConsts.OZONE_URI_DELIMITER;
 import static org.apache.hadoop.ozone.om.exceptions.OMException.ResultCodes.FILE_NOT_FOUND;
@@ -620,6 +622,16 @@ public void deleteKeys(List<String> keyList) throws IOException {
     proxy.deleteKeys(volumeName, name, keyList);
   }
 
+  /**
+   * Deletes the given list of keys from the bucket.
+   * @param keyList List of the key name to be deleted.
+   * @param quiet flag to not throw exception if delete fails
+   * @throws IOException
+   */
+  public Map<String, ErrorInfo> deleteKeys(List<String> keyList, boolean quiet) throws IOException {
+    return proxy.deleteKeys(volumeName, name, keyList, quiet);
+  }
+
   /**
    * Rename the keyname from fromKeyName to toKeyName.
    * @param fromKeyName The original key name.
@@ -1271,25 +1283,33 @@ protected void initDelimiterKeyPrefix() {
     protected List<OzoneKey> buildKeysWithKeyPrefix(
         List<OzoneFileStatusLight> statuses) {
       return statuses.stream()
-          .map(status -> {
-            BasicOmKeyInfo keyInfo = status.getKeyInfo();
-            String keyName = keyInfo.getKeyName();
-            if (status.isDirectory()) {
-              // add trailing slash to represent directory
-              keyName = OzoneFSUtils.addTrailingSlashIfNeeded(keyName);
-            }
-            return new OzoneKey(keyInfo.getVolumeName(),
-                keyInfo.getBucketName(), keyName,
-                keyInfo.getDataSize(), keyInfo.getCreationTime(),
-                keyInfo.getModificationTime(),
-                keyInfo.getReplicationConfig(), keyInfo.isFile());
-          })
+          .map(OzoneBucket::toOzoneKey)
           .filter(key -> StringUtils.startsWith(key.getName(), getKeyPrefix()))
           .collect(Collectors.toList());
     }
 
   }
 
+  private static OzoneKey toOzoneKey(OzoneFileStatusLight status) {
+    BasicOmKeyInfo keyInfo = status.getKeyInfo();
+    String keyName = keyInfo.getKeyName();
+    final Map<String, String> metadata;
+    if (status.isDirectory()) {
+      // add trailing slash to represent directory
+      keyName = OzoneFSUtils.addTrailingSlashIfNeeded(keyName);
+      metadata = Collections.emptyMap();
+    } else {
+      metadata = Collections.singletonMap(ETAG, keyInfo.getETag());
+    }
+    return new OzoneKey(keyInfo.getVolumeName(),
+        keyInfo.getBucketName(), keyName,
+        keyInfo.getDataSize(), keyInfo.getCreationTime(),
+        keyInfo.getModificationTime(),
+        keyInfo.getReplicationConfig(),
+        metadata,
+        keyInfo.isFile());
+  }
+
 
   /**
    * An Iterator to iterate over {@link OzoneKey} list.
@@ -1657,21 +1677,7 @@ private boolean getChildrenKeys(String keyPrefix, String startKey,
       for (int indx = 0; indx < statuses.size(); indx++) {
         OzoneFileStatusLight status = statuses.get(indx);
         BasicOmKeyInfo keyInfo = status.getKeyInfo();
-        String keyName = keyInfo.getKeyName();
-
-        OzoneKey ozoneKey;
-        // Add dir to the dirList
-        if (status.isDirectory()) {
-          // add trailing slash to represent directory
-          keyName = OzoneFSUtils.addTrailingSlashIfNeeded(keyName);
-        }
-        ozoneKey = new OzoneKey(keyInfo.getVolumeName(),
-            keyInfo.getBucketName(), keyName,
-            keyInfo.getDataSize(), keyInfo.getCreationTime(),
-            keyInfo.getModificationTime(),
-            keyInfo.getReplicationConfig(),
-            keyInfo.isFile());
-
+        OzoneKey ozoneKey = toOzoneKey(status);
         keysResultList.add(ozoneKey);
 
         if (status.isDirectory()) {

hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/protocol/ClientProtocol.java

@@ -49,6 +49,7 @@
 import org.apache.hadoop.ozone.om.OMConfigKeys;
 import org.apache.hadoop.ozone.om.exceptions.OMException;
 import org.apache.hadoop.ozone.om.helpers.DeleteTenantState;
+import org.apache.hadoop.ozone.om.helpers.ErrorInfo;
 import org.apache.hadoop.ozone.om.helpers.OmKeyLocationInfo;
 import org.apache.hadoop.ozone.om.helpers.OmMultipartInfo;
 import org.apache.hadoop.ozone.om.helpers.OmMultipartUploadCompleteInfo;
@@ -404,6 +405,18 @@ void deleteKeys(String volumeName, String bucketName,
                   List<String> keyNameList)
       throws IOException;
 
+  /**
+   * Deletes keys through the list.
+   * @param volumeName Name of the Volume
+   * @param bucketName Name of the Bucket
+   * @param keyNameList List of the Key
+   * @param quiet flag to not throw exception if delete fails
+   * @throws IOException
+   */
+  Map<String, ErrorInfo> deleteKeys(String volumeName, String bucketName,
+                                    List<String> keyNameList, boolean quiet)
+      throws IOException;
+
   /**
    * Renames an existing key within a bucket.
    * @param volumeName Name of the Volume

hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/rpc/RpcClient.java

@@ -98,6 +98,7 @@
 import org.apache.hadoop.ozone.om.helpers.BucketEncryptionKeyInfo;
 import org.apache.hadoop.ozone.om.helpers.BucketLayout;
 import org.apache.hadoop.ozone.om.helpers.DeleteTenantState;
+import org.apache.hadoop.ozone.om.helpers.ErrorInfo;
 import org.apache.hadoop.ozone.om.helpers.KeyInfoWithVolumeContext;
 import org.apache.hadoop.ozone.om.helpers.OmBucketArgs;
 import org.apache.hadoop.ozone.om.helpers.OmBucketInfo;
@@ -1556,6 +1557,18 @@ public void deleteKeys(
     ozoneManagerClient.deleteKeys(omDeleteKeys);
   }
 
+  @Override
+  public Map<String, ErrorInfo> deleteKeys(
+      String volumeName, String bucketName, List<String> keyNameList, boolean quiet)
+      throws IOException {
+    verifyVolumeName(volumeName);
+    verifyBucketName(bucketName);
+    Preconditions.checkNotNull(keyNameList);
+    OmDeleteKeys omDeleteKeys = new OmDeleteKeys(volumeName, bucketName,
+        keyNameList);
+    return ozoneManagerClient.deleteKeys(omDeleteKeys, quiet);
+  }
+
   @Override
   public void renameKey(String volumeName, String bucketName,
       String fromKeyName, String toKeyName) throws IOException {

hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/helpers/BasicOmKeyInfo.java

@@ -20,11 +20,14 @@
 import java.io.IOException;
 import java.util.Objects;
 
+import org.apache.commons.lang3.StringUtils;
 import org.apache.hadoop.hdds.client.ECReplicationConfig;
 import org.apache.hadoop.hdds.client.ReplicationConfig;
 import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.BasicKeyInfo;
 import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.ListKeysRequest;
 
+import static org.apache.hadoop.ozone.OzoneConsts.ETAG;
+
 /**
  * Lightweight OmKeyInfo class.
  */
@@ -38,11 +41,12 @@ public class BasicOmKeyInfo {
   private long modificationTime;
   private ReplicationConfig replicationConfig;
   private boolean isFile;
+  private final String eTag;
 
   @SuppressWarnings("parameternumber")
   public BasicOmKeyInfo(String volumeName, String bucketName, String keyName,
                         long dataSize, long creationTime, long modificationTime,
-                        ReplicationConfig replicationConfig, boolean isFile) {
+                        ReplicationConfig replicationConfig, boolean isFile, String eTag) {
     this.volumeName = volumeName;
     this.bucketName = bucketName;
     this.keyName = keyName;
@@ -51,6 +55,7 @@ public BasicOmKeyInfo(String volumeName, String bucketName, String keyName,
     this.modificationTime = modificationTime;
     this.replicationConfig = replicationConfig;
     this.isFile = isFile;
+    this.eTag = StringUtils.isNotEmpty(eTag) ? eTag : null;
   }
 
   public String getVolumeName() {
@@ -85,6 +90,10 @@ public boolean isFile() {
     return isFile;
   }
 
+  public String getETag() {
+    return eTag;
+  }
+
   /**
    * Builder of BasicOmKeyInfo.
    */
@@ -97,6 +106,7 @@ public static class Builder {
     private long modificationTime;
     private ReplicationConfig replicationConfig;
     private boolean isFile;
+    private String eTag;
 
     public Builder setVolumeName(String volumeName) {
       this.volumeName = volumeName;
@@ -138,9 +148,14 @@ public Builder setIsFile(boolean isFile) {
       return this;
     }
 
+    public Builder setETag(String etag) {
+      this.eTag = etag;
+      return this;
+    }
+
     public BasicOmKeyInfo build() {
       return new BasicOmKeyInfo(volumeName, bucketName, keyName, dataSize,
-          creationTime, modificationTime, replicationConfig, isFile);
+          creationTime, modificationTime, replicationConfig, isFile, eTag);
     }
   }
 
@@ -157,6 +172,9 @@ public BasicKeyInfo getProtobuf() {
     } else {
       builder.setFactor(ReplicationConfig.getLegacyFactor(replicationConfig));
     }
+    if (StringUtils.isNotEmpty(eTag)) {
+      builder.setETag(eTag);
+    }
 
     return builder.build();
   }
@@ -181,6 +199,7 @@ public static BasicOmKeyInfo getFromProtobuf(BasicKeyInfo basicKeyInfo,
             basicKeyInfo.getType(),
             basicKeyInfo.getFactor(),
             basicKeyInfo.getEcReplicationConfig()))
+        .setETag(basicKeyInfo.getETag())
         .setIsFile(!keyName.endsWith("/"));
 
     return builder.build();
@@ -205,6 +224,7 @@ public static BasicOmKeyInfo getFromProtobuf(String volumeName,
             basicKeyInfo.getType(),
             basicKeyInfo.getFactor(),
             basicKeyInfo.getEcReplicationConfig()))
+        .setETag(basicKeyInfo.getETag())
         .setIsFile(!keyName.endsWith("/"));
 
     return builder.build();
@@ -225,6 +245,7 @@ public boolean equals(Object o) {
         creationTime == basicOmKeyInfo.creationTime &&
         modificationTime == basicOmKeyInfo.modificationTime &&
         replicationConfig.equals(basicOmKeyInfo.replicationConfig) &&
+        Objects.equals(eTag, basicOmKeyInfo.eTag) &&
         isFile == basicOmKeyInfo.isFile;
   }
 
@@ -241,6 +262,7 @@ public static BasicOmKeyInfo fromOmKeyInfo(OmKeyInfo omKeyInfo) {
         omKeyInfo.getCreationTime(),
         omKeyInfo.getModificationTime(),
         omKeyInfo.getReplicationConfig(),
-        omKeyInfo.isFile());
+        omKeyInfo.isFile(),
+        omKeyInfo.getMetadata().get(ETAG));
   }
 }

hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/helpers/ErrorInfo.java

@@ -0,0 +1,49 @@
+/**
+ * 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
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * 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.ozone.om.helpers;
+
+/**
+ * Represent class which has info of error thrown for any operation.
+ */
+public class ErrorInfo {
+  private String code;
+  private String message;
+
+  public ErrorInfo(String errorCode, String errorMessage) {
+    this.code = errorCode;
+    this.message = errorMessage;
+  }
+
+  public String getCode() {
+    return code;
+  }
+
+  public void setCode(String errorCode) {
+    this.code = errorCode;
+  }
+
+  public String getMessage() {
+    return message;
+  }
+
+  public void setMessage(String errorMessage) {
+    this.message = errorMessage;
+  }
+
+}

hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/protocol/OzoneManagerProtocol.java

@@ -21,6 +21,7 @@
 import java.io.Closeable;
 import java.io.IOException;
 import java.util.List;
+import java.util.Map;
 import java.util.UUID;
 
 import javax.annotation.Nonnull;
@@ -32,6 +33,7 @@
 import org.apache.hadoop.ozone.om.exceptions.OMException;
 import org.apache.hadoop.ozone.om.helpers.DBUpdates;
 import org.apache.hadoop.ozone.om.helpers.DeleteTenantState;
+import org.apache.hadoop.ozone.om.helpers.ErrorInfo;
 import org.apache.hadoop.ozone.om.helpers.KeyInfoWithVolumeContext;
 import org.apache.hadoop.ozone.om.helpers.OmBucketArgs;
 import org.apache.hadoop.ozone.om.helpers.OmBucketInfo;
@@ -359,6 +361,21 @@ default void deleteKeys(OmDeleteKeys deleteKeys) throws IOException {
         "this to be implemented, as write requests use a new approach.");
   }
 
+  /**
+   * Deletes existing key/keys. This interface supports delete
+   * multiple keys and a single key. Used by deleting files
+   * through OzoneFileSystem.
+   *
+   * @param deleteKeys
+   * @param quiet - flag to not throw exception if delete fails
+   * @throws IOException
+   */
+  default Map<String, ErrorInfo> deleteKeys(OmDeleteKeys deleteKeys, boolean quiet)
+      throws IOException {
+    throw new UnsupportedOperationException("OzoneManager does not require " +
+        "this to be implemented, as write requests use a new approach.");
+  }
+
 
   /**
    * Deletes an existing empty bucket from volume.