From 2d239aac81d9862b2777585dd9e152908da838f2 Mon Sep 17 00:00:00 2001
From: vintmd <61688729+vintmd@users.noreply.github.com>
Date: Tue, 31 May 2022 12:51:50 +0800
Subject: [PATCH] add complete mpu 409 head retry (#60)
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Co-authored-by: alantong(佟明达) <alantong@tencent.com>
---
 .../hadoop/fs/CosNFSDataOutputStream.java     |  1 +
 .../hadoop/fs/CosNativeFileSystemStore.java   | 25 ++++++++++++++++---
 2 files changed, 23 insertions(+), 3 deletions(-)

diff --git a/src/main/java/org/apache/hadoop/fs/CosNFSDataOutputStream.java b/src/main/java/org/apache/hadoop/fs/CosNFSDataOutputStream.java
index 4dde832c..fb39c0a8 100644
--- a/src/main/java/org/apache/hadoop/fs/CosNFSDataOutputStream.java
+++ b/src/main/java/org/apache/hadoop/fs/CosNFSDataOutputStream.java
@@ -542,6 +542,7 @@ protected void complete() throws IOException {
                 throw new IOException("failed to multipart upload to cos, abort it.");
             }
 
+            // notice sometimes complete result may be null
             CompleteMultipartUploadResult completeResult =
                     nativeStore.completeMultipartUpload(cosKey, this.uploadId, new LinkedList<>(futurePartETagList));
             this.completed = true;
diff --git a/src/main/java/org/apache/hadoop/fs/CosNativeFileSystemStore.java b/src/main/java/org/apache/hadoop/fs/CosNativeFileSystemStore.java
index 838bf949..a3826d3d 100644
--- a/src/main/java/org/apache/hadoop/fs/CosNativeFileSystemStore.java
+++ b/src/main/java/org/apache/hadoop/fs/CosNativeFileSystemStore.java
@@ -544,7 +544,7 @@ public String getUploadId(String key) throws IOException {
     }
 
     /**
-     * complete cos mpu
+     * complete cos mpu, sometimes return null complete mpu result
      * @param key cos key
      * @param uploadId upload id
      * @param partETagList each part etag list
@@ -569,9 +569,28 @@ public int compare(PartETag o1, PartETag o2) {
                             partETagList);
             completeMultipartUploadRequest.setObjectMetadata(objectMetadata);
             return (CompleteMultipartUploadResult) this.callCOSClientWithRetry(completeMultipartUploadRequest);
+        } catch (CosServiceException cse) {
+            // when first calling with 503 access time out, next retry will 409.
+            int statusCode = cse.getStatusCode();
+            if (statusCode == 409) {
+                // check file whether exist
+                FileMetadata fileMetadata = this.queryObjectMetadata(key);
+                if (null == fileMetadata) {
+                    // if file not exist through exception
+                    handleException(cse, key);
+                }
+                LOG.warn("Upload the cos key [{}] complete mpu concurrently", key);
+            } else {
+                // other exception
+                String errMsg = String.format("Complete the multipart upload failed. " +
+                        "cos service exception, cos key: %s, upload id: %s, " +
+                        "exception: %s", key, uploadId, cse.toString());
+                handleException(new Exception(errMsg), key);
+            }
         } catch (Exception e) {
-            String errMsg = String.format("Complete the multipart upload failed. cos key: %s, upload id: %s, " +
-                    "exception: %s", key, uploadId, e);
+            String errMsg = String.format("Complete the multipart upload failed. " +
+                    "cos key: %s, upload id: %s, " +
+                    "exception: %s", key, uploadId, e.toString());
             handleException(new Exception(errMsg), key);
         }
         return null;