feat(services/gcs): Add abort support for writer (#2242)

* feat(services/gcs): writer abort

Signed-off-by: suyanhanx <suyanhanx@gmail.com>

* fix, success resp status code must be 499

Signed-off-by: suyanhanx <suyanhanx@gmail.com>

* remove unwrap

Signed-off-by: suyanhanx <suyanhanx@gmail.com>

* simplify

Signed-off-by: suyanhanx <suyanhanx@gmail.com>

---------

Signed-off-by: suyanhanx <suyanhanx@gmail.com>

core/src/services/gcs/core.rs

@@ -538,4 +538,18 @@ impl GcsCore {
 
         self.send(req).await
     }
+
+    pub async fn gcs_abort_resumable_upload(
+        &self,
+        location: &str,
+    ) -> Result<Response<IncomingAsyncBody>> {
+        let mut req = Request::delete(location)
+            .header(CONTENT_LENGTH, 0)
+            .body(AsyncBody::Empty)
+            .map_err(new_request_build_error)?;
+
+        self.sign(&mut req).await?;
+
+        self.send(req).await
+    }
 }

core/src/services/gcs/writer.rs

@@ -163,9 +163,26 @@ impl oio::Write for GcsWriter {
         }
     }
 
-    // TODO: we can cancel the upload by sending a DELETE request to the location
     async fn abort(&mut self) -> Result<()> {
-        Ok(())
+        let location = if let Some(location) = &self.location {
+            location
+        } else {
+            return Ok(());
+        };
+
+        let resp = self.core.gcs_abort_resumable_upload(location).await?;
+
+        match resp.status().as_u16() {
+            // gcs returns 499 if the upload aborted successfully
+            // reference: https://cloud.google.com/storage/docs/performing-resumable-uploads#cancel-upload-json
+            499 => {
+                resp.into_body().consume().await?;
+                self.location = None;
+                self.buffer.clear();
+                Ok(())
+            }
+            _ => Err(parse_error(resp).await?),
+        }
     }
 
     async fn close(&mut self) -> Result<()> {