diff --git a/CMakeLists.txt b/CMakeLists.txt
index 1d7d00002..4cfcb85a4 100644
--- a/CMakeLists.txt
+++ b/CMakeLists.txt
@@ -487,9 +487,6 @@ install(FILES "${CMAKE_CURRENT_BINARY_DIR}/PaimonConfig.cmake"
 
 config_summary_message()
 
-if(PAIMON_ENABLE_S3)
-    find_package(CURL REQUIRED)
-endif()
 add_subdirectory(src/paimon)
 add_subdirectory(src/paimon/fs/local)
 add_subdirectory(src/paimon/fs/jindo)
diff --git a/cmake_modules/ThirdpartyToolchain.cmake b/cmake_modules/ThirdpartyToolchain.cmake
index 872c70b8c..797b25139 100644
--- a/cmake_modules/ThirdpartyToolchain.cmake
+++ b/cmake_modules/ThirdpartyToolchain.cmake
@@ -1890,6 +1890,20 @@ endif()
 if(PAIMON_ENABLE_S3)
     include(BuildAwsAuth)
     build_aws_auth()
+
+    find_package(CURL 7.62 REQUIRED)
+    set(PAIMON_CURL_IS_STATIC OFF)
+    foreach(PAIMON_CURL_LIBRARY IN LISTS CURL_LIBRARIES)
+        if(PAIMON_CURL_LIBRARY MATCHES "\\${CMAKE_STATIC_LIBRARY_SUFFIX}$")
+            set(PAIMON_CURL_IS_STATIC ON)
+            break()
+        endif()
+    endforeach()
+
+    set(PAIMON_HTTP_LINK_LIBS CURL::libcurl)
+    if(PAIMON_CURL_IS_STATIC)
+        list(APPEND PAIMON_HTTP_LINK_LIBS OpenSSL::SSL OpenSSL::Crypto)
+    endif()
 endif()
 if(PAIMON_ENABLE_LUMINA)
     build_lumina()
diff --git a/src/paimon/CMakeLists.txt b/src/paimon/CMakeLists.txt
index 409c58a50..22e38d189 100644
--- a/src/paimon/CMakeLists.txt
+++ b/src/paimon/CMakeLists.txt
@@ -181,7 +181,6 @@ set(PAIMON_COMMON_SRCS
 if(PAIMON_ENABLE_S3)
     list(APPEND PAIMON_COMMON_SRCS common/fs/http_client.cpp
          common/fs/object_store_file_system.cpp)
-    set(PAIMON_OBJECT_STORE_LINK_LIBS CURL::libcurl)
 endif()
 
 set(PAIMON_CORE_SRCS
@@ -416,7 +415,7 @@ add_paimon_lib(paimon
                xxhash
                Threads::Threads
                RapidJSON
-               ${PAIMON_OBJECT_STORE_LINK_LIBS}
+               ${PAIMON_HTTP_LINK_LIBS}
                STATIC_LINK_LIBS
                arrow
                tbb
@@ -426,7 +425,7 @@ add_paimon_lib(paimon
                xxhash
                Threads::Threads
                RapidJSON
-               ${PAIMON_OBJECT_STORE_LINK_LIBS}
+               ${PAIMON_HTTP_LINK_LIBS}
                SHARED_LINK_FLAGS
                ${PAIMON_VERSION_SCRIPT_FLAGS})
 
@@ -852,6 +851,14 @@ if(PAIMON_BUILD_TESTS)
     if(PAIMON_ENABLE_S3)
         list(APPEND PAIMON_OBJECT_STORE_FS_TEST_SOURCES
              common/fs/object_store_file_system_test.cpp)
+
+        add_paimon_test(http_client_test
+                        SOURCES
+                        common/fs/http_client_test.cpp
+                        STATIC_LINK_LIBS
+                        paimon_shared
+                        fmt
+                        ${GTEST_LINK_TOOLCHAIN})
     endif()
 
     add_paimon_test(fs_test
diff --git a/src/paimon/common/fs/http_client.cpp b/src/paimon/common/fs/http_client.cpp
index afd816ce8..919d0eec9 100644
--- a/src/paimon/common/fs/http_client.cpp
+++ b/src/paimon/common/fs/http_client.cpp
@@ -18,6 +18,7 @@
 
 #include <curl/curl.h>
 
+#include <algorithm>
 #include <chrono>
 #include <mutex>
 #include <thread>
@@ -33,6 +34,7 @@ namespace paimon {
 namespace {
 
 constexpr int32_t kMaxAttempts = 3;
+constexpr size_t kMaxErrorBodySize = 4096;
 
 class CurlGlobalGuard {
  public:
@@ -78,6 +80,8 @@ size_t WriteCallback(char* data, size_t size, size_t count, void* user_data) {
     curl_easy_getinfo(context->handle, CURLINFO_RESPONSE_CODE, &status_code);
     context->response.status_code = static_cast<int32_t>(status_code);
     if (status_code >= 300) {
+        size_t remaining = kMaxErrorBodySize - context->response.error_body.size();
+        context->response.error_body.append(data, std::min(bytes, remaining));
         context->response.body_size += static_cast<int64_t>(bytes);
         return bytes;
     }
@@ -118,7 +122,8 @@ bool IsRetryable(CURLcode code, int64_t status_code) {
 
 class CurlHttpClient::Impl {
  public:
-    Impl() : guard_(GetCurlGlobalGuard()) {}
+    explicit Impl(CurlHttpClientOptions options)
+        : guard_(GetCurlGlobalGuard()), options_(std::move(options)) {}
 
     ~Impl() {
         for (CURL* handle : handles_) {
@@ -142,13 +147,19 @@ class CurlHttpClient::Impl {
         handles_.push_back(handle);
     }
 
+    const CurlHttpClientOptions& options() const {
+        return options_;
+    }
+
  private:
     std::shared_ptr<CurlGlobalGuard> guard_;
+    CurlHttpClientOptions options_;
     mutable std::mutex mutex_;
     mutable std::vector<CURL*> handles_;
 };
 
-CurlHttpClient::CurlHttpClient() : impl_(std::make_unique<Impl>()) {}
+CurlHttpClient::CurlHttpClient(CurlHttpClientOptions options)
+    : impl_(std::make_unique<Impl>(std::move(options))) {}
 CurlHttpClient::~CurlHttpClient() = default;
 
 Result<HttpResponse> CurlHttpClient::Execute(const HttpRequest& request,
@@ -172,7 +183,18 @@ Result<HttpResponse> CurlHttpClient::Execute(const HttpRequest& request,
         curl_easy_setopt(handle, CURLOPT_URL, request.url.c_str());
         curl_easy_setopt(handle, CURLOPT_HTTPHEADER, headers);
         curl_easy_setopt(handle, CURLOPT_NOSIGNAL, 1L);
-        curl_easy_setopt(handle, CURLOPT_CONNECTTIMEOUT_MS, 30000L);
+        curl_easy_setopt(
+            handle, CURLOPT_CONNECTTIMEOUT_MS,
+            static_cast<long>(impl_->options().connect_timeout_ms));  // NOLINT(runtime/int)
+        curl_easy_setopt(
+            handle, CURLOPT_TIMEOUT_MS,
+            static_cast<long>(impl_->options().request_timeout_ms));  // NOLINT(runtime/int)
+        curl_easy_setopt(handle, CURLOPT_LOW_SPEED_LIMIT,
+                         static_cast<long>(  // NOLINT(runtime/int): required by curl.
+                             impl_->options().low_speed_limit_bytes_per_second));
+        curl_easy_setopt(handle, CURLOPT_LOW_SPEED_TIME,
+                         static_cast<long>(  // NOLINT(runtime/int): required by curl.
+                             impl_->options().low_speed_time_seconds));
         curl_easy_setopt(handle, CURLOPT_WRITEFUNCTION, WriteCallback);
         curl_easy_setopt(handle, CURLOPT_WRITEDATA, &context);
         curl_easy_setopt(handle, CURLOPT_HEADERFUNCTION, HeaderCallback);
@@ -196,8 +218,7 @@ Result<HttpResponse> CurlHttpClient::Execute(const HttpRequest& request,
             (context.response.body_size > 0 && response_code < 300) ||
             attempt + 1 == kMaxAttempts) {
             if (code == CURLE_OK) {
-                return Status::IOError(fmt::format("HTTP request to {} returned status {}",
-                                                   request.url, response_code));
+                return context.response;
             }
             return Status::IOError(fmt::format("HTTP request to {} failed: {} (status {})",
                                                request.url, curl_easy_strerror(code),
diff --git a/src/paimon/common/fs/http_client.h b/src/paimon/common/fs/http_client.h
index 586972223..eb0695c2d 100644
--- a/src/paimon/common/fs/http_client.h
+++ b/src/paimon/common/fs/http_client.h
@@ -41,6 +41,7 @@ struct HttpResponse {
     int32_t status_code = 0;
     HttpHeaders headers;
     int64_t body_size = 0;
+    std::string error_body;
 };
 
 class PAIMON_EXPORT HttpClient {
@@ -50,9 +51,16 @@ class PAIMON_EXPORT HttpClient {
                                          const HttpBodyConsumer& consumer) const = 0;
 };
 
+struct CurlHttpClientOptions {
+    int64_t connect_timeout_ms = 30000;
+    int64_t request_timeout_ms = 300000;
+    int64_t low_speed_limit_bytes_per_second = 1;
+    int64_t low_speed_time_seconds = 30;
+};
+
 class PAIMON_EXPORT CurlHttpClient : public HttpClient {
  public:
-    CurlHttpClient();
+    explicit CurlHttpClient(CurlHttpClientOptions options = {});
     ~CurlHttpClient() override;
 
     Result<HttpResponse> Execute(const HttpRequest& request,
diff --git a/src/paimon/common/fs/http_client_test.cpp b/src/paimon/common/fs/http_client_test.cpp
new file mode 100644
index 000000000..e0416cedd
--- /dev/null
+++ b/src/paimon/common/fs/http_client_test.cpp
@@ -0,0 +1,214 @@
+/*
+ * Copyright 2026-present Alibaba Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#include "paimon/common/fs/http_client.h"
+
+#include <arpa/inet.h>
+#include <gtest/gtest.h>
+#include <netinet/in.h>
+#include <sys/socket.h>
+#include <unistd.h>
+
+#include <chrono>
+#include <cstdint>
+#include <cstdlib>
+#include <cstring>
+#include <optional>
+#include <string>
+#include <thread>
+#include <utility>
+#include <vector>
+
+#include "fmt/format.h"
+#include "paimon/testing/utils/testharness.h"
+
+namespace paimon::test {
+namespace {
+
+class ScopedEnvironmentVariable {
+ public:
+    ScopedEnvironmentVariable(const char* name, const char* value) : name_(name) {
+        const char* previous = std::getenv(name);
+        if (previous != nullptr) {
+            previous_ = previous;
+        }
+        setenv(name, value, 1);
+    }
+
+    ~ScopedEnvironmentVariable() {
+        if (previous_) {
+            setenv(name_.c_str(), previous_->c_str(), 1);
+        } else {
+            unsetenv(name_.c_str());
+        }
+    }
+
+ private:
+    std::string name_;
+    std::optional<std::string> previous_;
+};
+
+class HttpTestServer {
+ public:
+    explicit HttpTestServer(std::vector<std::string> responses,
+                            std::chrono::milliseconds response_delay = {})
+        : responses_(std::move(responses)), response_delay_(response_delay) {
+        socket_ = socket(AF_INET, SOCK_STREAM, 0);
+        if (socket_ < 0) {
+            return;
+        }
+        int32_t reuse_address = 1;
+        setsockopt(socket_, SOL_SOCKET, SO_REUSEADDR, &reuse_address, sizeof(reuse_address));
+        sockaddr_in address{};
+        address.sin_family = AF_INET;
+        address.sin_addr.s_addr = htonl(INADDR_LOOPBACK);
+        address.sin_port = 0;
+        if (bind(socket_, reinterpret_cast<sockaddr*>(&address), sizeof(address)) != 0 ||
+            listen(socket_, 4) != 0) {
+            close(socket_);
+            socket_ = -1;
+            return;
+        }
+        socklen_t address_size = sizeof(address);
+        if (getsockname(socket_, reinterpret_cast<sockaddr*>(&address), &address_size) != 0) {
+            close(socket_);
+            socket_ = -1;
+            return;
+        }
+        port_ = ntohs(address.sin_port);
+        thread_ = std::thread([this] { Serve(); });
+    }
+
+    ~HttpTestServer() {
+        if (socket_ >= 0) {
+            shutdown(socket_, SHUT_RDWR);
+            close(socket_);
+        }
+        if (thread_.joinable()) {
+            thread_.join();
+        }
+    }
+
+    bool ok() const {
+        return socket_ >= 0;
+    }
+
+    std::string url() const {
+        return fmt::format("http://127.0.0.1:{}/test", port_);
+    }
+
+ private:
+    void Serve() const {
+        for (const std::string& response : responses_) {
+            int32_t client = accept(socket_, nullptr, nullptr);
+            if (client < 0) {
+                return;
+            }
+            char request[4096];
+            recv(client, request, sizeof(request), 0);
+            if (response_delay_.count() > 0) {
+                std::this_thread::sleep_for(response_delay_);
+            }
+            size_t sent = 0;
+            while (sent < response.size()) {
+                ssize_t result = send(client, response.data() + sent, response.size() - sent, 0);
+                if (result <= 0) {
+                    break;
+                }
+                sent += static_cast<size_t>(result);
+            }
+            shutdown(client, SHUT_RDWR);
+            close(client);
+        }
+    }
+
+    int32_t socket_ = -1;
+    uint16_t port_ = 0;
+    std::vector<std::string> responses_;
+    std::chrono::milliseconds response_delay_;
+    std::thread thread_;
+};
+
+TEST(CurlHttpClientTest, TestCapturesBoundedErrorBodyAndHeaders) {
+    ScopedEnvironmentVariable no_proxy("NO_PROXY", "127.0.0.1");
+    ScopedEnvironmentVariable lowercase_no_proxy("no_proxy", "127.0.0.1");
+    std::string body(5000, 'x');
+    HttpTestServer server(
+        {"HTTP/1.1 403 Forbidden\r\nContent-Length: 5000\r\n"
+         "X-Amz-Bucket-Region: eu-west-1\r\nConnection: close\r\n\r\n" +
+         body});
+    ASSERT_TRUE(server.ok());
+
+    CurlHttpClient client;
+    bool consumer_called = false;
+    ASSERT_OK_AND_ASSIGN(HttpResponse response,
+                         client.Execute({HttpMethod::GET, server.url(), {}},
+                                        [&consumer_called](const char*, int64_t) {
+                                            consumer_called = true;
+                                            return Status::OK();
+                                        }));
+    ASSERT_EQ(response.status_code, 403);
+    ASSERT_EQ(response.body_size, 5000);
+    ASSERT_EQ(response.error_body.size(), 4096);
+    ASSERT_EQ(response.headers["x-amz-bucket-region"], "eu-west-1");
+    ASSERT_FALSE(consumer_called);
+}
+
+TEST(CurlHttpClientTest, TestConfiguredRequestTimeout) {
+    ScopedEnvironmentVariable no_proxy("NO_PROXY", "127.0.0.1");
+    ScopedEnvironmentVariable lowercase_no_proxy("no_proxy", "127.0.0.1");
+    HttpTestServer server({"", "", ""}, std::chrono::milliseconds(100));
+    ASSERT_TRUE(server.ok());
+    CurlHttpClientOptions options;
+    options.connect_timeout_ms = 100;
+    options.request_timeout_ms = 50;
+    options.low_speed_limit_bytes_per_second = 0;
+    options.low_speed_time_seconds = 0;
+    CurlHttpClient client(options);
+
+    auto start = std::chrono::steady_clock::now();
+    Result<HttpResponse> response = client.Execute(
+        {HttpMethod::GET, server.url(), {}}, [](const char*, int64_t) { return Status::OK(); });
+    auto elapsed = std::chrono::steady_clock::now() - start;
+    ASSERT_TRUE(response.status().IsIOError());
+    ASSERT_LT(elapsed, std::chrono::seconds(3));
+}
+
+TEST(CurlHttpClientTest, TestRetriesServerErrors) {
+    ScopedEnvironmentVariable no_proxy("NO_PROXY", "127.0.0.1");
+    ScopedEnvironmentVariable lowercase_no_proxy("no_proxy", "127.0.0.1");
+    std::string server_error =
+        "HTTP/1.1 500 Internal Server Error\r\nContent-Length: 5\r\n"
+        "Connection: close\r\n\r\nerror";
+    HttpTestServer server(
+        {server_error, server_error,
+         "HTTP/1.1 200 OK\r\nContent-Length: 4\r\nConnection: close\r\n\r\ndata"});
+    ASSERT_TRUE(server.ok());
+
+    CurlHttpClient client;
+    std::string body;
+    ASSERT_OK_AND_ASSIGN(HttpResponse response,
+                         client.Execute({HttpMethod::GET, server.url(), {}},
+                                        [&body](const char* data, int64_t size) {
+                                            body.append(data, static_cast<size_t>(size));
+                                            return Status::OK();
+                                        }));
+    ASSERT_EQ(response.status_code, 200);
+    ASSERT_EQ(body, "data");
+}
+
+}  // namespace
+}  // namespace paimon::test
diff --git a/src/paimon/fs/s3/CMakeLists.txt b/src/paimon/fs/s3/CMakeLists.txt
index d419bdbbd..7e8a8f110 100644
--- a/src/paimon/fs/s3/CMakeLists.txt
+++ b/src/paimon/fs/s3/CMakeLists.txt
@@ -23,6 +23,7 @@ if(PAIMON_ENABLE_S3)
                    paimon_shared
                    CURL::libcurl
                    STATIC_LINK_LIBS
+                   arrow
                    CURL::libcurl
                    aws_auth_minimal
                    fmt
diff --git a/src/paimon/fs/s3/s3_file_system.cpp b/src/paimon/fs/s3/s3_file_system.cpp
index c3f9d376a..b5acd8d9b 100644
--- a/src/paimon/fs/s3/s3_file_system.cpp
+++ b/src/paimon/fs/s3/s3_file_system.cpp
@@ -16,6 +16,7 @@
 
 #include "paimon/fs/s3/s3_file_system.h"
 
+#include <arrow/util/value_parsing.h>
 #include <aws/auth/auth.h>
 #include <aws/auth/credentials.h>
 #include <aws/auth/signable.h>
@@ -33,9 +34,12 @@
 
 #include <algorithm>
 #include <cctype>
+#include <climits>
 #include <condition_variable>
+#include <cstdint>
 #include <cstdlib>
 #include <cstring>
+#include <ctime>
 #include <limits>
 #include <mutex>
 #include <optional>
@@ -52,10 +56,10 @@ namespace paimon::s3 {
 namespace {
 
 Result<std::string> PercentEncode(std::string_view value, bool preserve_slash) {
-    if (value.size() > std::numeric_limits<int>::max()) {
+    if (value.size() > std::numeric_limits<int32_t>::max()) {
         return Status::IOError("S3 URL component is too large to encode");
     }
-    char* encoded = curl_easy_escape(nullptr, value.data(), static_cast<int>(value.size()));
+    char* encoded = curl_easy_escape(nullptr, value.data(), static_cast<int32_t>(value.size()));
     if (encoded == nullptr) {
         return Status::IOError("failed to URL encode S3 component");
     }
@@ -69,19 +73,18 @@ Result<std::string> PercentEncode(std::string_view value, bool preserve_slash) {
 
 Result<std::string> PercentDecode(std::string_view value, const std::string& field) {
     for (size_t position = 0; position < value.size(); ++position) {
-        if (value[position] == '%' &&
-            (position + 2 >= value.size() ||
-             !std::isxdigit(static_cast<unsigned char>(value[position + 1])) ||
-             !std::isxdigit(static_cast<unsigned char>(value[position + 2])))) {
+        if (value[position] == '%' && (position + 2 >= value.size() ||
+                                       !std::isxdigit(static_cast<uint8_t>(value[position + 1])) ||
+                                       !std::isxdigit(static_cast<uint8_t>(value[position + 2])))) {
             return Status::IOError(fmt::format("invalid URL encoding in S3 {}", field));
         }
     }
-    if (value.size() > std::numeric_limits<int>::max()) {
+    if (value.size() > std::numeric_limits<int32_t>::max()) {
         return Status::IOError(fmt::format("S3 {} is too large to URL decode", field));
     }
-    int decoded_size = 0;
-    char* decoded =
-        curl_easy_unescape(nullptr, value.data(), static_cast<int>(value.size()), &decoded_size);
+    int32_t decoded_size = 0;
+    char* decoded = curl_easy_unescape(nullptr, value.data(), static_cast<int32_t>(value.size()),
+                                       &decoded_size);
     if (decoded == nullptr) {
         return Status::IOError(fmt::format("failed to URL decode S3 {}", field));
     }
@@ -98,11 +101,24 @@ Result<int64_t> ParseNonNegativeInt64(const std::string& value, const std::strin
     return *result;
 }
 
-int64_t ParseModificationTime(const std::string& value) {
+int64_t ParseHttpModificationTime(const std::string& value) {
     time_t seconds = curl_getdate(value.c_str(), nullptr);
     return seconds == static_cast<time_t>(-1) ? 0 : static_cast<int64_t>(seconds) * 1000;
 }
 
+int64_t ParseIso8601ModificationTime(const std::string& value) {
+    if (value.size() < 20 || value[4] != '-' || value[7] != '-' || value[10] != 'T' ||
+        value[13] != ':' || value[16] != ':' || value.back() != 'Z') {
+        return 0;
+    }
+    static const std::shared_ptr<arrow::TimestampParser> parser =
+        arrow::TimestampParser::MakeISO8601();
+    int64_t milliseconds = 0;
+    return (*parser)(value.data(), value.size(), arrow::TimeUnit::MILLI, &milliseconds)
+               ? milliseconds
+               : 0;
+}
+
 std::string XmlUnescape(const std::string& value) {
     const std::pair<const char*, const char*> entities[] = {
         {"&amp;", "&"}, {"&lt;", "<"}, {"&gt;", ">"}, {"&quot;", "\""}, {"&apos;", "'"}};
@@ -261,14 +277,11 @@ class AwsAuthRuntime {
 };
 
 Result<AwsAuthRuntime*> GetAwsAuthRuntime() {
-    static const Result<AwsAuthRuntime*> runtime = [] {
-        Result<std::unique_ptr<AwsAuthRuntime>> created = AwsAuthRuntime::Create();
-        if (!created.ok()) {
-            return Result<AwsAuthRuntime*>(created.status());
-        }
-        return Result<AwsAuthRuntime*>(std::move(created).value().release());
-    }();
-    return runtime;
+    static const Result<std::unique_ptr<AwsAuthRuntime>> runtime = AwsAuthRuntime::Create();
+    if (!runtime.ok()) {
+        return runtime.status();
+    }
+    return runtime.value().get();
 }
 
 aws_byte_cursor Cursor(const std::string& value) {
@@ -298,6 +311,18 @@ const char* CanonicalS3OptionName(const std::string& option) {
     if (option == "profile") {
         return kS3ProfileOption;
     }
+    if (option == "connection.establish.timeout") {
+        return kS3ConnectionEstablishTimeoutOption;
+    }
+    if (option == "connection.request.timeout") {
+        return kS3ConnectionRequestTimeoutOption;
+    }
+    if (option == "low-speed-limit") {
+        return kS3LowSpeedLimitOption;
+    }
+    if (option == "low-speed-time-seconds") {
+        return kS3LowSpeedTimeSecondsOption;
+    }
     return nullptr;
 }
 
@@ -325,6 +350,34 @@ std::map<std::string, std::string> NormalizeS3Options(
     return normalized;
 }
 
+Result<int64_t> ResolveCurlOption(const std::map<std::string, std::string>& options,
+                                  const std::string& key, int64_t default_value) {
+    PAIMON_ASSIGN_OR_RAISE(int64_t value,
+                           OptionsUtils::GetValueFromMap<int64_t>(options, key, default_value));
+    if (value < 0 || value > LONG_MAX) {
+        return Status::Invalid(fmt::format("{} must be between 0 and {}", key, LONG_MAX));
+    }
+    return value;
+}
+
+Result<CurlHttpClientOptions> ResolveCurlHttpClientOptions(
+    const std::map<std::string, std::string>& options) {
+    CurlHttpClientOptions result;
+    PAIMON_ASSIGN_OR_RAISE(
+        result.connect_timeout_ms,
+        ResolveCurlOption(options, kS3ConnectionEstablishTimeoutOption, result.connect_timeout_ms));
+    PAIMON_ASSIGN_OR_RAISE(
+        result.request_timeout_ms,
+        ResolveCurlOption(options, kS3ConnectionRequestTimeoutOption, result.request_timeout_ms));
+    PAIMON_ASSIGN_OR_RAISE(result.low_speed_limit_bytes_per_second,
+                           ResolveCurlOption(options, kS3LowSpeedLimitOption,
+                                             result.low_speed_limit_bytes_per_second));
+    PAIMON_ASSIGN_OR_RAISE(
+        result.low_speed_time_seconds,
+        ResolveCurlOption(options, kS3LowSpeedTimeSecondsOption, result.low_speed_time_seconds));
+    return result;
+}
+
 std::shared_ptr<aws_credentials_provider> WrapProvider(aws_credentials_provider* provider) {
     return std::shared_ptr<aws_credentials_provider>(provider, aws_credentials_provider_release);
 }
@@ -352,16 +405,20 @@ Result<std::string> ResolveRegion(const std::map<std::string, std::string>& opti
         profile_override_ptr = &profile_override;
     }
     aws_string* config_path = aws_get_config_file_path(runtime->allocator(), nullptr);
+    ScopeGuard destroy_config_path([config_path] { aws_string_destroy(config_path); });
     aws_string* profile_name = aws_get_profile_name(runtime->allocator(), profile_override_ptr);
+    ScopeGuard destroy_profile_name([profile_name] { aws_string_destroy(profile_name); });
     aws_profile_collection* profiles = config_path == nullptr
                                            ? nullptr
                                            : aws_profile_collection_new_from_file(
                                                  runtime->allocator(), config_path, AWS_PST_CONFIG);
+    ScopeGuard release_profiles([profiles] { aws_profile_collection_release(profiles); });
     const aws_profile* selected_profile =
         profiles == nullptr || profile_name == nullptr
             ? nullptr
             : aws_profile_collection_get_profile(profiles, profile_name);
     aws_string* region_name = aws_string_new_from_c_str(runtime->allocator(), "region");
+    ScopeGuard destroy_region_name([region_name] { aws_string_destroy(region_name); });
     const aws_profile_property* property =
         selected_profile == nullptr || region_name == nullptr
             ? nullptr
@@ -369,10 +426,6 @@ Result<std::string> ResolveRegion(const std::map<std::string, std::string>& opti
     const aws_string* value =
         property == nullptr ? nullptr : aws_profile_property_get_value(property);
     std::string resolved = value == nullptr ? "" : aws_string_c_str(value);
-    aws_string_destroy(region_name);
-    aws_profile_collection_release(profiles);
-    aws_string_destroy(profile_name);
-    aws_string_destroy(config_path);
     return resolved.empty() ? "us-east-1" : resolved;
 }
 
@@ -407,6 +460,11 @@ Result<std::shared_ptr<aws_credentials_provider>> MakeCredentialsProvider(
     }
 
     std::vector<aws_credentials_provider*> providers;
+    ScopeGuard release_providers([&providers] {
+        for (aws_credentials_provider* provider : providers) {
+            aws_credentials_provider_release(provider);
+        }
+    });
     aws_credentials_provider_environment_options environment_options{};
     providers.push_back(
         aws_credentials_provider_new_environment(runtime->allocator(), &environment_options));
@@ -453,18 +511,15 @@ Result<std::shared_ptr<aws_credentials_provider>> MakeCredentialsProvider(
     chain_options.provider_count = providers.size();
     aws_credentials_provider* chain =
         aws_credentials_provider_new_chain(runtime->allocator(), &chain_options);
-    for (aws_credentials_provider* provider : providers) {
-        aws_credentials_provider_release(provider);
-    }
     if (chain == nullptr) {
         return std::shared_ptr<aws_credentials_provider>();
     }
+    ScopeGuard release_chain([chain] { aws_credentials_provider_release(chain); });
     aws_credentials_provider_cached_options cached_options{};
     cached_options.source = chain;
     cached_options.refresh_time_in_milliseconds = 15 * 60 * 1000;
     aws_credentials_provider* cached =
         aws_credentials_provider_new_cached(runtime->allocator(), &cached_options);
-    aws_credentials_provider_release(chain);
     return WrapProvider(cached);
 }
 
@@ -486,7 +541,7 @@ Result<Endpoint> ParseEndpoint(std::string endpoint) {
     CURLUcode code = curl_url_set(url, CURLUPART_URL, endpoint.c_str(), 0);
     if (code != CURLUE_OK) {
         return Status::Invalid(
-            fmt::format("invalid S3 endpoint {}: code {}", endpoint, static_cast<int>(code)));
+            fmt::format("invalid S3 endpoint {}: code {}", endpoint, static_cast<int32_t>(code)));
     }
     auto get_part = [url, &endpoint](CURLUPart part, CURLUcode no_value,
                                      const char* name) -> Result<std::optional<std::string>> {
@@ -497,7 +552,7 @@ Result<Endpoint> ParseEndpoint(std::string endpoint) {
         }
         if (result != CURLUE_OK) {
             return Status::Invalid(fmt::format("invalid S3 endpoint {} {}: code {}", endpoint, name,
-                                               static_cast<int>(result)));
+                                               static_cast<int32_t>(result)));
         }
         ScopeGuard free_value([value] { curl_free(value); });
         return std::optional<std::string>(value);
@@ -522,22 +577,30 @@ Result<Endpoint> ParseEndpoint(std::string endpoint) {
         return Status::Invalid(fmt::format(
             "S3 endpoint {} must not contain user, password, query, or fragment", endpoint));
     }
-    PAIMON_ASSIGN_OR_RAISE(std::optional<std::string> port,
-                           get_part(CURLUPART_PORT, CURLUE_NO_PORT, "port"));
     char* path = nullptr;
     code = curl_url_get(url, CURLUPART_PATH, &path, 0);
     if (code != CURLUE_OK) {
-        return Status::Invalid(
-            fmt::format("invalid S3 endpoint {} path: code {}", endpoint, static_cast<int>(code)));
+        return Status::Invalid(fmt::format("invalid S3 endpoint {} path: code {}", endpoint,
+                                           static_cast<int32_t>(code)));
     }
     ScopeGuard free_path([path] { curl_free(path); });
-    std::string authority = *host;
-    if (authority.find(':') != std::string::npos) {
-        authority = "[" + authority + "]";
-    }
-    if (port) {
-        authority += ":" + *port;
+    char* normalized_url = nullptr;
+    code = curl_url_get(url, CURLUPART_URL, &normalized_url, 0);
+    if (code != CURLUE_OK) {
+        return Status::Invalid(fmt::format("invalid S3 endpoint {} URL: code {}", endpoint,
+                                           static_cast<int32_t>(code)));
+    }
+    ScopeGuard free_normalized_url([normalized_url] { curl_free(normalized_url); });
+    std::string_view normalized_url_view(normalized_url);
+    size_t scheme_end = normalized_url_view.find("://");
+    size_t authority_start =
+        scheme_end == std::string_view::npos ? std::string_view::npos : scheme_end + 3;
+    size_t authority_end = normalized_url_view.find('/', authority_start);
+    if (authority_start == std::string_view::npos) {
+        return Status::Invalid(fmt::format("invalid S3 endpoint {}", endpoint));
     }
+    std::string authority(
+        normalized_url_view.substr(authority_start, authority_end - authority_start));
     return Endpoint{std::move(normalized_scheme), std::move(authority), path};
 }
 
@@ -547,7 +610,7 @@ bool IsVirtualHostableS3Bucket(const std::string& bucket, bool allow_subdomains)
     }
     bool label_start = true;
     for (size_t index = 0; index < bucket.size(); ++index) {
-        const auto character = static_cast<unsigned char>(bucket[index]);
+        const auto character = static_cast<uint8_t>(bucket[index]);
         if (std::islower(character) || std::isdigit(character)) {
             label_start = false;
             continue;
@@ -580,15 +643,15 @@ bool IsIpAddressAuthority(const std::string& authority) {
         host = host.substr(0, port_separator);
     }
     size_t component_start = 0;
-    int component_count = 0;
+    int32_t component_count = 0;
     while (component_start < host.size()) {
         size_t component_end = host.find('.', component_start);
         std::string_view component = host.substr(component_start, component_end - component_start);
         if (component.empty() || component.size() > 3) {
             return false;
         }
-        int value = 0;
-        for (unsigned char character : component) {
+        int32_t value = 0;
+        for (uint8_t character : component) {
             if (!std::isdigit(character)) {
                 return false;
             }
@@ -636,11 +699,11 @@ struct SigningContext {
     aws_http_message* message;
     std::mutex mutex;
     std::condition_variable condition;
-    int error_code = AWS_ERROR_SUCCESS;
+    int32_t error_code = AWS_ERROR_SUCCESS;
     bool complete = false;
 };
 
-void OnSigningComplete(aws_signing_result* result, int error_code, void* user_data) {
+void OnSigningComplete(aws_signing_result* result, int32_t error_code, void* user_data) {
     auto* context = static_cast<SigningContext*>(user_data);
     if (error_code == AWS_ERROR_SUCCESS &&
         aws_apply_signing_result_to_http_request(context->message, context->allocator, result)) {
@@ -654,12 +717,18 @@ void OnSigningComplete(aws_signing_result* result, int error_code, void* user_da
     context->condition.notify_one();
 }
 
+std::shared_ptr<Executor> GetS3Executor() {
+    static std::shared_ptr<Executor> executor = CreateDefaultExecutor();
+    return executor;
+}
+
 class S3ObjectStoreClient : public ObjectStoreClient,
                             public std::enable_shared_from_this<S3ObjectStoreClient> {
  public:
     static Result<std::shared_ptr<ObjectStoreClient>> Create(
         const std::map<std::string, std::string>& options, std::shared_ptr<HttpClient> http_client,
-        std::shared_ptr<aws_credentials_provider> credentials, std::unique_ptr<Executor> executor) {
+        std::shared_ptr<aws_credentials_provider> credentials,
+        const std::shared_ptr<Executor>& executor) {
         PAIMON_ASSIGN_OR_RAISE(std::string region, ResolveRegion(options));
         auto endpoint = options.find(kS3EndpointOption);
         bool use_default_endpoint = endpoint == options.end() || endpoint->second.empty();
@@ -668,10 +737,8 @@ class S3ObjectStoreClient : public ObjectStoreClient,
             ParseEndpoint(use_default_endpoint ? fmt::format("https://s3.{}.{}", region,
                                                              AwsDnsSuffixForRegion(region))
                                                : endpoint->second));
-        auto path_style = options.find(kS3PathStyleAccessOption);
-        bool use_path_style =
-            path_style != options.end() &&
-            OptionsUtils::GetValueFromMap<bool>(options, kS3PathStyleAccessOption).value();
+        PAIMON_ASSIGN_OR_RAISE(bool use_path_style, OptionsUtils::GetValueFromMap<bool>(
+                                                        options, kS3PathStyleAccessOption, false));
         return std::shared_ptr<ObjectStoreClient>(new S3ObjectStoreClient(
             std::move(http_client), std::move(credentials), std::move(executor),
             std::move(parsed_endpoint), std::move(region), use_path_style, use_default_endpoint));
@@ -692,7 +759,7 @@ class S3ObjectStoreClient : public ObjectStoreClient,
         int64_t modification_time = 0;
         auto modified = response.headers.find("last-modified");
         if (modified != response.headers.end()) {
-            modification_time = ParseModificationTime(modified->second);
+            modification_time = ParseHttpModificationTime(modified->second);
         }
         PAIMON_ASSIGN_OR_RAISE(int64_t object_size,
                                ParseNonNegativeInt64(length->second, "Content-Length"));
@@ -745,7 +812,7 @@ class S3ObjectStoreClient : public ObjectStoreClient,
             int64_t modified = 0;
             auto last_modified = TagValue(block, "LastModified");
             if (last_modified) {
-                modified = ParseModificationTime(*last_modified);
+                modified = ParseIso8601ModificationTime(*last_modified);
             }
             result.objects.push_back(ObjectMetadata{decoded_key, object_size, modified});
         }
@@ -803,9 +870,14 @@ class S3ObjectStoreClient : public ObjectStoreClient,
 
     void GetObjectRangeAsync(const ObjectStorePath& path, int64_t offset, int64_t size,
                              char* buffer, std::function<void(Status)>&& callback) const override {
+        std::shared_ptr<Executor> executor = executor_.lock();
+        if (!executor) {
+            callback(Status::IOError("S3 executor is unavailable"));
+            return;
+        }
         auto self = shared_from_this();
-        executor_->Add([self = std::move(self), path, offset, size, buffer,
-                        callback = std::move(callback)]() mutable {
+        executor->Add([self = std::move(self), path, offset, size, buffer,
+                       callback = std::move(callback)]() mutable {
             Result<int64_t> result = self->GetObjectRange(path, offset, size, buffer);
             callback(result.ok() ? Status::OK() : result.status());
         });
@@ -814,8 +886,8 @@ class S3ObjectStoreClient : public ObjectStoreClient,
  private:
     S3ObjectStoreClient(std::shared_ptr<HttpClient> http_client,
                         std::shared_ptr<aws_credentials_provider> credentials,
-                        std::unique_ptr<Executor> executor, Endpoint endpoint, std::string region,
-                        bool path_style, bool use_default_endpoint)
+                        const std::shared_ptr<Executor>& executor, Endpoint endpoint,
+                        std::string region, bool path_style, bool use_default_endpoint)
         : http_client_(std::move(http_client)),
           credentials_(std::move(credentials)),
           endpoint_(std::move(endpoint)),
@@ -823,13 +895,27 @@ class S3ObjectStoreClient : public ObjectStoreClient,
           path_style_(path_style),
           use_default_endpoint_(use_default_endpoint),
           executor_(std::move(executor)) {}
+
     Status CheckResponse(const HttpResponse& response, const std::string& operation,
                          const ObjectStorePath& path) const {
         if (response.status_code >= 200 && response.status_code < 300) {
             return Status::OK();
         }
-        return Status::IOError(fmt::format("{} failed for s3://{}/{}: HTTP {}", operation,
-                                           path.bucket, path.key, response.status_code));
+        std::string detail;
+        auto error_code = TagValue(response.error_body, "Code");
+        auto error_message = TagValue(response.error_body, "Message");
+        if (error_code) {
+            detail = ", S3 error " + *error_code;
+        }
+        if (error_message) {
+            detail += error_code ? ": " + *error_message : ", S3 message " + *error_message;
+        }
+        auto bucket_region = response.headers.find("x-amz-bucket-region");
+        if (bucket_region != response.headers.end()) {
+            detail += ", bucket region " + bucket_region->second;
+        }
+        return Status::IOError(fmt::format("{} failed for s3://{}/{}: HTTP {}{}", operation,
+                                           path.bucket, path.key, response.status_code, detail));
     }
 
     Result<HttpResponse> Execute(const ObjectStorePath& object, HttpMethod method,
@@ -844,8 +930,8 @@ class S3ObjectStoreClient : public ObjectStoreClient,
             path_style_ ||
             (use_default_endpoint_
                  ? !IsVirtualHostableS3Bucket(object.bucket, false)
-                 : endpoint_.scheme != "http" || IsIpAddressAuthority(endpoint_.authority) ||
-                       !IsVirtualHostableS3Bucket(object.bucket, true));
+                 : IsIpAddressAuthority(endpoint_.authority) ||
+                       !IsVirtualHostableS3Bucket(object.bucket, endpoint_.scheme == "http"));
         if (use_path_style) {
             PAIMON_ASSIGN_OR_RAISE(std::string encoded_bucket, PercentEncode(object.bucket, false));
             request_path += encoded_bucket + "/";
@@ -896,15 +982,15 @@ class S3ObjectStoreClient : public ObjectStoreClient,
         config.credentials_provider = credentials_.get();
 
         SigningContext context(runtime->allocator(), message);
-        int result = aws_sign_request_aws(runtime->allocator(), signable,
-                                          reinterpret_cast<aws_signing_config_base*>(&config),
-                                          OnSigningComplete, &context);
+        int32_t result = aws_sign_request_aws(runtime->allocator(), signable,
+                                              reinterpret_cast<aws_signing_config_base*>(&config),
+                                              OnSigningComplete, &context);
         if (result == AWS_OP_SUCCESS) {
             std::unique_lock<std::mutex> lock(context.mutex);
             context.condition.wait(lock, [&context] { return context.complete; });
         }
         if (result != AWS_OP_SUCCESS || context.error_code != AWS_ERROR_SUCCESS) {
-            int error = result == AWS_OP_SUCCESS ? context.error_code : aws_last_error();
+            int32_t error = result == AWS_OP_SUCCESS ? context.error_code : aws_last_error();
             return Status::IOError(
                 fmt::format("failed to sign S3 request: {}", aws_error_debug_str(error)));
         }
@@ -933,7 +1019,7 @@ class S3ObjectStoreClient : public ObjectStoreClient,
     std::string region_;
     bool path_style_ = false;
     bool use_default_endpoint_ = false;
-    std::unique_ptr<Executor> executor_;
+    std::weak_ptr<Executor> executor_;
 };
 
 }  // namespace
@@ -959,15 +1045,16 @@ Status ValidateS3Options(const std::map<std::string, std::string>& options) {
     if (has_secret && secret->second.empty()) {
         return Status::Invalid(fmt::format("{} must not be empty", kS3SecretKeyOption));
     }
-    if (options.find(kS3PathStyleAccessOption) == options.end()) {
-        return Status::OK();
-    }
-    Result<bool> parsed = OptionsUtils::GetValueFromMap<bool>(options, kS3PathStyleAccessOption);
-    if (!parsed.ok()) {
-        return Status::Invalid(
-            fmt::format("{} {}", kS3PathStyleAccessOption, parsed.status().message()));
+    if (options.find(kS3PathStyleAccessOption) != options.end()) {
+        Result<bool> parsed =
+            OptionsUtils::GetValueFromMap<bool>(options, kS3PathStyleAccessOption);
+        if (!parsed.ok()) {
+            return Status::Invalid(
+                fmt::format("{} {}", kS3PathStyleAccessOption, parsed.status().message()));
+        }
     }
-    return Status::OK();
+    Result<CurlHttpClientOptions> http_options = ResolveCurlHttpClientOptions(options);
+    return http_options.ok() ? Status::OK() : http_options.status();
 }
 
 S3FileSystem::S3FileSystem(std::shared_ptr<ObjectStoreClient> client)
@@ -975,8 +1062,13 @@ S3FileSystem::S3FileSystem(std::shared_ptr<ObjectStoreClient> client)
 
 Result<std::unique_ptr<FileSystem>> S3FileSystem::Create(
     const std::map<std::string, std::string>& options) {
-    PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<ObjectStoreClient> client,
-                           MakeS3ObjectStoreClient(options, std::make_shared<CurlHttpClient>()));
+    std::map<std::string, std::string> normalized_options = NormalizeS3Options(options);
+    PAIMON_ASSIGN_OR_RAISE(CurlHttpClientOptions http_options,
+                           ResolveCurlHttpClientOptions(normalized_options));
+    PAIMON_ASSIGN_OR_RAISE(
+        std::shared_ptr<ObjectStoreClient> client,
+        MakeS3ObjectStoreClient(normalized_options,
+                                std::make_shared<CurlHttpClient>(std::move(http_options))));
     return std::unique_ptr<FileSystem>(new S3FileSystem(std::move(client)));
 }
 
@@ -989,9 +1081,8 @@ Result<std::shared_ptr<ObjectStoreClient>> MakeS3ObjectStoreClient(
     if (!credentials) {
         return Status::IOError("failed to initialize S3 credentials provider");
     }
-    std::unique_ptr<Executor> executor = CreateDefaultExecutor();
     return S3ObjectStoreClient::Create(normalized_options, std::move(http_client),
-                                       std::move(credentials), std::move(executor));
+                                       std::move(credentials), GetS3Executor());
 }
 
 }  // namespace paimon::s3
diff --git a/src/paimon/fs/s3/s3_file_system.h b/src/paimon/fs/s3/s3_file_system.h
index 66c796f89..55107b902 100644
--- a/src/paimon/fs/s3/s3_file_system.h
+++ b/src/paimon/fs/s3/s3_file_system.h
@@ -32,6 +32,10 @@ inline constexpr char kS3ProfileOption[] = "s3.profile";
 inline constexpr char kS3AccessKeyOption[] = "s3.access-key";
 inline constexpr char kS3SecretKeyOption[] = "s3.secret-key";
 inline constexpr char kS3SessionTokenOption[] = "s3.session.token";
+inline constexpr char kS3ConnectionEstablishTimeoutOption[] = "s3.connection.establish.timeout";
+inline constexpr char kS3ConnectionRequestTimeoutOption[] = "s3.connection.request.timeout";
+inline constexpr char kS3LowSpeedLimitOption[] = "s3.low-speed-limit";
+inline constexpr char kS3LowSpeedTimeSecondsOption[] = "s3.low-speed-time-seconds";
 
 Status ValidateS3Options(const std::map<std::string, std::string>& options);
 Result<std::shared_ptr<ObjectStoreClient>> MakeS3ObjectStoreClient(
diff --git a/src/paimon/fs/s3/s3_file_system_test.cpp b/src/paimon/fs/s3/s3_file_system_test.cpp
index cab0c7ae9..828c69672 100644
--- a/src/paimon/fs/s3/s3_file_system_test.cpp
+++ b/src/paimon/fs/s3/s3_file_system_test.cpp
@@ -18,11 +18,16 @@
 
 #include <gtest/gtest.h>
 
+#include <chrono>
+#include <condition_variable>
 #include <cstdlib>
 #include <cstring>
 #include <filesystem>
 #include <fstream>
+#include <future>
+#include <mutex>
 #include <optional>
+#include <thread>
 #include <utility>
 #include <vector>
 
@@ -30,7 +35,7 @@
 #include "paimon/fs/s3/s3_file_system_factory.h"
 #include "paimon/testing/utils/testharness.h"
 
-namespace paimon::s3 {
+namespace paimon::s3::test {
 namespace {
 
 class MockHttpClient : public HttpClient {
@@ -42,7 +47,11 @@ class MockHttpClient : public HttpClient {
         response.status_code = status_code_;
         response.headers = response_headers_;
         if (!body_.empty()) {
-            PAIMON_RETURN_NOT_OK(consumer(body_.data(), body_.size()));
+            if (status_code_ >= 300) {
+                response.error_body = body_;
+            } else {
+                PAIMON_RETURN_NOT_OK(consumer(body_.data(), body_.size()));
+            }
             response.body_size = body_.size();
         }
         return response;
@@ -54,6 +63,39 @@ class MockHttpClient : public HttpClient {
     std::string body_;
 };
 
+class BlockingMockHttpClient : public MockHttpClient {
+ public:
+    Result<HttpResponse> Execute(const HttpRequest& request,
+                                 const HttpBodyConsumer& consumer) const override {
+        if (request.method == HttpMethod::GET) {
+            std::unique_lock<std::mutex> lock(mutex_);
+            get_started_ = true;
+            condition_.notify_all();
+            condition_.wait(lock, [this] { return release_get_; });
+        }
+        return MockHttpClient::Execute(request, consumer);
+    }
+
+    bool WaitForGetStart() const {
+        std::unique_lock<std::mutex> lock(mutex_);
+        return condition_.wait_for(lock, std::chrono::seconds(5), [this] { return get_started_; });
+    }
+
+    void ReleaseGet() {
+        {
+            std::scoped_lock lock(mutex_);
+            release_get_ = true;
+        }
+        condition_.notify_all();
+    }
+
+ private:
+    mutable std::mutex mutex_;
+    mutable std::condition_variable condition_;
+    mutable bool get_started_ = false;
+    bool release_get_ = false;
+};
+
 class ScopedEnvironmentVariable {
  public:
     ScopedEnvironmentVariable(const char* name, std::optional<std::string> value) : name_(name) {
@@ -145,7 +187,7 @@ TEST(S3ObjectStoreClientTest, TestCustomEndpointAddressing) {
     ASSERT_OK_AND_ASSIGN(std::shared_ptr<ObjectStoreClient> https_client,
                          MakeS3ObjectStoreClient(https_options, http));
     ASSERT_OK(https_client->HeadObject({"bucket", "file"}));
-    ASSERT_EQ(http->request_.url, "https://s3.example.com/bucket/file");
+    ASSERT_EQ(http->request_.url, "https://bucket.s3.example.com/file");
     ASSERT_OK(https_client->HeadObject({"paimon.prod.data", "file"}));
     ASSERT_EQ(http->request_.url, "https://s3.example.com/paimon.prod.data/file");
 
@@ -163,6 +205,22 @@ TEST(S3ObjectStoreClientTest, TestCustomEndpointAddressing) {
     ASSERT_OK(ip_client->HeadObject({"bucket", "file"}));
     ASSERT_EQ(http->request_.url, "http://127.0.0.1:9000/bucket/file");
 
+    auto ipv6_options = StaticOptions();
+    ipv6_options[kS3EndpointOption] = "http://[::1]:9000";
+    ASSERT_OK_AND_ASSIGN(std::shared_ptr<ObjectStoreClient> ipv6_client,
+                         MakeS3ObjectStoreClient(ipv6_options, http));
+    ASSERT_OK(ipv6_client->HeadObject({"bucket", "file"}));
+    ASSERT_EQ(http->request_.url, "http://[::1]:9000/bucket/file");
+    ASSERT_EQ(http->request_.headers["host"], "[::1]:9000");
+
+    auto ipv6_zone_options = StaticOptions();
+    ipv6_zone_options[kS3EndpointOption] = "http://[fe80::1%25eth0]:9000";
+    ASSERT_OK_AND_ASSIGN(std::shared_ptr<ObjectStoreClient> ipv6_zone_client,
+                         MakeS3ObjectStoreClient(ipv6_zone_options, http));
+    ASSERT_OK(ipv6_zone_client->HeadObject({"bucket", "file"}));
+    ASSERT_EQ(http->request_.url, "http://[fe80::1%25eth0]:9000/bucket/file");
+    ASSERT_EQ(http->request_.headers["host"], "[fe80::1%25eth0]:9000");
+
     auto base_path_options = StaticOptions();
     base_path_options[kS3EndpointOption] = "HTTPS://s3.example.com/storage";
     base_path_options[kS3PathStyleAccessOption] = "true";
@@ -296,6 +354,53 @@ TEST(S3ObjectStoreClientTest, TestInvalidModificationTime) {
         "<LastModified>invalid</LastModified><Size>12</Size></Contents></ListBucketResult>";
     ASSERT_OK_AND_ASSIGN(auto result, client->ListObjects({"bucket", ""}, "", 0));
     ASSERT_EQ(result.objects[0].modification_time, 0);
+
+    http->body_ =
+        "<ListBucketResult><IsTruncated>false</IsTruncated><Contents><Key>file</Key>"
+        "<LastModified>2026-02-30T00:00:00.000Z</LastModified>"
+        "<Size>12</Size></Contents></ListBucketResult>";
+    ASSERT_OK_AND_ASSIGN(result, client->ListObjects({"bucket", ""}, "", 0));
+    ASSERT_EQ(result.objects[0].modification_time, 0);
+}
+
+TEST(S3ObjectStoreClientTest, TestModificationTimeFormats) {
+    auto http = std::make_shared<MockHttpClient>();
+    http->response_headers_["content-length"] = "12";
+    http->response_headers_["last-modified"] = "Mon, 12 Oct 2009 17:50:30 GMT";
+    ASSERT_OK_AND_ASSIGN(std::shared_ptr<ObjectStoreClient> client,
+                         MakeS3ObjectStoreClient(StaticOptions(), http));
+    ASSERT_OK_AND_ASSIGN(ObjectMetadata head, client->HeadObject({"bucket", "file"}));
+    ASSERT_EQ(head.modification_time, 1255369830000);
+
+    http->body_ =
+        "<ListBucketResult><IsTruncated>false</IsTruncated><Contents><Key>file</Key>"
+        "<LastModified>2009-10-12T17:50:30.123Z</LastModified>"
+        "<Size>12</Size></Contents></ListBucketResult>";
+    ASSERT_OK_AND_ASSIGN(ListObjectsResult list, client->ListObjects({"bucket", ""}, "", 0));
+    ASSERT_EQ(list.objects[0].modification_time, 1255369830123);
+}
+
+TEST(S3ObjectStoreClientTest, TestNotFoundAndErrorDetails) {
+    auto http = std::make_shared<MockHttpClient>();
+    http->status_code_ = 404;
+    ASSERT_OK_AND_ASSIGN(std::shared_ptr<ObjectStoreClient> client,
+                         MakeS3ObjectStoreClient(StaticOptions(), http));
+    ASSERT_TRUE(client->HeadObject({"bucket", "missing"}).status().IsNotExist());
+    char buffer[1];
+    ASSERT_TRUE(client->GetObjectRange({"bucket", "missing"}, 0, 1, buffer).status().IsNotExist());
+    ASSERT_TRUE(client->ListObjects({"missing", ""}, "", 0).status().IsNotExist());
+
+    http->status_code_ = 403;
+    http->body_ = "<Error><Code>AccessDenied</Code><Message>access is denied</Message></Error>";
+    ASSERT_NOK_WITH_MSG(client->GetObjectRange({"bucket", "file"}, 0, 1, buffer),
+                        "S3 error AccessDenied");
+    ASSERT_NOK_WITH_MSG(client->GetObjectRange({"bucket", "file"}, 0, 1, buffer),
+                        "access is denied");
+
+    http->status_code_ = 301;
+    http->body_.clear();
+    http->response_headers_["x-amz-bucket-region"] = "eu-west-1";
+    ASSERT_NOK_WITH_MSG(client->HeadObject({"bucket", "file"}), "bucket region eu-west-1");
 }
 
 TEST(S3ObjectStoreClientTest, TestRegionFromEnvironment) {
@@ -425,6 +530,7 @@ TEST(S3ObjectStoreClientTest, TestRangeAndListObjects) {
     ASSERT_TRUE(result.is_truncated);
     ASSERT_EQ(result.continuation_token, "next token");
     ASSERT_EQ(result.objects[0].key, "dir/a&b");
+    ASSERT_EQ(result.objects[0].modification_time, 1767225600000);
     ASSERT_EQ(result.objects[1].key, "dir/a&lt;b");
     ASSERT_EQ(result.common_prefixes[0], "dir/sub/");
     ASSERT_NE(http->request_.url.find("amazonaws.com/?list-type=2"), std::string::npos);
@@ -432,6 +538,37 @@ TEST(S3ObjectStoreClientTest, TestRangeAndListObjects) {
     ASSERT_NE(http->request_.url.find("continuation-token=old%20token"), std::string::npos);
 }
 
+TEST(S3ObjectStoreClientTest, TestAsyncReadCanReleaseStreamAndClient) {
+    auto http = std::make_shared<BlockingMockHttpClient>();
+    http->response_headers_["content-length"] = "4";
+    ASSERT_OK_AND_ASSIGN(std::shared_ptr<ObjectStoreClient> client,
+                         MakeS3ObjectStoreClient(StaticOptions(), http));
+    std::weak_ptr<ObjectStoreClient> weak_client = client;
+    auto file_system = std::make_unique<ObjectStoreFileSystem>("s3", client);
+    ASSERT_OK_AND_ASSIGN(std::unique_ptr<InputStream> stream, file_system->Open("s3://bucket/key"));
+    http->body_ = "data";
+
+    char buffer[4];
+    std::promise<Status> completion;
+    std::future<Status> completed = completion.get_future();
+    stream->ReadAsync(buffer, sizeof(buffer), 0,
+                      [&completion](Status status) { completion.set_value(std::move(status)); });
+    bool get_started = http->WaitForGetStart();
+    stream.reset();
+    file_system.reset();
+    client.reset();
+    http->ReleaseGet();
+
+    ASSERT_TRUE(get_started);
+    ASSERT_EQ(completed.wait_for(std::chrono::seconds(5)), std::future_status::ready);
+    ASSERT_OK(completed.get());
+    ASSERT_EQ(std::string(buffer, sizeof(buffer)), "data");
+    for (int32_t attempt = 0; attempt < 100 && !weak_client.expired(); ++attempt) {
+        std::this_thread::sleep_for(std::chrono::milliseconds(10));
+    }
+    ASSERT_TRUE(weak_client.expired());
+}
+
 TEST(S3ObjectStoreClientTest, TestUrlEncodedListObjects) {
     auto http = std::make_shared<MockHttpClient>();
     ASSERT_OK_AND_ASSIGN(std::shared_ptr<ObjectStoreClient> client,
@@ -494,7 +631,21 @@ TEST(S3FileSystemFactoryTest, TestOptionValidation) {
         MakeS3ObjectStoreClient({{kS3AccessKeyOption, "access"}}, http).status().IsInvalid());
     ASSERT_TRUE(
         MakeS3ObjectStoreClient({{kS3PathStyleAccessOption, "treu"}}, http).status().IsInvalid());
+    ASSERT_TRUE(MakeS3ObjectStoreClient({{kS3ConnectionEstablishTimeoutOption, "-1"}}, http)
+                    .status()
+                    .IsInvalid());
+    ASSERT_TRUE(MakeS3ObjectStoreClient({{kS3ConnectionRequestTimeoutOption, "-1"}}, http)
+                    .status()
+                    .IsInvalid());
+    ASSERT_TRUE(
+        MakeS3ObjectStoreClient({{kS3LowSpeedLimitOption, "invalid"}}, http).status().IsInvalid());
+    ASSERT_TRUE(MakeS3ObjectStoreClient({{"s3a.connection.establish.timeout", "-1"}}, http)
+                    .status()
+                    .IsInvalid());
+    ASSERT_TRUE(MakeS3ObjectStoreClient({{"fs.s3a.connection.request.timeout", "-1"}}, http)
+                    .status()
+                    .IsInvalid());
 }
 
 }  // namespace
-}  // namespace paimon::s3
+}  // namespace paimon::s3::test