HDDS-11661. ITestS3AContractBulkDelete fails with FSO bucket

hadoop-hdds/common/src/main/java/org/apache/hadoop/ozone/ClientVersion.java

@@ -42,6 +42,9 @@ public enum ClientVersion implements ComponentVersion {
       "This client version has support for Object Store and File " +
           "System Optimized Bucket Layouts."),
 
+  FSO_BULK_DELETE(4,
+      "This client version guarantees that the directory path ends with slash when deleting keys."),
+
   FUTURE_VERSION(-1, "Used internally when the server side is older and an"
       + " unknown client version has arrived from the client.");
 

hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/key/OMKeysDeleteRequest.java

@@ -174,7 +174,7 @@ public OMClientResponse validateAndUpdateCache(OzoneManager ozoneManager, TermIn
           OzoneFileStatus fileStatus = getOzoneKeyStatus(
               ozoneManager, omMetadataManager, volumeName, bucketName, keyName);
           addKeyToAppropriateList(omKeyInfoList, omKeyInfo, dirList,
-              fileStatus);
+              fileStatus, getOmRequest().getVersion());
         } catch (Exception ex) {
           deleteStatus = false;
           LOG.error("Acl check failed for Key: {}", objectKey, ex);
@@ -331,7 +331,7 @@ protected long markKeysAsDeletedInCache(OzoneManager ozoneManager,
   }
 
   protected void addKeyToAppropriateList(List<OmKeyInfo> omKeyInfoList,
-      OmKeyInfo omKeyInfo, List<OmKeyInfo> dirList, OzoneFileStatus keyStatus) {
+      OmKeyInfo omKeyInfo, List<OmKeyInfo> dirList, OzoneFileStatus keyStatus, int version) {
     omKeyInfoList.add(omKeyInfo);
   }
 

hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/key/OmKeysDeleteRequestWithFSO.java

@@ -20,6 +20,7 @@
 import org.apache.hadoop.hdds.utils.db.Table;
 import org.apache.hadoop.hdds.utils.db.cache.CacheKey;
 import org.apache.hadoop.hdds.utils.db.cache.CacheValue;
+import org.apache.hadoop.ozone.ClientVersion;
 import org.apache.hadoop.ozone.OzoneConsts;
 import org.apache.hadoop.ozone.om.OMMetadataManager;
 import org.apache.hadoop.ozone.om.OzoneManager;
@@ -41,6 +42,7 @@
 import java.util.List;
 import java.util.Map;
 
+import static org.apache.hadoop.hdds.scm.net.NetConstants.PATH_SEPARATOR_STR;
 import static org.apache.hadoop.ozone.OzoneConsts.DELETED_HSYNC_KEY;
 import static org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.Status.OK;
 import static org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.Status.PARTIAL_DELETE;
@@ -70,9 +72,14 @@ protected OmKeyInfo getOmKeyInfo(
 
   @Override
   protected void addKeyToAppropriateList(List<OmKeyInfo> omKeyInfoList,
-      OmKeyInfo omKeyInfo, List<OmKeyInfo> dirList, OzoneFileStatus keyStatus) {
+      OmKeyInfo omKeyInfo, List<OmKeyInfo> dirList, OzoneFileStatus keyStatus, int version) {
+    LOG.info("client version: {}", version);
     if (keyStatus.isDirectory()) {
-      dirList.add(omKeyInfo);
+      if (ClientVersion.fromProtoValue(version).compareTo(ClientVersion.FSO_BULK_DELETE) < 0) {
+        dirList.add(omKeyInfo);
+      } else if (keyStatus.getKeyInfo().getKeyName().endsWith(PATH_SEPARATOR_STR)) {
+        dirList.add(omKeyInfo);
+      }
     } else {
       omKeyInfoList.add(omKeyInfo);
     }

hadoop-ozone/ozonefs-common/src/main/java/org/apache/hadoop/fs/ozone/BasicRootedOzoneFileSystem.java

@@ -1458,6 +1458,9 @@ boolean iterate() throws IOException {
             String keyName =
                 new OFSPath(fileStatus.getPath().toString(),
                     ozoneConfiguration).getKeyName();
+            if (fileStatus.isDir()) {
+              keyName = OzoneFSUtils.addTrailingSlashIfNeeded(keyName);
+            }
             keyPathList.add(ofsPathPrefix + keyName);
           }
           if (keyPathList.size() >= batchSize) {