add complete mpu 409 head retry (#60)

Co-authored-by: alantong(佟明达) <[email protected]>
tencentyun · May 31, 2022 · 2d239aa · 2d239aa
1 parent da38a2f
commit 2d239aa
Show file tree

Hide file tree

Showing 2 changed files with 23 additions and 3 deletions.
diff --git 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
@@ -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;