[Test](client) Add s3 storage client test sketelon (#57456)

### What problem does this PR solve?

If you want to run unit tests using a real cloud provider, you can
configure it in `custom_env.sh`.
```
# Enable S3 client for unit tests
export ENABLE_S3_CLIENT=1
# S3 client config
export S3_AK="xxx"
export S3_SK="xxx"
export S3_ENDPOINT="xxx"
export S3_PROVIDER="xxx"
export S3_BUCKET="xxx"
export S3_REGION="xxx"
export S3_PREFIX=""
```

Author: wyxxxcat

be/src/io/fs/azure_obj_storage_client.cpp

@@ -207,7 +207,7 @@ ObjectStorageUploadResponse AzureObjStorageClient::upload_part(const ObjectStora
         // The blockId must be base64 encoded
         s3_put_rate_limit([&]() {
             SCOPED_BVAR_LATENCY(s3_bvar::s3_multi_part_upload_latency);
-            client.StageBlock(base64_encode_part_num(part_num), memory_body);
+            return client.StageBlock(base64_encode_part_num(part_num), memory_body);
         });
     } catch (Azure::Core::RequestFailedException& e) {
         auto msg = fmt::format(

be/src/io/fs/s3_obj_storage_client.cpp

@@ -311,9 +311,6 @@ ObjectStorageHeadResponse S3ObjStorageClient::head_object(const ObjectStoragePat
                          static_cast<int>(outcome.GetError().GetResponseCode()),
                          outcome.GetError().GetRequestId()}};
     }
-    return ObjectStorageHeadResponse {
-            .resp = ObjectStorageResponse::OK(),
-    };
 }
 
 ObjectStorageResponse S3ObjStorageClient::get_object(const ObjectStoragePathOptions& opts,
@@ -334,6 +331,8 @@ ObjectStorageResponse S3ObjStorageClient::get_object(const ObjectStoragePathOpti
                 outcome.GetError().GetRequestId()};
     }
     *size_return = outcome.GetResult().GetContentLength();
+    // case for incomplete read
+    SYNC_POINT_CALLBACK("s3_obj_storage_client::get_object", size_return);
     if (*size_return != bytes_read) {
         return {convert_to_obj_response(Status::InternalError(
                 "failed to read from {}(bytes read: {}, bytes req: {}), request_id: {}",
@@ -405,6 +404,8 @@ ObjectStorageResponse S3ObjStorageClient::delete_objects(const ObjectStoragePath
                 static_cast<int>(delete_outcome.GetError().GetResponseCode()),
                 delete_outcome.GetError().GetRequestId()};
     }
+    // case for partial delete object failure
+    SYNC_POINT_CALLBACK("s3_obj_storage_client::delete_objects", &delete_outcome);
     if (!delete_outcome.GetResult().GetErrors().empty()) {
         const auto& e = delete_outcome.GetResult().GetErrors().front();
         return {convert_to_obj_response(
@@ -470,6 +471,9 @@ ObjectStorageResponse S3ObjStorageClient::delete_objects_recursively(
                         static_cast<int>(delete_outcome.GetError().GetResponseCode()),
                         delete_outcome.GetError().GetRequestId()};
             }
+            // case for partial delete object failure
+            SYNC_POINT_CALLBACK("s3_obj_storage_client::delete_objects_recursively",
+                                &delete_outcome);
             if (!delete_outcome.GetResult().GetErrors().empty()) {
                 const auto& e = delete_outcome.GetResult().GetErrors().front();
                 return {convert_to_obj_response(Status::InternalError(

be/src/util/s3_util.h

@@ -62,6 +62,7 @@ extern bvar::LatencyRecorder s3_copy_object_latency;
 }; // namespace s3_bvar
 
 std::string hide_access_key(const std::string& ak);
+int reset_s3_rate_limiter(S3RateLimitType type, size_t max_speed, size_t max_burst, size_t limit);
 
 class S3URI;
 struct S3ClientConf {