Fix the issue to get a tag of a non-existed in OSS/COS

### What changes are proposed in this pull request?

Add exception handling for OSS and COS ufs in getting object tags and other apis.

### Why are the changes needed?
When handling exceptions, we need to identify exceptions that can be allowed, and other errors need to be converted to an Alluxio exception.

### Does this PR introduce any user facing changes?

Please list the user-facing changes introduced by your change, including
  1. change in user-facing APIs
  2. addition or removal of property keys
  3. webui

			pr-link: Alluxio#18388
			change-id: cid-2eb458f3ec6955321981dff350dacc9f33104c1b

dora/core/server/worker/src/main/java/alluxio/worker/dora/DoraMetaManager.java

@@ -117,7 +117,7 @@ public Optional<FileStatus> getFromUfs(String path) throws IOException {
       UfsStatus status = ufs.getStatus(path,
           GetStatusOptions.defaults().setIncludeRealContentHash(mGetRealContentHash));
       Map<String, String> xattrMap = null;
-      if (mXAttrWriteToUFSEnabled) {
+      if (status != null && mXAttrWriteToUFSEnabled) {
         xattrMap = ufs.getAttributes(path);
       }
       DoraMeta.FileStatus fs = PagedDoraWorker.buildFileStatusFromUfsStatus(

dora/underfs/cos/src/main/java/alluxio/underfs/cos/AlluxioCosException.java

@@ -0,0 +1,100 @@
+/*
+ * The Alluxio Open Foundation licenses this work under the Apache License, version 2.0
+ * (the "License"). You may not use this work except in compliance with the License, which is
+ * available at www.apache.org/licenses/LICENSE-2.0
+ *
+ * This software is distributed on an "AS IS" basis, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND,
+ * either express or implied, as more fully set forth in the License.
+ *
+ * See the NOTICE file distributed with this work for information regarding copyright ownership.
+ */
+
+package alluxio.underfs.cos;
+
+import alluxio.exception.runtime.AlluxioRuntimeException;
+import alluxio.grpc.ErrorType;
+
+import com.qcloud.cos.exception.CosClientException;
+import com.qcloud.cos.exception.CosServiceException;
+import io.grpc.Status;
+
+import java.net.HttpURLConnection;
+
+/**
+ * Alluxio exception for cos.
+ */
+public class AlluxioCosException extends AlluxioRuntimeException {
+  private static final ErrorType ERROR_TYPE = ErrorType.External;
+
+  /**
+   * Converts an AmazonClientException to a corresponding AlluxioCosException.
+   * @param cause cos exception
+   * @return alluxio cos exception
+   */
+  public static AlluxioCosException from(CosClientException cause) {
+    return from(null, cause);
+  }
+
+  /**
+   * Converts an CosClientException with errormessage to a corresponding AlluxioCosException.
+   * @param errorMessage error message
+   * @param cause cos exception
+   * @return alluxio cos exception
+   */
+  public static AlluxioCosException from(String errorMessage, CosClientException cause) {
+    Status status = Status.UNKNOWN;
+    String errorDescription = "ClientException:" + cause.getMessage();
+    if (cause instanceof CosServiceException) {
+      CosServiceException exception = (CosServiceException) cause;
+      status = httpStatusToGrpcStatus(exception.getStatusCode());
+      errorDescription = exception.getErrorCode() + ":" + exception.getErrorMessage();
+    }
+    if (errorMessage == null) {
+      errorMessage = errorDescription;
+    }
+    return new AlluxioCosException(status, errorMessage, cause, cause.isRetryable());
+  }
+
+  private AlluxioCosException(Status status, String message, Throwable cause, boolean isRetryAble) {
+    super(status, message, cause, ERROR_TYPE, isRetryAble);
+  }
+
+  private static Status httpStatusToGrpcStatus(int httpStatusCode) {
+    if (httpStatusCode >= 100 && httpStatusCode < 200) {
+      // 1xx. These headers should have been ignored.
+      return Status.INTERNAL;
+    }
+    switch (httpStatusCode) {
+      case HttpURLConnection.HTTP_BAD_REQUEST:  // 400
+        return Status.INVALID_ARGUMENT;
+      case HttpURLConnection.HTTP_UNAUTHORIZED:  // 401
+        return Status.UNAUTHENTICATED;
+      case HttpURLConnection.HTTP_FORBIDDEN:  // 403
+        return Status.PERMISSION_DENIED;
+      case HttpURLConnection.HTTP_NOT_FOUND:  // 404
+        return Status.NOT_FOUND;
+      case HttpURLConnection.HTTP_BAD_METHOD: // 405
+      case HttpURLConnection.HTTP_NOT_IMPLEMENTED: // 501
+        return Status.UNIMPLEMENTED;
+      case HttpURLConnection.HTTP_CONFLICT:  // 409
+        return Status.ABORTED;
+      case HttpURLConnection.HTTP_LENGTH_REQUIRED: // 411
+      case HttpURLConnection.HTTP_PRECON_FAILED: // 412
+        return Status.FAILED_PRECONDITION;
+      case 416: // Requested Range Not Satisfiable
+        return Status.OUT_OF_RANGE;
+      case HttpURLConnection.HTTP_INTERNAL_ERROR: //500
+        return Status.INTERNAL;
+      case HttpURLConnection.HTTP_MOVED_PERM:  // 301
+      case HttpURLConnection.HTTP_NOT_MODIFIED: //304
+      case 307: // Moved Temporarily
+      case HttpURLConnection.HTTP_BAD_GATEWAY:  // 502
+      case HttpURLConnection.HTTP_UNAVAILABLE:  // 503
+        return Status.UNAVAILABLE;
+      case HttpURLConnection.HTTP_GATEWAY_TIMEOUT:  // 504
+        return Status.DEADLINE_EXCEEDED;
+      default:
+        return Status.UNKNOWN;
+    }
+  }
+}

dora/underfs/cos/src/main/java/alluxio/underfs/cos/COSUnderFileSystem.java

@@ -46,6 +46,7 @@
 import com.qcloud.cos.model.SetObjectTaggingRequest;
 import com.qcloud.cos.model.Tag.Tag;
 import com.qcloud.cos.region.Region;
+import io.grpc.Status;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -177,12 +178,21 @@ public void setObjectTagging(String path, String name, String value) throws IOEx
 
   @Override
   public Map<String, String> getObjectTags(String path) throws IOException {
-    GetObjectTaggingRequest getTaggingReq = new GetObjectTaggingRequest(mBucketNameInternal, path);
-    GetObjectTaggingResult taggingResult = mClient.getObjectTagging(getTaggingReq);
-    List<Tag> tagList = taggingResult.getTagSet();
-    return Collections.unmodifiableMap(tagList.stream()
-        .collect(HashMap::new, (map, tag) -> map.put(tag.getKey(), tag.getValue()),
-            HashMap::putAll));
+    try {
+      GetObjectTaggingRequest getTaggingReq =
+          new GetObjectTaggingRequest(mBucketNameInternal, path);
+      GetObjectTaggingResult taggingResult = mClient.getObjectTagging(getTaggingReq);
+      List<Tag> tagList = taggingResult.getTagSet();
+      return Collections.unmodifiableMap(tagList.stream()
+          .collect(HashMap::new, (map, tag) -> map.put(tag.getKey(), tag.getValue()),
+              HashMap::putAll));
+    } catch (CosClientException e) {
+      AlluxioCosException exception = AlluxioCosException.from(e);
+      if (exception.getStatus().equals(Status.NOT_FOUND)) {
+        return null;
+      }
+      throw exception;
+    }
   }
 
   @Override
@@ -245,7 +255,8 @@ protected List<String> deleteObjects(List<String> keys) throws IOException {
           .map(DeleteObjectsResult.DeletedObject::getKey)
           .collect(Collectors.toList());
     } catch (CosClientException e) {
-      throw new IOException("failed to delete objects", e);
+      LOG.warn("failed to delete objects");
+      throw AlluxioCosException.from(e);
     }
   }
 
@@ -400,7 +411,7 @@ protected InputStream openObject(String key, OpenOptions options,
       return new COSInputStream(mBucketNameInternal, key, mClient, options.getOffset(), retryPolicy,
           mUfsConf.getBytes(PropertyKey.UNDERFS_OBJECT_STORE_MULTI_RANGE_CHUNK_SIZE));
     } catch (CosClientException e) {
-      throw new IOException(e.getMessage());
+      throw AlluxioCosException.from(e);
     }
   }
 }

dora/underfs/cos/src/test/java/alluxio/underfs/cos/COSUnderFileSystemTest.java

@@ -164,7 +164,7 @@ public void testDeleteObjects() throws IOException {
     try {
       mCOSUnderFileSystem.deleteObjects(keys);
     } catch (Exception e) {
-      Assert.assertTrue(e instanceof IOException);
+      Assert.assertTrue(e instanceof AlluxioCosException);
     }
   }
 

dora/underfs/oss/src/main/java/alluxio/underfs/oss/OSSUnderFileSystem.java

@@ -29,6 +29,7 @@
 import com.aliyun.oss.ClientBuilderConfiguration;
 import com.aliyun.oss.OSS;
 import com.aliyun.oss.OSSClientBuilder;
+import com.aliyun.oss.OSSException;
 import com.aliyun.oss.ServiceException;
 import com.aliyun.oss.common.comm.Protocol;
 import com.aliyun.oss.model.AbortMultipartUploadRequest;
@@ -73,6 +74,8 @@ public class OSSUnderFileSystem extends ObjectUnderFileSystem {
   /** Suffix for an empty file to flag it as a directory. */
   private static final String FOLDER_SUFFIX = "_$folder$";
 
+  private static final String NO_SUCH_KEY = "NoSuchKey";
+
   /** Aliyun OSS client. */
   private final OSS mClient;
 
@@ -209,8 +212,18 @@ public void setObjectTagging(String path, String name, String value) throws IOEx
 
   @Override
   public Map<String, String> getObjectTags(String path) throws IOException {
-    TagSet taggingResult = mClient.getObjectTagging(mBucketName, path);
-    return Collections.unmodifiableMap(taggingResult.getAllTags());
+    try {
+      TagSet taggingResult = mClient.getObjectTagging(mBucketName, path);
+      return Collections.unmodifiableMap(taggingResult.getAllTags());
+    } catch (ServiceException e) {
+      if (e instanceof OSSException) {
+        OSSException ossException = (OSSException) e;
+        if (NO_SUCH_KEY.equals(ossException.getErrorCode())) {
+          return null;
+        }
+      }
+      throw new IOException("Failed to get object tagging", e);
+    }
   }
 
   // No ACL integration currently, no-op