diff --git a/be/src/common/config.cpp b/be/src/common/config.cpp
index dfa7783075fcb8..e19bcdb220ab69 100644
--- a/be/src/common/config.cpp
+++ b/be/src/common/config.cpp
@@ -1523,6 +1523,9 @@ DEFINE_mInt64(s3_put_token_per_second, "1000000000000000000");
 DEFINE_Validator(s3_put_token_per_second, [](int64_t config) -> bool { return config > 0; });
 
 DEFINE_mInt64(s3_put_token_limit, "0");
+// Log active S3 rate limiter every N throttled/rejected requests, 0 means no log.
+DEFINE_mInt64(s3_rate_limiter_log_interval, "1000");
+DEFINE_Validator(s3_rate_limiter_log_interval, [](int64_t config) -> bool { return config >= 0; });
 
 DEFINE_String(trino_connector_plugin_dir, "${DORIS_HOME}/plugins/connectors");
 
diff --git a/be/src/common/config.h b/be/src/common/config.h
index cb174ff2fa47a2..a003f3acdc44da 100644
--- a/be/src/common/config.h
+++ b/be/src/common/config.h
@@ -1604,6 +1604,7 @@ DECLARE_mInt64(s3_get_token_limit);
 DECLARE_mInt64(s3_put_bucket_tokens);
 DECLARE_mInt64(s3_put_token_per_second);
 DECLARE_mInt64(s3_put_token_limit);
+DECLARE_mInt64(s3_rate_limiter_log_interval);
 // max s3 client retry times
 DECLARE_mInt32(max_s3_client_retry);
 // When meet s3 429 error, the "get" request will
diff --git a/be/src/io/fs/azure_obj_storage_client.cpp b/be/src/io/fs/azure_obj_storage_client.cpp
index 85b81c1deb23df..4ac9117fad9ae3 100644
--- a/be/src/io/fs/azure_obj_storage_client.cpp
+++ b/be/src/io/fs/azure_obj_storage_client.cpp
@@ -43,6 +43,7 @@
 #include "common/exception.h"
 #include "common/logging.h"
 #include "common/status.h"
+#include "cpp/obj_retry_strategy.h"
 #include "io/fs/obj_storage_client.h"
 #include "util/bvar_helper.h"
 #include "util/coding.h"
@@ -74,7 +75,7 @@ auto s3_rate_limit(doris::S3RateLimitType op, Func callback) -> decltype(callbac
     if (!doris::config::enable_s3_rate_limiter) {
         return callback();
     }
-    auto sleep_duration = doris::S3ClientFactory::instance().rate_limiter(op)->add(1);
+    auto sleep_duration = doris::apply_s3_rate_limit(op);
     if (sleep_duration < 0) {
         throw std::runtime_error("Azure exceeds request limit");
     }
@@ -124,6 +125,7 @@ ObjectStorageResponse do_azure_client_call(Func f, const ObjectStoragePathOption
     try {
         f();
     } catch (Azure::Core::RequestFailedException& e) {
+        doris::record_object_request_failed(static_cast<int>(e.StatusCode));
         auto tls_debug_suffix = build_azure_tls_debug_suffix(
                 fmt::format("{} {}", e.what(), e.Message), tls_debug_context);
         auto msg = fmt::format(
@@ -186,6 +188,7 @@ struct AzureBatchDeleter {
                     0 == strcmp(e.ErrorCode.c_str(), BlobNotFound)) {
                     continue;
                 }
+                doris::record_object_request_failed(static_cast<int>(e.StatusCode));
                 auto msg = fmt::format(
                         "Azure request failed because {}, error msg {}, http code {}, path msg "
                         "{}{}",
diff --git a/be/src/io/fs/s3_obj_storage_client.cpp b/be/src/io/fs/s3_obj_storage_client.cpp
index 1b8bf7b473c076..f9ed8e155ff59c 100644
--- a/be/src/io/fs/s3_obj_storage_client.cpp
+++ b/be/src/io/fs/s3_obj_storage_client.cpp
@@ -66,6 +66,7 @@
 
 #include "common/logging.h"
 #include "common/status.h"
+#include "cpp/obj_retry_strategy.h"
 #include "cpp/sync_point.h"
 #include "io/fs/err_utils.h"
 #include "io/fs/s3_common.h"
@@ -82,7 +83,7 @@ auto s3_rate_limit(doris::S3RateLimitType op, Func callback) -> decltype(callbac
     if (!doris::config::enable_s3_rate_limiter) {
         return callback();
     }
-    auto sleep_duration = doris::S3ClientFactory::instance().rate_limiter(op)->add(1);
+    auto sleep_duration = doris::apply_s3_rate_limit(op);
     if (sleep_duration < 0) {
         return T(s3_error_factory());
     }
@@ -98,6 +99,10 @@ template <typename Func>
 auto s3_put_rate_limit(Func callback) -> decltype(callback()) {
     return s3_rate_limit(doris::S3RateLimitType::PUT, std::move(callback));
 }
+
+void record_s3_request_failed(const Aws::S3::S3Error& error) {
+    doris::record_object_request_failed(static_cast<int>(error.GetResponseCode()));
+}
 } // namespace
 
 namespace Aws::S3::Model {
@@ -140,6 +145,7 @@ ObjectStorageUploadResponse S3ObjStorageClient::create_multipart_upload(
             << ", request_id=" << request_id << ", bucket=" << opts.bucket << ", key=" << opts.key;
 
     if (!outcome.IsSuccess()) {
+        record_s3_request_failed(outcome.GetError());
         auto st = s3fs_error(outcome.GetError(), fmt::format("failed to CreateMultipartUpload: {} ",
                                                              opts.path.native()));
         LOG(WARNING) << st << " request_id=" << request_id;
@@ -177,6 +183,7 @@ ObjectStorageResponse S3ObjStorageClient::put_object(const ObjectStoragePathOpti
                                                  : outcome.GetError().GetRequestId();
 
     if (!outcome.IsSuccess()) {
+        record_s3_request_failed(outcome.GetError());
         auto st = s3fs_error(outcome.GetError(),
                              fmt::format("failed to put object: {}", opts.path.native()));
         LOG(WARNING) << st << ", request_id=" << request_id;
@@ -222,6 +229,7 @@ ObjectStorageUploadResponse S3ObjStorageClient::upload_part(const ObjectStorageP
 
     TEST_SYNC_POINT_CALLBACK("S3FileWriter::_upload_one_part", &outcome);
     if (!outcome.IsSuccess()) {
+        record_s3_request_failed(outcome.GetError());
         auto st = Status::IOError(
                 "failed to UploadPart bucket={}, key={}, part_num={}, upload_id={}, message={}, "
                 "exception_name={}, response_code={}, request_id={}",
@@ -275,6 +283,7 @@ ObjectStorageResponse S3ObjStorageClient::complete_multipart_upload(
                                                  : outcome.GetError().GetRequestId();
 
     if (!outcome.IsSuccess()) {
+        record_s3_request_failed(outcome.GetError());
         auto st = s3fs_error(outcome.GetError(),
                              fmt::format("failed to CompleteMultipartUpload: {}, upload_id={}",
                                          opts.path.native(), *opts.upload_id));
@@ -305,6 +314,7 @@ ObjectStorageHeadResponse S3ObjStorageClient::head_object(const ObjectStoragePat
     } else if (outcome.GetError().GetResponseCode() == Aws::Http::HttpResponseCode::NOT_FOUND) {
         return {.resp = {convert_to_obj_response(Status::Error<ErrorCode::NOT_FOUND, false>(""))}};
     } else {
+        record_s3_request_failed(outcome.GetError());
         return {.resp = {convert_to_obj_response(
                                  s3fs_error(outcome.GetError(),
                                             fmt::format("failed to check exists {}", opts.key))),
@@ -324,6 +334,7 @@ ObjectStorageResponse S3ObjStorageClient::get_object(const ObjectStoragePathOpti
     SCOPED_BVAR_LATENCY(s3_bvar::s3_get_latency);
     auto outcome = s3_get_rate_limit([&]() { return _client->GetObject(request); });
     if (!outcome.IsSuccess()) {
+        record_s3_request_failed(outcome.GetError());
         return {convert_to_obj_response(s3fs_error(
                         outcome.GetError(), fmt::format("failed to read from {}", opts.key))),
                 static_cast<int>(outcome.GetError().GetResponseCode()),
@@ -362,6 +373,7 @@ ObjectStorageResponse S3ObjStorageClient::list_objects(const ObjectStoragePathOp
                 return ObjectStorageResponse::OK();
             }
 
+            record_s3_request_failed(outcome.GetError());
             return {convert_to_obj_response(s3fs_error(
                             outcome.GetError(), fmt::format("failed to list {}", opts.prefix))),
                     static_cast<int>(outcome.GetError().GetResponseCode()),
@@ -406,6 +418,7 @@ ObjectStorageResponse S3ObjStorageClient::delete_objects(const ObjectStoragePath
     auto delete_outcome =
             s3_put_rate_limit([&]() { return _client->DeleteObjects(delete_request); });
     if (!delete_outcome.IsSuccess()) {
+        record_s3_request_failed(delete_outcome.GetError());
         return {convert_to_obj_response(
                         s3fs_error(delete_outcome.GetError(),
                                    fmt::format("failed to delete dir {}", opts.key))),
@@ -433,6 +446,7 @@ ObjectStorageResponse S3ObjStorageClient::delete_object(const ObjectStoragePathO
         outcome.GetError().GetResponseCode() == Aws::Http::HttpResponseCode::NOT_FOUND) {
         return ObjectStorageResponse::OK();
     }
+    record_s3_request_failed(outcome.GetError());
     return {convert_to_obj_response(s3fs_error(outcome.GetError(),
                                                fmt::format("failed to delete file {}", opts.key))),
             static_cast<int>(outcome.GetError().GetResponseCode()),
@@ -453,6 +467,7 @@ ObjectStorageResponse S3ObjStorageClient::delete_objects_recursively(
             outcome = s3_get_rate_limit([&]() { return _client->ListObjectsV2(request); });
         }
         if (!outcome.IsSuccess()) {
+            record_s3_request_failed(outcome.GetError());
             return {convert_to_obj_response(s3fs_error(
                             outcome.GetError(),
                             fmt::format("failed to list objects when delete dir {}", opts.prefix))),
@@ -473,6 +488,7 @@ ObjectStorageResponse S3ObjStorageClient::delete_objects_recursively(
             auto delete_outcome =
                     s3_put_rate_limit([&]() { return _client->DeleteObjects(delete_request); });
             if (!delete_outcome.IsSuccess()) {
+                record_s3_request_failed(delete_outcome.GetError());
                 return {convert_to_obj_response(
                                 s3fs_error(delete_outcome.GetError(),
                                            fmt::format("failed to delete dir {}", opts.key))),
@@ -502,4 +518,4 @@ std::string S3ObjStorageClient::generate_presigned_url(const ObjectStoragePathOp
                                          expiration_secs);
 }
 
-} // namespace doris::io
\ No newline at end of file
+} // namespace doris::io
diff --git a/be/src/util/s3_util.cpp b/be/src/util/s3_util.cpp
index fc4c81d4bd96f2..a9b571a2d17929 100644
--- a/be/src/util/s3_util.cpp
+++ b/be/src/util/s3_util.cpp
@@ -158,11 +158,6 @@ constexpr char S3_EXTERNAL_ID[] = "AWS_EXTERNAL_ID";
 constexpr char S3_CREDENTIALS_PROVIDER_TYPE[] = "AWS_CREDENTIALS_PROVIDER_TYPE";
 } // namespace
 
-bvar::Adder<int64_t> get_rate_limit_ns("get_rate_limit_ns");
-bvar::Adder<int64_t> get_rate_limit_exceed_req_num("get_rate_limit_exceed_req_num");
-bvar::Adder<int64_t> put_rate_limit_ns("put_rate_limit_ns");
-bvar::Adder<int64_t> put_rate_limit_exceed_req_num("put_rate_limit_exceed_req_num");
-
 static std::atomic<int64_t> last_s3_get_token_bucket_tokens {0};
 static std::atomic<int64_t> last_s3_get_token_limit {0};
 static std::atomic<int64_t> last_s3_get_token_per_second {0};
@@ -230,6 +225,11 @@ int reset_s3_rate_limiter(S3RateLimitType type, size_t max_speed, size_t max_bur
     return S3ClientFactory::instance().rate_limiter(type)->reset(max_speed, max_burst, limit);
 }
 
+int64_t apply_s3_rate_limit(S3RateLimitType type) {
+    return doris::apply_s3_rate_limit(type, S3ClientFactory::instance().rate_limiter(type),
+                                      config::s3_rate_limiter_log_interval);
+}
+
 S3ClientFactory::S3ClientFactory() {
     _aws_options = Aws::SDKOptions {};
     auto logLevel = static_cast<Aws::Utils::Logging::LogLevel>(config::aws_log_level);
@@ -242,12 +242,10 @@ S3ClientFactory::S3ClientFactory() {
     _rate_limiters = {
             std::make_unique<S3RateLimiterHolder>(
                     config::s3_get_token_per_second, config::s3_get_bucket_tokens,
-                    config::s3_get_token_limit,
-                    metric_func_factory(get_rate_limit_ns, get_rate_limit_exceed_req_num)),
+                    config::s3_get_token_limit, s3_rate_limiter_metric_func(S3RateLimitType::GET)),
             std::make_unique<S3RateLimiterHolder>(
                     config::s3_put_token_per_second, config::s3_put_bucket_tokens,
-                    config::s3_put_token_limit,
-                    metric_func_factory(put_rate_limit_ns, put_rate_limit_exceed_req_num))};
+                    config::s3_put_token_limit, s3_rate_limiter_metric_func(S3RateLimitType::PUT))};
 
 #ifdef USE_AZURE
     auto azureLogLevel =
diff --git a/be/src/util/s3_util.h b/be/src/util/s3_util.h
index 5546db857c91b1..064f88accd8539 100644
--- a/be/src/util/s3_util.h
+++ b/be/src/util/s3_util.h
@@ -64,6 +64,7 @@ extern bvar::LatencyRecorder s3_copy_object_latency;
 
 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);
+int64_t apply_s3_rate_limit(S3RateLimitType type);
 // Rebuild the S3 GET/PUT rate limiters if the related configs have changed.
 // Safe to call periodically; it is a no-op when nothing changed.
 void check_s3_rate_limiter_config_changed();
diff --git a/cloud/src/common/config.h b/cloud/src/common/config.h
index 29e795ca078423..ad4814a13eaba5 100644
--- a/cloud/src/common/config.h
+++ b/cloud/src/common/config.h
@@ -282,6 +282,9 @@ CONF_mBool(enable_s3_rate_limiter, "false");
 // s3_rate_limit_inject_probility is the probability (0-100) of injecting a rate limit error.
 CONF_mBool(enable_s3_rate_limit_inject, "false");
 CONF_mInt32(s3_rate_limit_inject_probility, "30");
+// Log active S3 rate limiter every N throttled/rejected requests, 0 means no log.
+CONF_mInt64(s3_rate_limiter_log_interval, "1000");
+CONF_Validator(s3_rate_limiter_log_interval, [](int64_t config) -> bool { return config >= 0; });
 CONF_mInt64(s3_get_bucket_tokens, "1000000000000000000");
 CONF_Validator(s3_get_bucket_tokens, [](int64_t config) -> bool { return config > 0; });
 
diff --git a/cloud/src/recycler/azure_obj_client.cpp b/cloud/src/recycler/azure_obj_client.cpp
index 94901cde48b6c1..5390c1c5f99608 100644
--- a/cloud/src/recycler/azure_obj_client.cpp
+++ b/cloud/src/recycler/azure_obj_client.cpp
@@ -35,6 +35,7 @@
 #include "common/config.h"
 #include "common/logging.h"
 #include "common/stopwatch.h"
+#include "cpp/obj_retry_strategy.h"
 #include "cpp/sync_point.h"
 #include "cpp/token_bucket_rate_limiter.h"
 #include "recycler/s3_accessor.h"
@@ -56,7 +57,9 @@ auto s3_rate_limit(S3RateLimitType op, Func callback) -> decltype(callback()) {
     if (!config::enable_s3_rate_limiter) {
         return callback();
     }
-    auto sleep_duration = AccessorRateLimiter::instance().rate_limiter(op)->add(1);
+    auto sleep_duration =
+            doris::apply_s3_rate_limit(op, AccessorRateLimiter::instance().rate_limiter(op),
+                                       config::s3_rate_limiter_log_interval);
     if (sleep_duration < 0) {
         throw std::runtime_error("Azure exceeds request limit");
     }
@@ -81,6 +84,7 @@ ObjectStorageResponse do_azure_client_call(Func f, std::string_view url, std::st
     try {
         f();
     } catch (Azure::Core::RequestFailedException& e) {
+        doris::record_object_request_failed(static_cast<int>(e.StatusCode));
         auto msg = fmt::format(
                 "Azure request failed because {}, http_code: {}, request_id: {}, url: {}, "
                 "key: {}",
@@ -280,6 +284,7 @@ ObjectStorageResponse AzureObjClient::delete_objects(const std::string& bucket,
                     0 == strcmp(e.ErrorCode.c_str(), BlobNotFound)) {
                     continue;
                 }
+                doris::record_object_request_failed(static_cast<int>(e.StatusCode));
                 auto msg = fmt::format(
                         "Azure request failed because {}, http code {}, request id {}, url {}",
                         e.Message, static_cast<int>(e.StatusCode), e.RequestId, client_->GetUrl());
@@ -333,4 +338,4 @@ ObjectStorageResponse AzureObjClient::abort_multipart_upload(ObjectStoragePathRe
     return delete_object(path);
 }
 
-} // namespace doris::cloud
\ No newline at end of file
+} // namespace doris::cloud
diff --git a/cloud/src/recycler/s3_accessor.cpp b/cloud/src/recycler/s3_accessor.cpp
index 51cc9023904aa5..bbd1b7500caa49 100644
--- a/cloud/src/recycler/s3_accessor.cpp
+++ b/cloud/src/recycler/s3_accessor.cpp
@@ -73,22 +73,15 @@ bvar::LatencyRecorder s3_get_bucket_version_latency("s3_get_bucket_version");
 bvar::LatencyRecorder s3_copy_object_latency("s3_copy_object");
 }; // namespace s3_bvar
 
-bvar::Adder<int64_t> get_rate_limit_ns("get_rate_limit_ns");
-bvar::Adder<int64_t> get_rate_limit_exceed_req_num("get_rate_limit_exceed_req_num");
-bvar::Adder<int64_t> put_rate_limit_ns("put_rate_limit_ns");
-bvar::Adder<int64_t> put_rate_limit_exceed_req_num("put_rate_limit_exceed_req_num");
-
 AccessorRateLimiter::AccessorRateLimiter()
-        : _rate_limiters(
-                  {std::make_unique<S3RateLimiterHolder>(
-                           config::s3_get_token_per_second, config::s3_get_bucket_tokens,
-                           config::s3_get_token_limit,
-                           metric_func_factory(get_rate_limit_ns, get_rate_limit_exceed_req_num)),
-                   std::make_unique<S3RateLimiterHolder>(
-                           config::s3_put_token_per_second, config::s3_put_bucket_tokens,
-                           config::s3_put_token_limit,
-                           metric_func_factory(put_rate_limit_ns,
-                                               put_rate_limit_exceed_req_num))}) {}
+        : _rate_limiters({std::make_unique<S3RateLimiterHolder>(
+                                  config::s3_get_token_per_second, config::s3_get_bucket_tokens,
+                                  config::s3_get_token_limit,
+                                  s3_rate_limiter_metric_func(S3RateLimitType::GET)),
+                          std::make_unique<S3RateLimiterHolder>(
+                                  config::s3_put_token_per_second, config::s3_put_bucket_tokens,
+                                  config::s3_put_token_limit,
+                                  s3_rate_limiter_metric_func(S3RateLimitType::PUT))}) {}
 
 S3RateLimiterHolder* AccessorRateLimiter::rate_limiter(S3RateLimitType type) {
     CHECK(type == S3RateLimitType::GET || type == S3RateLimitType::PUT) << to_string(type);
diff --git a/cloud/src/recycler/s3_obj_client.cpp b/cloud/src/recycler/s3_obj_client.cpp
index c45a6aae168c0c..3f299d0fef03e5 100644
--- a/cloud/src/recycler/s3_obj_client.cpp
+++ b/cloud/src/recycler/s3_obj_client.cpp
@@ -32,6 +32,7 @@
 #include "common/config.h"
 #include "common/logging.h"
 #include "common/stopwatch.h"
+#include "cpp/obj_retry_strategy.h"
 #include "cpp/sync_point.h"
 #include "cpp/token_bucket_rate_limiter.h"
 #include "recycler/s3_accessor.h"
@@ -43,6 +44,10 @@ namespace doris::cloud {
     return {Aws::S3::S3Errors::INTERNAL_FAILURE, "exceeds limit", "exceeds limit", false};
 }
 
+void record_s3_request_failed(const Aws::S3::S3Error& error) {
+    doris::record_object_request_failed(static_cast<int>(error.GetResponseCode()));
+}
+
 template <typename Func>
 auto s3_rate_limit(S3RateLimitType op, Func callback) -> decltype(callback()) {
     using T = decltype(callback());
@@ -55,7 +60,9 @@ auto s3_rate_limit(S3RateLimitType op, Func callback) -> decltype(callback()) {
     if (!config::enable_s3_rate_limiter) {
         return callback();
     }
-    auto sleep_duration = AccessorRateLimiter::instance().rate_limiter(op)->add(1);
+    auto sleep_duration =
+            doris::apply_s3_rate_limit(op, AccessorRateLimiter::instance().rate_limiter(op),
+                                       config::s3_rate_limiter_log_interval);
     if (sleep_duration < 0) {
         return T(s3_error_factory());
     }
@@ -118,6 +125,7 @@ class S3ObjListIterator final : public ObjectListIterator {
                 return false;
             }
 
+            record_s3_request_failed(outcome.GetError());
             LOG_WARNING("failed to list objects")
                     .tag("endpoint", endpoint_)
                     .tag("bucket", req_.GetBucket())
@@ -210,6 +218,7 @@ ObjectStorageResponse S3ObjClient::put_object(ObjectStoragePathRef path, std::st
         return s3_client_->PutObject(request);
     });
     if (!outcome.IsSuccess()) {
+        record_s3_request_failed(outcome.GetError());
         LOG_WARNING("failed to put object")
                 .tag("endpoint", endpoint_)
                 .tag("bucket", path.bucket)
@@ -237,6 +246,7 @@ ObjectStorageResponse S3ObjClient::head_object(ObjectStoragePathRef path, Object
     } else if (outcome.GetError().GetResponseCode() == Aws::Http::HttpResponseCode::NOT_FOUND) {
         return 1;
     } else {
+        record_s3_request_failed(outcome.GetError());
         LOG_WARNING("failed to head object")
                 .tag("endpoint", endpoint_)
                 .tag("bucket", path.bucket)
@@ -276,6 +286,7 @@ ObjectStorageResponse S3ObjClient::delete_objects(const std::string& bucket,
             return s3_client_->DeleteObjects(delete_request);
         });
         if (!delete_outcome.IsSuccess()) {
+            record_s3_request_failed(delete_outcome.GetError());
             LOG_WARNING("failed to delete objects")
                     .tag("endpoint", endpoint_)
                     .tag("bucket", bucket)
@@ -326,6 +337,10 @@ ObjectStorageResponse S3ObjClient::delete_object(ObjectStoragePathRef path) {
     });
     TEST_SYNC_POINT_CALLBACK("S3ObjClient::delete_object", &outcome);
     if (!outcome.IsSuccess()) {
+        if (outcome.GetError().GetResponseCode() == Aws::Http::HttpResponseCode::NOT_FOUND) {
+            return {ObjectStorageResponse::NOT_FOUND, outcome.GetError().GetMessage()};
+        }
+        record_s3_request_failed(outcome.GetError());
         LOG_WARNING("failed to delete object")
                 .tag("endpoint", endpoint_)
                 .tag("bucket", path.bucket)
@@ -334,9 +349,6 @@ ObjectStorageResponse S3ObjClient::delete_object(ObjectStoragePathRef path) {
                 .tag("error", outcome.GetError().GetMessage())
                 .tag("exception", outcome.GetError().GetExceptionName())
                 .tag("request_id", outcome.GetError().GetRequestId());
-        if (outcome.GetError().GetResponseCode() == Aws::Http::HttpResponseCode::NOT_FOUND) {
-            return {ObjectStorageResponse::NOT_FOUND, outcome.GetError().GetMessage()};
-        }
         return {ObjectStorageResponse::UNDEFINED, outcome.GetError().GetMessage()};
     }
     return {ObjectStorageResponse::OK};
@@ -365,6 +377,7 @@ ObjectStorageResponse S3ObjClient::get_life_cycle(const std::string& bucket,
             }
         }
     } else {
+        record_s3_request_failed(outcome.GetError());
         LOG_WARNING("Err for check interval: failed to get bucket lifecycle")
                 .tag("endpoint", endpoint_)
                 .tag("bucket", bucket)
@@ -397,6 +410,7 @@ ObjectStorageResponse S3ObjClient::check_versioning(const std::string& bucket) {
             return -1;
         }
     } else {
+        record_s3_request_failed(outcome.GetError());
         LOG_WARNING("Err for check interval: failed to get status of bucket versioning")
                 .tag("endpoint", endpoint_)
                 .tag("bucket", bucket)
@@ -414,6 +428,10 @@ ObjectStorageResponse S3ObjClient::abort_multipart_upload(ObjectStoragePathRef p
     request.WithBucket(path.bucket).WithKey(path.key).WithUploadId(upload_id);
     auto outcome = s3_put_rate_limit([&]() { return s3_client_->AbortMultipartUpload(request); });
     if (!outcome.IsSuccess()) {
+        if (outcome.GetError().GetResponseCode() == Aws::Http::HttpResponseCode::NOT_FOUND) {
+            return {ObjectStorageResponse::OK};
+        }
+        record_s3_request_failed(outcome.GetError());
         LOG_WARNING("failed to abort multipart upload")
                 .tag("endpoint", endpoint_)
                 .tag("bucket", path.bucket)
@@ -423,9 +441,6 @@ ObjectStorageResponse S3ObjClient::abort_multipart_upload(ObjectStoragePathRef p
                 .tag("error", outcome.GetError().GetMessage())
                 .tag("exception", outcome.GetError().GetExceptionName())
                 .tag("request_id", outcome.GetError().GetRequestId());
-        if (outcome.GetError().GetResponseCode() == Aws::Http::HttpResponseCode::NOT_FOUND) {
-            return {ObjectStorageResponse::OK};
-        }
         return {ObjectStorageResponse::UNDEFINED, outcome.GetError().GetMessage()};
     }
     return {ObjectStorageResponse::OK};
diff --git a/cloud/test/s3_rate_limiter_test.cpp b/cloud/test/s3_rate_limiter_test.cpp
index b1fd5a8115fa6b..05fbb2ce35ec03 100644
--- a/cloud/test/s3_rate_limiter_test.cpp
+++ b/cloud/test/s3_rate_limiter_test.cpp
@@ -29,6 +29,10 @@
 
 using namespace doris::cloud;
 
+namespace doris {
+extern bvar::Adder<int64_t> s3_put_rate_limit_rejected_count;
+} // namespace doris
+
 int main(int argc, char** argv) {
     auto conf_file = "doris_cloud.conf";
     if (!doris::cloud::config::init(conf_file, true)) {
@@ -107,18 +111,79 @@ TEST(S3RateLimiterTest, ExceedLimit) {
 }
 
 TEST(S3RateLimiterHolderTest, BvarMetric) {
-    bvar::Adder<int64_t> rate_limit_ns("rate_limit_ns");
-    bvar::Adder<int64_t> rate_limit_exceed_req_num("rate_limit_exceed_req_num");
+    bvar::Adder<int64_t> rate_limit_sleep_ns("rate_limit_sleep_ns");
+    bvar::Adder<int64_t> rate_limit_sleep_count("rate_limit_sleep_count");
+    bvar::Adder<int64_t> rate_limit_rejected_count("rate_limit_rejected_count");
 
     auto rate_limiter_holder = doris::S3RateLimiterHolder(
-            125, 250, 500, doris::metric_func_factory(rate_limit_ns, rate_limit_exceed_req_num));
+            125, 250, 251,
+            doris::metric_func_factory(rate_limit_sleep_ns, rate_limit_sleep_count,
+                                       &rate_limit_rejected_count));
     int64_t sleep_time = rate_limiter_holder.add(250);
     EXPECT_EQ(sleep_time, 0);
-    EXPECT_EQ(rate_limit_ns.get_value(), 0);
-    EXPECT_EQ(rate_limit_exceed_req_num.get_value(), 0);
+    EXPECT_EQ(rate_limit_sleep_ns.get_value(), 0);
+    EXPECT_EQ(rate_limit_sleep_count.get_value(), 0);
+    EXPECT_EQ(rate_limit_rejected_count.get_value(), 0);
 
     sleep_time = rate_limiter_holder.add(1);
-    EXPECT_GT(rate_limit_ns.get_value(), 0);
-    EXPECT_EQ(rate_limit_exceed_req_num.get_value(), 1);
     EXPECT_GT(sleep_time, 0);
-}
\ No newline at end of file
+    EXPECT_GT(rate_limit_sleep_ns.get_value(), 0);
+    EXPECT_EQ(rate_limit_sleep_count.get_value(), 1);
+    EXPECT_EQ(rate_limit_rejected_count.get_value(), 0);
+
+    sleep_time = rate_limiter_holder.add(1);
+    EXPECT_EQ(sleep_time, -1);
+    EXPECT_EQ(rate_limit_rejected_count.get_value(), 1);
+}
+
+TEST(S3RateLimiterHolderTest, ApplyS3RateLimitRecordsRejectedMetric) {
+    auto rate_limiter_holder = doris::S3RateLimiterHolder(
+            125, 250, 1, doris::s3_rate_limiter_metric_func(doris::S3RateLimitType::PUT));
+    auto rejected_count = doris::s3_put_rate_limit_rejected_count.get_value();
+
+    int64_t sleep_time =
+            doris::apply_s3_rate_limit(doris::S3RateLimitType::PUT, &rate_limiter_holder, 1);
+    EXPECT_EQ(sleep_time, 0);
+    EXPECT_EQ(doris::s3_put_rate_limit_rejected_count.get_value(), rejected_count);
+
+    sleep_time = doris::apply_s3_rate_limit(doris::S3RateLimitType::PUT, &rate_limiter_holder, 1);
+    EXPECT_EQ(sleep_time, -1);
+    EXPECT_EQ(doris::s3_put_rate_limit_rejected_count.get_value(), rejected_count + 1);
+}
+
+TEST(S3RateLimiterHolderTest, ConcurrentResetReturnsConsistentConfig) {
+    constexpr size_t config_a_speed = 101;
+    constexpr size_t config_a_burst = 102;
+    constexpr size_t config_a_limit = 103;
+    constexpr size_t config_b_speed = 201;
+    constexpr size_t config_b_burst = 202;
+    constexpr size_t config_b_limit = 203;
+
+    doris::S3RateLimiterHolder rate_limiter_holder(config_a_speed, config_a_burst, config_a_limit,
+                                                   [](int64_t) {});
+    std::atomic<bool> start {false};
+    std::thread reset_thread([&]() {
+        while (!start.load(std::memory_order_acquire)) {
+            std::this_thread::yield();
+        }
+        for (size_t i = 0; i < 10000; ++i) {
+            if (i % 2 == 0) {
+                rate_limiter_holder.reset(config_b_speed, config_b_burst, config_b_limit);
+            } else {
+                rate_limiter_holder.reset(config_a_speed, config_a_burst, config_a_limit);
+            }
+        }
+    });
+
+    start.store(true, std::memory_order_release);
+    for (size_t i = 0; i < 10000; ++i) {
+        auto result = rate_limiter_holder.add_with_config(0);
+        bool is_config_a = result.max_speed == config_a_speed &&
+                           result.max_burst == config_a_burst && result.limit == config_a_limit;
+        bool is_config_b = result.max_speed == config_b_speed &&
+                           result.max_burst == config_b_burst && result.limit == config_b_limit;
+        EXPECT_TRUE(is_config_a || is_config_b);
+    }
+
+    reset_thread.join();
+}
diff --git a/common/cpp/obj_retry_strategy.cpp b/common/cpp/obj_retry_strategy.cpp
index 58ca47f6a6808c..8399225536424c 100644
--- a/common/cpp/obj_retry_strategy.cpp
+++ b/common/cpp/obj_retry_strategy.cpp
@@ -24,6 +24,16 @@
 namespace doris {
 
 bvar::Adder<int64_t> object_request_retry_count("object_request_retry_count");
+bvar::Adder<int64_t> s3_request_retry_too_many_requests_count(
+        "s3_request_retry_too_many_requests_count");
+bvar::Adder<int64_t> s3_request_failed_too_many_requests_count(
+        "s3_request_failed_too_many_requests_count");
+
+void record_object_request_failed(int http_code) {
+    if (http_code == static_cast<int>(Aws::Http::HttpResponseCode::TOO_MANY_REQUESTS)) {
+        s3_request_failed_too_many_requests_count << 1;
+    }
+}
 
 S3CustomRetryStrategy::S3CustomRetryStrategy(int maxRetries, bool retry_slow_down)
         : DefaultRetryStrategy(maxRetries), _retry_slow_down(retry_slow_down) {}
@@ -43,6 +53,9 @@ bool S3CustomRetryStrategy::ShouldRetry(const Aws::Client::AWSError<Aws::Client:
 
     if (Aws::Http::IsRetryableHttpResponseCode(error.GetResponseCode()) || error.ShouldRetry()) {
         object_request_retry_count << 1;
+        if (error.GetResponseCode() == Aws::Http::HttpResponseCode::TOO_MANY_REQUESTS) {
+            s3_request_retry_too_many_requests_count << 1;
+        }
         LOG(INFO) << "retry due to error: " << error << ", attempt: " << attemptedRetries + 1 << "/"
                   << m_maxRetries;
         return true;
@@ -65,6 +78,10 @@ std::unique_ptr<Azure::Core::Http::RawResponse> AzureRetryRecordPolicy::Send(
         static_cast<int>(response->GetStatusCode()) < 200) {
         if (retry_count > 0) {
             object_request_retry_count << 1;
+            if (static_cast<int>(response->GetStatusCode()) ==
+                static_cast<int>(Aws::Http::HttpResponseCode::TOO_MANY_REQUESTS)) {
+                s3_request_retry_too_many_requests_count << 1;
+            }
         }
 
         // If the response is not successful, we log the retry attempt and status code.
diff --git a/common/cpp/obj_retry_strategy.h b/common/cpp/obj_retry_strategy.h
index aaf35f6b10020d..2eafe114564b21 100644
--- a/common/cpp/obj_retry_strategy.h
+++ b/common/cpp/obj_retry_strategy.h
@@ -25,6 +25,8 @@
 #endif
 
 namespace doris {
+void record_object_request_failed(int http_code);
+
 class S3CustomRetryStrategy final : public Aws::Client::DefaultRetryStrategy {
 public:
     S3CustomRetryStrategy(int maxRetries, bool retry_slow_down = true);
diff --git a/common/cpp/token_bucket_rate_limiter.cpp b/common/cpp/token_bucket_rate_limiter.cpp
index 0e30f870de368e..128501ae7cf469 100644
--- a/common/cpp/token_bucket_rate_limiter.cpp
+++ b/common/cpp/token_bucket_rate_limiter.cpp
@@ -20,6 +20,7 @@
 #include <bthread/bthread.h>
 #include <glog/logging.h> // IWYU pragma: export
 
+#include <atomic>
 #include <chrono>
 #include <mutex>
 #include <thread>
@@ -32,6 +33,18 @@ namespace doris {
 // Just 10^6.
 static constexpr auto NS = 1000000000UL;
 
+bvar::Adder<int64_t> s3_get_rate_limit_sleep_ns("s3_get_rate_limit_sleep_ns");
+bvar::Adder<int64_t> s3_get_rate_limit_sleep_count("s3_get_rate_limit_sleep_count");
+bvar::Adder<int64_t> s3_get_rate_limit_rejected_count("s3_get_rate_limit_rejected_count");
+bvar::Adder<int64_t> s3_put_rate_limit_sleep_ns("s3_put_rate_limit_sleep_ns");
+bvar::Adder<int64_t> s3_put_rate_limit_sleep_count("s3_put_rate_limit_sleep_count");
+bvar::Adder<int64_t> s3_put_rate_limit_rejected_count("s3_put_rate_limit_rejected_count");
+
+static std::atomic<int64_t> s3_get_rate_limit_sleep_log_count {0};
+static std::atomic<int64_t> s3_get_rate_limit_rejected_log_count {0};
+static std::atomic<int64_t> s3_put_rate_limit_sleep_log_count {0};
+static std::atomic<int64_t> s3_put_rate_limit_rejected_log_count {0};
+
 class TokenBucketRateLimiter::SimpleSpinLock {
 public:
     SimpleSpinLock() = default;
@@ -118,13 +131,20 @@ TokenBucketRateLimiterHolder::TokenBucketRateLimiterHolder(size_t max_speed, siz
           metric_func(std::move(metric_func)) {}
 
 int64_t TokenBucketRateLimiterHolder::add(size_t amount) {
-    int64_t sleep;
+    return add_with_config(amount).sleep_duration;
+}
+
+TokenBucketRateLimiterResult TokenBucketRateLimiterHolder::add_with_config(size_t amount) {
+    TokenBucketRateLimiterResult result;
     {
         std::shared_lock read {rate_limiter_rw_lock};
-        sleep = rate_limiter->add(amount);
+        result = {.sleep_duration = rate_limiter->add(amount),
+                  .max_speed = rate_limiter->get_max_speed(),
+                  .max_burst = rate_limiter->get_max_burst(),
+                  .limit = rate_limiter->get_limit()};
     }
-    metric_func(sleep);
-    return sleep;
+    metric_func(result.sleep_duration);
+    return result;
 }
 
 int TokenBucketRateLimiterHolder::reset(size_t max_speed, size_t max_burst, size_t limit) {
@@ -135,6 +155,21 @@ int TokenBucketRateLimiterHolder::reset(size_t max_speed, size_t max_burst, size
     return 0;
 }
 
+size_t TokenBucketRateLimiterHolder::get_max_speed() const {
+    std::shared_lock read {rate_limiter_rw_lock};
+    return rate_limiter->get_max_speed();
+}
+
+size_t TokenBucketRateLimiterHolder::get_max_burst() const {
+    std::shared_lock read {rate_limiter_rw_lock};
+    return rate_limiter->get_max_burst();
+}
+
+size_t TokenBucketRateLimiterHolder::get_limit() const {
+    std::shared_lock read {rate_limiter_rw_lock};
+    return rate_limiter->get_limit();
+}
+
 std::string to_string(S3RateLimitType type) {
     switch (type) {
     case S3RateLimitType::GET:
@@ -154,4 +189,52 @@ S3RateLimitType string_to_s3_rate_limit_type(std::string_view value) {
     }
     return S3RateLimitType::UNKNOWN;
 }
+
+std::function<void(int64_t)> s3_rate_limiter_metric_func(S3RateLimitType type) {
+    switch (type) {
+    case S3RateLimitType::GET:
+        return metric_func_factory(s3_get_rate_limit_sleep_ns, s3_get_rate_limit_sleep_count,
+                                   &s3_get_rate_limit_rejected_count);
+    case S3RateLimitType::PUT:
+        return metric_func_factory(s3_put_rate_limit_sleep_ns, s3_put_rate_limit_sleep_count,
+                                   &s3_put_rate_limit_rejected_count);
+    default:
+        return [](int64_t) {};
+    }
+}
+
+int64_t apply_s3_rate_limit(S3RateLimitType type, S3RateLimiterHolder* rate_limiter,
+                            int64_t log_interval) {
+    auto result = rate_limiter->add_with_config(1);
+    auto sleep_duration = result.sleep_duration;
+    if (log_interval <= 0 || sleep_duration == 0) {
+        return sleep_duration;
+    }
+
+    auto is_get = type == S3RateLimitType::GET;
+    auto* sleep_log_count =
+            is_get ? &s3_get_rate_limit_sleep_log_count : &s3_put_rate_limit_sleep_log_count;
+    auto* rejected_log_count =
+            is_get ? &s3_get_rate_limit_rejected_log_count : &s3_put_rate_limit_rejected_log_count;
+
+    if (sleep_duration > 0) {
+        int64_t count = sleep_log_count->fetch_add(1, std::memory_order_relaxed) + 1;
+        if (count == 1 || count % log_interval == 0) {
+            LOG(INFO) << "S3 " << to_string(type) << " request is throttled by local rate limiter"
+                      << ", sleep_ms=" << sleep_duration / 1000000 << ", sleep_count=" << count
+                      << ", token_per_second=" << result.max_speed
+                      << ", bucket_tokens=" << result.max_burst << ", token_limit=" << result.limit;
+        }
+    } else {
+        int64_t count = rejected_log_count->fetch_add(1, std::memory_order_relaxed) + 1;
+        if (count == 1 || count % log_interval == 0) {
+            LOG(WARNING) << "S3 " << to_string(type) << " request is rejected by local rate limiter"
+                         << ", rejected_count=" << count
+                         << ", token_per_second=" << result.max_speed
+                         << ", bucket_tokens=" << result.max_burst
+                         << ", token_limit=" << result.limit;
+        }
+    }
+    return sleep_duration;
+}
 } // namespace doris
diff --git a/common/cpp/token_bucket_rate_limiter.h b/common/cpp/token_bucket_rate_limiter.h
index dbeca3b897db03..57f9205527acc6 100644
--- a/common/cpp/token_bucket_rate_limiter.h
+++ b/common/cpp/token_bucket_rate_limiter.h
@@ -21,6 +21,8 @@
 #include <functional>
 #include <memory>
 #include <shared_mutex>
+#include <string>
+#include <string_view>
 
 namespace doris {
 enum class S3RateLimitType : int {
@@ -32,11 +34,15 @@ enum class S3RateLimitType : int {
 extern std::string to_string(S3RateLimitType type);
 extern S3RateLimitType string_to_s3_rate_limit_type(std::string_view value);
 
-inline auto metric_func_factory(bvar::Adder<int64_t>& ns_bvar, bvar::Adder<int64_t>& req_num_bvar) {
-    return [&](int64_t ns) {
+inline auto metric_func_factory(bvar::Adder<int64_t>& sleep_ns_bvar,
+                                bvar::Adder<int64_t>& sleep_count_bvar,
+                                bvar::Adder<int64_t>* rejected_count_bvar = nullptr) {
+    return [&, rejected_count_bvar](int64_t ns) {
         if (ns > 0) {
-            ns_bvar << ns;
-            req_num_bvar << 1;
+            sleep_ns_bvar << ns;
+            sleep_count_bvar << 1;
+        } else if (ns < 0 && rejected_count_bvar != nullptr) {
+            *rejected_count_bvar << 1;
         }
     };
 }
@@ -71,6 +77,13 @@ class TokenBucketRateLimiter {
     long _prev_ns_count {0}; // Previous `add` call time (in nanoseconds).
 };
 
+struct TokenBucketRateLimiterResult {
+    int64_t sleep_duration;
+    size_t max_speed;
+    size_t max_burst;
+    size_t limit;
+};
+
 class TokenBucketRateLimiterHolder {
 public:
     TokenBucketRateLimiterHolder(size_t max_speed, size_t max_burst, size_t limit,
@@ -78,17 +91,16 @@ class TokenBucketRateLimiterHolder {
     ~TokenBucketRateLimiterHolder();
 
     int64_t add(size_t amount);
+    TokenBucketRateLimiterResult add_with_config(size_t amount);
 
     int reset(size_t max_speed, size_t max_burst, size_t limit);
 
-    size_t get_max_speed() const { return rate_limiter->get_max_speed(); }
-
-    size_t get_max_burst() const { return rate_limiter->get_max_burst(); }
-
-    size_t get_limit() const { return rate_limiter->get_limit(); }
+    size_t get_max_speed() const;
+    size_t get_max_burst() const;
+    size_t get_limit() const;
 
 private:
-    std::shared_mutex rate_limiter_rw_lock;
+    mutable std::shared_mutex rate_limiter_rw_lock;
     std::unique_ptr<TokenBucketRateLimiter> rate_limiter;
     // Record the correspoding sleeping time(unit is ms)
     std::function<void(int64_t)> metric_func;
@@ -96,4 +108,8 @@ class TokenBucketRateLimiterHolder {
 
 using S3RateLimiter = TokenBucketRateLimiter;
 using S3RateLimiterHolder = TokenBucketRateLimiterHolder;
-} // namespace doris
\ No newline at end of file
+
+std::function<void(int64_t)> s3_rate_limiter_metric_func(S3RateLimitType type);
+int64_t apply_s3_rate_limit(S3RateLimitType type, S3RateLimiterHolder* rate_limiter,
+                            int64_t log_interval);
+} // namespace doris