HDDS-12911. Key deletion should not validate the name

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

@@ -521,6 +521,20 @@ public static String validateAndNormalizeKey(boolean enableFileSystemPaths,
     }
   }
 
+  /**
+   * Normalizes the key path based on the bucket layout.  This should be used for existing keys. 
+   * For new key creation, please see {@link #validateAndNormalizeKey(boolean, String, BucketLayout)}
+   *
+   * @return normalized key path
+   */
+  public static String normalizeKeyPath(boolean enableFileSystemPaths,
+      String keyPath, BucketLayout bucketLayout) throws OMException {
+    if (bucketLayout.shouldNormalizePaths(enableFileSystemPaths)) {
+      keyPath = OmUtils.normalizeKey(keyPath, false);
+    }
+    return keyPath;
+  }
+
   public static String validateAndNormalizeKey(boolean enableFileSystemPaths,
       String keyPath, BucketLayout bucketLayout) throws OMException {
     LOG.debug("Bucket Layout: {}", bucketLayout);

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

@@ -83,8 +83,7 @@ public OMRequest preExecute(OzoneManager ozoneManager) throws IOException {
     String keyPath = keyArgs.getKeyName();
 
     OmUtils.verifyKeyNameWithSnapshotReservedWordForDeletion(keyPath);
-    keyPath = validateAndNormalizeKey(ozoneManager.getEnableFileSystemPaths(),
-        keyPath, getBucketLayout());
+    keyPath = normalizeKeyPath(ozoneManager.getEnableFileSystemPaths(), keyPath, getBucketLayout());
 
     OzoneManagerProtocolProtos.KeyArgs.Builder newKeyArgs =
         keyArgs.toBuilder().setModificationTime(Time.now()).setKeyName(keyPath);

hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/request/key/TestOMKeyDeleteRequest.java

@@ -172,7 +172,7 @@ protected OMRequest createDeleteKeyRequest() {
     return createDeleteKeyRequest(keyName);
   }
 
-  private OMRequest createDeleteKeyRequest(String testKeyName) {
+  protected OMRequest createDeleteKeyRequest(String testKeyName) {
     KeyArgs keyArgs = KeyArgs.newBuilder().setBucketName(bucketName)
         .setVolumeName(volumeName).setKeyName(testKeyName).build();
 

hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/request/key/TestOMKeyDeleteRequestWithFSO.java

@@ -21,8 +21,10 @@
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertFalse;
 import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertNull;
 import static org.junit.jupiter.api.Assertions.assertThrows;
 import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.Mockito.when;
 
 import java.io.IOException;
 import java.util.Iterator;
@@ -31,9 +33,12 @@
 import org.apache.hadoop.ozone.om.OzonePrefixPathImpl;
 import org.apache.hadoop.ozone.om.exceptions.OMException;
 import org.apache.hadoop.ozone.om.helpers.BucketLayout;
+import org.apache.hadoop.ozone.om.helpers.OmDirectoryInfo;
 import org.apache.hadoop.ozone.om.helpers.OmKeyInfo;
 import org.apache.hadoop.ozone.om.helpers.OzoneFileStatus;
 import org.apache.hadoop.ozone.om.request.OMRequestTestUtils;
+import org.apache.hadoop.ozone.om.response.OMClientResponse;
+import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos;
 import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.OMRequest;
 import org.apache.hadoop.ozone.security.acl.OzonePrefixPath;
 import org.junit.jupiter.api.Test;
@@ -227,4 +232,31 @@ public void testRecursiveAccessCheck() throws Exception {
     // false for file1.
     assertFalse(pathViewer.isCheckRecursiveAccess());
   }
+
+  @Test
+  public void testDeleteDirectoryWithColonInFSOBucket() throws Exception {
+    when(ozoneManager.getEnableFileSystemPaths()).thenReturn(true);
+    OMRequestTestUtils.addVolumeAndBucketToDB(volumeName, bucketName, omMetadataManager, getBucketLayout());
+
+    String dirName = "foo:dir/";
+    String dirKeyPath = addKeyToDirTable(volumeName, bucketName, dirName);
+
+    long parentObjectID = 0L;
+    long dirObjectID = 12345L;
+    OmDirectoryInfo omDirectoryInfo = OMRequestTestUtils.createOmDirectoryInfo(dirName, dirObjectID, parentObjectID);
+    omMetadataManager.getDirectoryTable().put(dirKeyPath, omDirectoryInfo);
+
+    OmDirectoryInfo storedDirInfo = omMetadataManager.getDirectoryTable().get(dirKeyPath);
+    assertNotNull(storedDirInfo);
+    assertEquals(dirName, storedDirInfo.getName());
+    assertEquals(dirObjectID, storedDirInfo.getObjectID());
+    assertEquals(parentObjectID, storedDirInfo.getParentObjectID());
+
+    OMRequest deleteRequest = doPreExecute(createDeleteKeyRequest(dirName));
+    OMKeyDeleteRequest omKeyDeleteRequest = getOmKeyDeleteRequest(deleteRequest);
+    OMClientResponse response = omKeyDeleteRequest.validateAndUpdateCache(ozoneManager, 100L);
+
+    assertEquals(OzoneManagerProtocolProtos.Status.OK, response.getOMResponse().getStatus());
+    assertNull(omMetadataManager.getDirectoryTable().get(dirName));
+  }
 }