fix(s3): parse MultipartUploadResponse to check error in body (#4735)

core/src/services/s3/error.rs

@@ -16,6 +16,7 @@
 // under the License.
 
 use bytes::Buf;
+use http::response::Parts;
 use http::Response;
 use quick_xml::de;
 use serde::Deserialize;
@@ -24,13 +25,13 @@ use crate::raw::*;
 use crate::*;
 
 /// S3Error is the error returned by s3 service.
-#[derive(Default, Debug, Deserialize)]
+#[derive(Default, Debug, Deserialize, PartialEq, Eq)]
 #[serde(default, rename_all = "PascalCase")]
-struct S3Error {
-    code: String,
-    message: String,
-    resource: String,
-    request_id: String,
+pub(crate) struct S3Error {
+    pub code: String,
+    pub message: String,
+    pub resource: String,
+    pub request_id: String,
 }
 
 /// Parse error response into Error.
@@ -68,6 +69,21 @@ pub async fn parse_error(resp: Response<Buffer>) -> Result<Error> {
     Ok(err)
 }
 
+/// Util function to build [`Error`] from a [`S3Error`] object.
+pub(crate) fn from_s3_error(s3_error: S3Error, parts: Parts) -> Error {
+    let (kind, retryable) =
+        parse_s3_error_code(s3_error.code.as_str()).unwrap_or((ErrorKind::Unexpected, false));
+    let mut err = Error::new(kind, &format!("{s3_error:?}"));
+
+    err = with_error_response_context(err, parts);
+
+    if retryable {
+        err = err.set_temporary();
+    }
+
+    err
+}
+
 /// Returns the `Error kind` of this code and whether the error is retryable.
 /// All possible error code: <https://docs.aws.amazon.com/AmazonS3/latest/API/ErrorResponses.html#ErrorCodeList>
 pub fn parse_s3_error_code(code: &str) -> Option<(ErrorKind, bool)> {
@@ -123,4 +139,22 @@ mod tests {
         assert_eq!(out.resource, "/mybucket/myfoto.jpg");
         assert_eq!(out.request_id, "4442587FB7D0A2F9");
     }
+
+    #[test]
+    fn test_parse_error_from_unrelated_input() {
+        let bs = bytes::Bytes::from(
+            r#"
+<?xml version="1.0" encoding="UTF-8"?>
+<CompleteMultipartUploadResult xmlns="http://s3.amazonaws.com/doc/2006-03-01/">
+  <Location>http://Example-Bucket.s3.ap-southeast-1.amazonaws.com/Example-Object</Location>
+  <Bucket>Example-Bucket</Bucket>
+  <Key>Example-Object</Key> 
+  <ETag>"3858f62230ac3c915f300c664312c11f-9"</ETag>
+</CompleteMultipartUploadResult>
+"#,
+        );
+
+        let out: S3Error = de::from_reader(bs.reader()).expect("must success");
+        assert_eq!(out, S3Error::default());
+    }
 }

core/src/services/s3/writer.rs

@@ -21,7 +21,7 @@ use bytes::Buf;
 use http::StatusCode;
 
 use super::core::*;
-use super::error::parse_error;
+use super::error::{from_s3_error, parse_error, S3Error};
 use crate::raw::*;
 use crate::*;
 
@@ -158,7 +158,18 @@ impl oio::MultipartWrite for S3Writer {
         let status = resp.status();
 
         match status {
-            StatusCode::OK => Ok(()),
+            StatusCode::OK => {
+                // still check if there is any error because S3 might return error for status code 200
+                // https://docs.aws.amazon.com/AmazonS3/latest/API/API_CompleteMultipartUpload.html#API_CompleteMultipartUpload_Example_4
+                let (parts, body) = resp.into_parts();
+                let maybe_error: S3Error =
+                    quick_xml::de::from_reader(body.reader()).map_err(new_xml_deserialize_error)?;
+                if !maybe_error.code.is_empty() {
+                    return Err(from_s3_error(maybe_error, parts));
+                }
+
+                Ok(())
+            }
             _ => Err(parse_error(resp).await?),
         }
     }