Clean tmp directory and file when initiate multipart upload

Alluxio#14477

Do some clean work when initiate multipart upload.

pr-link: Alluxio#14487
change-id: cid-0803aa47945440351e0ec66e40c61d253f2b39c6

core/server/proxy/src/main/java/alluxio/proxy/s3/S3RestServiceHandler.java

@@ -458,6 +458,9 @@ public Response createObjectOrUploadPart(@HeaderParam("Authorization") String au
       }
       AlluxioURI objectURI = new AlluxioURI(objectPath);
 
+      // remove exist object
+      deleteExistObject(fs, objectURI);
+
       CreateFilePOptions dirOptions =
           CreateFilePOptions.newBuilder().setRecursive(true).setWriteType(getS3WriteType()).build();
       // not copying from an existing file
@@ -477,10 +480,6 @@ public Response createObjectOrUploadPart(@HeaderParam("Authorization") String au
           } else {
             toRead = Integer.parseInt(contentLength);
           }
-          // overwrite existing object
-          if (fs.exists(objectURI)) {
-            fs.delete(objectURI, DeletePOptions.newBuilder().setUnchecked(true).build());
-          }
           FileOutStream os = fs.createFile(objectURI, dirOptions);
           try (DigestOutputStream digestOutputStream = new DigestOutputStream(os, md5)) {
             long read = ByteStreams.copy(ByteStreams.limit(readStream, toRead), digestOutputStream);
@@ -575,6 +574,10 @@ private Response initiateMultipartUpload(final FileSystem fs,
       AlluxioURI multipartTemporaryDir =
           new AlluxioURI(S3RestUtils.getMultipartTemporaryDirForObject(bucketPath, object));
 
+      // remove exist object
+      deleteExistObject(fs, new AlluxioURI(objectPath));
+      // remove exist multipartTemporaryDir
+      deleteExistObject(fs, multipartTemporaryDir, true);
       CreateDirectoryPOptions options = CreateDirectoryPOptions.newBuilder()
           .setRecursive(true).setWriteType(getS3WriteType()).build();
       try {
@@ -840,6 +843,40 @@ private void deleteObject(FileSystem fs, String bucket, String object) throws S3
     }
   }
 
+  /**
+   * Delete an existing key.
+   *
+   * @param fs instance of {@link FileSystem}
+   * @param objectURI the key uri
+   */
+  private void deleteExistObject(final FileSystem fs, AlluxioURI objectURI)
+      throws S3Exception {
+    deleteExistObject(fs, objectURI, false);
+  }
+
+  /**
+   * Delete an existing key.
+   *
+   * @param fs instance of {@link FileSystem}
+   * @param objectURI the key uri
+   * @param recursive if delete option is recursive
+   */
+  private void deleteExistObject(final FileSystem fs, AlluxioURI objectURI, Boolean recursive)
+      throws S3Exception {
+    try {
+      if (fs.exists(objectURI)) {
+        if (recursive) {
+          fs.delete(objectURI, DeletePOptions.newBuilder().setRecursive(true).build());
+        } else {
+          fs.delete(objectURI);
+        }
+        LOG.info("Remove exist object: {} for overwrite.", objectURI.getPath());
+      }
+    } catch (IOException | AlluxioException e) {
+      throw toObjectS3Exception(e, objectURI.getPath());
+    }
+  }
+
   private S3Exception toBucketS3Exception(Exception exception, String resource) {
     try {
       throw exception;

docs/en/api/S3-API.md

@@ -23,6 +23,13 @@ Alluxio proxy, introducing an extra hop. For optimal performance, it is recommen
 server and an Alluxio worker on each compute node. It is also recommended to put all the proxy
 servers behind a load balancer.
 
+As described in the Aws s3 document [Aws PutObject](https://docs.aws.amazon.com/AmazonS3/latest/API/API_PutObject.html):  
+> _Amazon S3 is a distributed system. If it receives multiple write requests for the same object simultaneously, it overwrites all but the last object written._  
+> _Amazon S3 does not provide object locking; if you need this, make sure to build it into your application layer or use versioning instead._  
+
+Alluxio s3 will overwrite the existing key and the temporary directory for multipart upload. 
+
+
 ## Features support
 The following table describes the support status for current Amazon S3 functional features: