diff --git a/CMakeLists.txt b/CMakeLists.txt
index 3c8da72e0..876ceac96 100644
--- a/CMakeLists.txt
+++ b/CMakeLists.txt
@@ -57,6 +57,8 @@ option(PAIMON_ENABLE_AVRO "Whether to enable avro file format" ON)
 option(PAIMON_ENABLE_ORC "Whether to enable orc file format" ON)
 option(PAIMON_ENABLE_LANCE "Whether to enable lance file format" OFF)
 option(PAIMON_ENABLE_JINDO "Whether to enable jindo file system" OFF)
+option(PAIMON_ENABLE_NETWORK_TESTS
+       "Whether to enable tests that access real remote services over the network" OFF)
 option(PAIMON_ENABLE_LUMINA "Whether to enable lumina vector index" OFF)
 option(PAIMON_ENABLE_LUCENE "Whether to enable lucene index" OFF)
 option(PAIMON_ENABLE_TANTIVY
@@ -70,6 +72,11 @@ endif()
 if(PAIMON_ENABLE_JINDO)
     add_definitions(-DPAIMON_ENABLE_JINDO)
 endif()
+if(PAIMON_ENABLE_NETWORK_TESTS)
+    if(NOT PAIMON_BUILD_TESTS)
+        message(FATAL_ERROR "PAIMON_ENABLE_NETWORK_TESTS requires PAIMON_BUILD_TESTS=ON")
+    endif()
+endif()
 if(PAIMON_USE_CXX11_ABI)
     add_definitions(-D_GLIBCXX_USE_CXX11_ABI=1)
 else()
diff --git a/cmake_modules/DefineOptions.cmake b/cmake_modules/DefineOptions.cmake
index cf7f964be..7c500415c 100644
--- a/cmake_modules/DefineOptions.cmake
+++ b/cmake_modules/DefineOptions.cmake
@@ -107,6 +107,9 @@ if("${CMAKE_SOURCE_DIR}" STREQUAL "${CMAKE_CURRENT_SOURCE_DIR}")
 
     define_option(PAIMON_BUILD_TESTS "Build the Paimon googletest unit tests" OFF)
 
+    define_option(PAIMON_ENABLE_NETWORK_TESTS
+                  "Enable tests that access real remote services over the network" OFF)
+
     define_option(PAIMON_BUILD_BENCHMARKS
                   "Build the Paimon Google Benchmark performance benchmarks" OFF)
 
diff --git a/include/paimon/fs/file_system.h b/include/paimon/fs/file_system.h
index 55207a63f..b8133156d 100644
--- a/include/paimon/fs/file_system.h
+++ b/include/paimon/fs/file_system.h
@@ -98,6 +98,7 @@ class PAIMON_EXPORT InputStream : public Stream {
     /// @param callback The callback function to be invoked upon completion of the read operation.
     ///                 The callback will receive a Status object indicating the success or failure
     ///                 of the read operation.
+    /// @note The caller must keep the input stream and buffer alive until the callback is invoked.
     virtual void ReadAsync(char* buffer, int64_t size, int64_t offset,
                            std::function<void(Status)>&& callback) = 0;
 
diff --git a/src/paimon/CMakeLists.txt b/src/paimon/CMakeLists.txt
index 46cc13854..58f4549f4 100644
--- a/src/paimon/CMakeLists.txt
+++ b/src/paimon/CMakeLists.txt
@@ -791,12 +791,17 @@ if(PAIMON_BUILD_TESTS)
                     ${TEST_STATIC_LINK_LIBS}
                     ${GTEST_LINK_TOOLCHAIN})
 
-    # jindo_utils_test only checks the in-memory status conversion and does not need an
-    # OSS cluster, so it runs whenever jindo is built. The other jindo tests need real
-    # OSS access and stay disabled.
-    set(FS_TEST_JINDO_SOURCES)
+    set(PAIMON_JINDO_FS_TEST_SOURCES)
+    set(PAIMON_JINDO_FS_TEST_LINK_LIBS)
     if(PAIMON_ENABLE_JINDO)
-        list(APPEND FS_TEST_JINDO_SOURCES fs/jindo/jindo_utils_test.cpp)
+        list(APPEND PAIMON_JINDO_FS_TEST_SOURCES fs/jindo/jindo_utils_test.cpp)
+        if(PAIMON_ENABLE_NETWORK_TESTS)
+            list(APPEND PAIMON_JINDO_FS_TEST_SOURCES
+                 fs/jindo/jindo_file_system_factory_test.cpp
+                 fs/jindo/jindo_file_system_test.cpp)
+        endif()
+        list(APPEND PAIMON_JINDO_FS_TEST_LINK_LIBS
+             ${PAIMON_JINDO_FILE_SYSTEM_STATIC_LINK_LIBS})
     endif()
 
     add_paimon_test(fs_test
@@ -804,16 +809,18 @@ if(PAIMON_BUILD_TESTS)
                     common/fs/file_system_test.cpp
                     common/fs/resolving_file_system_test.cpp
                     fs/local/local_file_test.cpp
-                    ${FS_TEST_JINDO_SOURCES}
-                    # fs/jindo/jindo_file_system_factory_test.cpp
-                    # fs/jindo/jindo_file_system_test.cpp
+                    ${PAIMON_JINDO_FS_TEST_SOURCES}
                     STATIC_LINK_LIBS
                     paimon_shared
                     ${PAIMON_LOCAL_FILE_SYSTEM_STATIC_LINK_LIBS}
-                    ${PAIMON_JINDO_FILE_SYSTEM_STATIC_LINK_LIBS}
+                    ${PAIMON_JINDO_FS_TEST_LINK_LIBS}
                     test_utils_static
                     ${GTEST_LINK_TOOLCHAIN}
                     EXTRA_INCLUDES
                     ${JINDOSDK_INCLUDE_DIR})
 
+    if(PAIMON_ENABLE_NETWORK_TESTS)
+        target_compile_definitions(paimon-fs-test PRIVATE PAIMON_ENABLE_NETWORK_TESTS)
+    endif()
+
 endif()
diff --git a/src/paimon/common/fs/file_system_test.cpp b/src/paimon/common/fs/file_system_test.cpp
index d7599d513..176159275 100644
--- a/src/paimon/common/fs/file_system_test.cpp
+++ b/src/paimon/common/fs/file_system_test.cpp
@@ -1474,7 +1474,9 @@ TEST_P(FileSystemTest, TestAtomicStoreAlreadyExist) {
 std::vector<std::string> GetTestValuesForFileSystemTest() {
     std::vector<std::string> values;
     values.emplace_back("local");
-    // values.emplace_back("jindo");
+#if defined(PAIMON_ENABLE_NETWORK_TESTS) && defined(PAIMON_ENABLE_JINDO)
+    values.emplace_back("jindo");
+#endif
     return values;
 }
 
diff --git a/src/paimon/core/operation/key_value_file_store_write_test.cpp b/src/paimon/core/operation/key_value_file_store_write_test.cpp
index 2fa199af1..4b6b39486 100644
--- a/src/paimon/core/operation/key_value_file_store_write_test.cpp
+++ b/src/paimon/core/operation/key_value_file_store_write_test.cpp
@@ -320,7 +320,6 @@ TEST_F(KeyValueFileStoreWriteTest, TestSharedShreddingMapRestoreInitializesNextW
         {"fields.tags.map.shared-shredding.column-placement-policy", "plain"},
         {"write-only", "true"},
         {"bucket", "1"},
-        {"enable-pk-commit-in-inte-test", ""},
     };
     auto logical_schema = arrow::schema({
         arrow::field("id", arrow::int32(), /*nullable=*/false),
diff --git a/src/paimon/fs/jindo/jindo_file_system.cpp b/src/paimon/fs/jindo/jindo_file_system.cpp
index 6246780dd..c689642d1 100644
--- a/src/paimon/fs/jindo/jindo_file_system.cpp
+++ b/src/paimon/fs/jindo/jindo_file_system.cpp
@@ -16,7 +16,9 @@
 
 #include "paimon/fs/jindo/jindo_file_system.h"
 
+#include <atomic>
 #include <cassert>
+#include <string_view>
 #include <utility>
 
 #include "JdoFileInfo.hpp"    // NOLINT(build/include_subdir)
@@ -52,6 +54,29 @@ class JindoFileSystemImpl {
     std::unique_ptr<JdoFileSystem> fs_;
 };
 
+namespace {
+
+class AsyncReadState {
+ public:
+    explicit AsyncReadState(std::function<void(Status)>&& callback)
+        : callback_(std::move(callback)) {}
+
+    void Complete(JdoStatus status) {
+        if (completed_.exchange(true)) {
+            return;
+        }
+        callback_(status.ok() ? Status::OK() : Status::IOError(status.errMsg()));
+    }
+
+    std::string_view result;
+
+ private:
+    std::atomic<bool> completed_{false};
+    std::function<void(Status)> callback_;
+};
+
+}  // namespace
+
 JindoFileSystem::JindoFileSystem(std::unique_ptr<JdoFileSystem>&& fs)
     : impl_(std::make_shared<JindoFileSystemImpl>(std::move(fs))) {}
 
@@ -215,15 +240,17 @@ Result<int64_t> JindoInputStream::Length() const {
 
 Result<int64_t> JindoInputStream::Read(char* buffer, int64_t size) {
     PAIMON_RETURN_NOT_OK(ValidateValueNonNegative(size, "read length"));
-    PAIMON_RETURN_NOT_OK_FROM_JINDO(reader_->read(size, &result_, buffer));
-    return result_.length();
+    std::string_view result;
+    PAIMON_RETURN_NOT_OK_FROM_JINDO(reader_->read(size, &result, buffer));
+    return result.length();
 }
 
 Result<int64_t> JindoInputStream::Read(char* buffer, int64_t size, int64_t offset) {
     PAIMON_RETURN_NOT_OK(ValidateValueNonNegative(size, "read length"));
     PAIMON_RETURN_NOT_OK(ValidateValueNonNegative(offset, "read offset"));
-    PAIMON_RETURN_NOT_OK_FROM_JINDO(reader_->pread(offset, size, &result_, buffer));
-    return result_.length();
+    std::string_view result;
+    PAIMON_RETURN_NOT_OK_FROM_JINDO(reader_->pread(offset, size, &result, buffer));
+    return result.length();
 }
 
 void JindoInputStream::ReadAsync(char* buffer, int64_t size, int64_t offset,
@@ -238,12 +265,17 @@ void JindoInputStream::ReadAsync(char* buffer, int64_t size, int64_t offset,
         callback(validate_status);
         return;
     }
-    auto outer_callback = [=](JdoStatus status) {
-        callback(status.ok() ? Status::OK() : Status::IOError(status.errMsg()));
-    };
-    auto task = reader_->preadAsync(offset, size, &result_, buffer, outer_callback);
+    std::shared_ptr<AsyncReadState> state = std::make_shared<AsyncReadState>(std::move(callback));
+    auto task = reader_->preadAsync(offset, size, &state->result, buffer,
+                                    [state](JdoStatus status) { state->Complete(status); });
     assert(task);
-    [[maybe_unused]] auto perform_status = task->perform();
+
+    auto perform_status = task->perform();
+    if (!perform_status.ok()) {
+        state->Complete(perform_status);
+        [[maybe_unused]] auto status = task->cancel();
+        return;
+    }
 }
 
 Status JindoInputStream::Close() {
diff --git a/src/paimon/fs/jindo/jindo_file_system.h b/src/paimon/fs/jindo/jindo_file_system.h
index 3b413b3c0..4e3c475a0 100644
--- a/src/paimon/fs/jindo/jindo_file_system.h
+++ b/src/paimon/fs/jindo/jindo_file_system.h
@@ -20,7 +20,6 @@
 #include <functional>
 #include <memory>
 #include <string>
-#include <string_view>
 #include <vector>
 
 #include "JdoFileSystem.hpp"  // NOLINT(build/include_subdir)
@@ -80,7 +79,6 @@ class JindoInputStream : public InputStream {
     // the Jindo Reader.
     std::shared_ptr<JindoFileSystemImpl> fs_;
     std::unique_ptr<JdoReader> reader_;
-    std::string_view result_;
 };
 
 class JindoOutputStream : public OutputStream {
diff --git a/src/paimon/fs/jindo/jindo_file_system_test.cpp b/src/paimon/fs/jindo/jindo_file_system_test.cpp
index 4372292b2..5424952ed 100644
--- a/src/paimon/fs/jindo/jindo_file_system_test.cpp
+++ b/src/paimon/fs/jindo/jindo_file_system_test.cpp
@@ -14,10 +14,19 @@
  * limitations under the License.
  */
 
+#include <chrono>
+#include <cstdint>
+#include <future>
+#include <string>
+#include <unordered_set>
+#include <vector>
+
 #include "gtest/gtest.h"
 #include "paimon/fs/jindo/jindo_file_system_factory.h"
 #include "paimon/testing/utils/testharness.h"
+
 namespace paimon::jindo::test {
+
 // This test shows inconsistent behavior with the local file system in some abnormal scenarios.
 class JindoFileSystemTest : public ::testing::Test {
  public:
@@ -124,4 +133,73 @@ TEST_F(JindoFileSystemTest, TestSeek) {
     ASSERT_OK(in_stream->Close());
 }
 
+TEST(JindoFileSystemPaginationTest, TestListDirAcrossOssPageBoundary) {
+    constexpr int32_t kFileCount = 1234;
+    const std::string test_dir = "oss://paimon-unittest/test_data/jindo_listdir_truncated_1234/";
+    std::map<std::string, std::string> options = paimon::test::GetJindoTestOptions();
+
+    auto fs_factory = std::make_shared<JindoFileSystemFactory>();
+    ASSERT_OK_AND_ASSIGN(std::unique_ptr<FileSystem> fs, fs_factory->Create(test_dir, options));
+
+    std::vector<std::unique_ptr<BasicFileStatus>> file_statuses;
+    ASSERT_OK(fs->ListDir(test_dir, &file_statuses));
+    ASSERT_EQ(file_statuses.size(), kFileCount);
+
+    std::unordered_set<std::string> actual_paths;
+    for (const std::unique_ptr<BasicFileStatus>& file_status : file_statuses) {
+        ASSERT_TRUE(actual_paths.insert(file_status->GetPath()).second)
+            << "duplicate path: " << file_status->GetPath();
+    }
+    for (int32_t i = 0; i < kFileCount; ++i) {
+        std::string index = std::to_string(i);
+        index.insert(/*pos=*/0, /*count=*/4 - index.size(), /*ch=*/'0');
+        ASSERT_NE(actual_paths.find(test_dir + "file-" + index + ".txt"), actual_paths.end());
+    }
+}
+
+TEST(JindoFileSystemAsyncReadTest, TestConcurrentReadAsyncAndReadFromOss) {
+    constexpr int32_t kConcurrentReads = 64;
+    constexpr int64_t kAsyncReadSize = 7;
+    const std::string file_path = "oss://paimon-unittest/test_data/jindo_read_async_128mb.bin";
+    std::map<std::string, std::string> options = paimon::test::GetJindoTestOptions();
+    auto fs_factory = std::make_shared<JindoFileSystemFactory>();
+    ASSERT_OK_AND_ASSIGN(std::unique_ptr<FileSystem> fs, fs_factory->Create(file_path, options));
+    ASSERT_OK_AND_ASSIGN(std::unique_ptr<InputStream> input_stream, fs->Open(file_path));
+
+    std::vector<std::vector<char>> async_buffers(kConcurrentReads,
+                                                 std::vector<char>(kAsyncReadSize));
+    std::vector<std::promise<Status>> promises(kConcurrentReads);
+    std::vector<std::future<Status>> futures;
+    futures.reserve(kConcurrentReads);
+    for (std::promise<Status>& promise : promises) {
+        futures.push_back(promise.get_future());
+    }
+
+    std::vector<Status> sync_read_statuses;
+    std::vector<int64_t> sync_read_sizes;
+    sync_read_statuses.reserve(kConcurrentReads);
+    sync_read_sizes.reserve(kConcurrentReads);
+    for (int32_t i = 0; i < kConcurrentReads; ++i) {
+        input_stream->ReadAsync(
+            async_buffers[i].data(), async_buffers[i].size(), /*offset=*/0,
+            [&promises, i](Status status) { promises[i].set_value(std::move(status)); });
+
+        char sync_buffer = 0;
+        Result<int64_t> sync_read_result =
+            input_stream->Read(&sync_buffer, /*size=*/1, /*offset=*/i);
+        sync_read_statuses.push_back(sync_read_result.status());
+        sync_read_sizes.push_back(sync_read_result.ok() ? sync_read_result.value() : -1);
+    }
+
+    for (int32_t i = 0; i < kConcurrentReads; ++i) {
+        ASSERT_EQ(futures[i].wait_for(std::chrono::seconds(60)), std::future_status::ready)
+            << "async read=" << i;
+        ASSERT_OK(futures[i].get());
+    }
+    for (int32_t i = 0; i < kConcurrentReads; ++i) {
+        ASSERT_OK(sync_read_statuses[i]) << "sync read=" << i;
+        ASSERT_EQ(sync_read_sizes[i], 1) << "sync read=" << i;
+    }
+}
+
 }  // namespace paimon::jindo::test
diff --git a/src/paimon/testing/utils/test_helper.h b/src/paimon/testing/utils/test_helper.h
index 7b710d6d9..b423986bc 100644
--- a/src/paimon/testing/utils/test_helper.h
+++ b/src/paimon/testing/utils/test_helper.h
@@ -59,19 +59,16 @@ class TestHelper {
         const std::vector<std::string>& primary_keys,
         const std::map<std::string, std::string>& options, bool is_streaming_mode,
         bool ignore_if_exists = false, const std::string& temp_directory = "") {
-        // only for test && only check the key
-        auto new_options = options;
-        new_options["enable-object-store-catalog-in-inte-test"] = "";
-        PAIMON_ASSIGN_OR_RAISE(auto catalog, Catalog::Create(root_path, new_options));
-        PAIMON_RETURN_NOT_OK(catalog->CreateDatabase("foo", new_options, ignore_if_exists));
+        PAIMON_ASSIGN_OR_RAISE(auto catalog, Catalog::Create(root_path, options));
+        PAIMON_RETURN_NOT_OK(catalog->CreateDatabase("foo", options, ignore_if_exists));
         ::ArrowSchema c_schema;
         ScopeGuard guard([schema = &c_schema]() { ArrowSchemaRelease(schema); });
         PAIMON_RETURN_NOT_OK_FROM_ARROW(arrow::ExportSchema(*schema, &c_schema));
         PAIMON_RETURN_NOT_OK(catalog->CreateTable(Identifier("foo", "bar"), &c_schema,
-                                                  partition_keys, primary_keys, new_options,
+                                                  partition_keys, primary_keys, options,
                                                   ignore_if_exists));
         std::string table_path = PathUtil::JoinPath(root_path, "foo.db/bar");
-        return Create(table_path, new_options, is_streaming_mode, temp_directory);
+        return Create(table_path, options, is_streaming_mode, temp_directory);
     }
 
     static Result<std::unique_ptr<TestHelper>> Create(
@@ -96,14 +93,10 @@ class TestHelper {
                                    .Finish());
         PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<FileStoreWrite> write,
                                FileStoreWrite::Create(std::move(write_context)));
-        std::map<std::string, std::string> new_options = options;
-        // only for test && only check the key
-        new_options["enable-pk-commit-in-inte-test"] = "";
-        new_options["enable-object-store-commit-in-inte-test"] = "";
         CommitContextBuilder commit_context_builder(table_path, commit_user);
         PAIMON_ASSIGN_OR_RAISE(
             std::unique_ptr<CommitContext> commit_context,
-            commit_context_builder.SetOptions(new_options).IgnoreEmptyCommit(false).Finish());
+            commit_context_builder.SetOptions(options).IgnoreEmptyCommit(false).Finish());
         PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<FileStoreCommit> commit,
                                FileStoreCommit::Create(std::move(commit_context)));
         return std::unique_ptr<TestHelper>(new TestHelper(std::move(file_system), std::move(write),
diff --git a/src/paimon/testing/utils/testharness.cpp b/src/paimon/testing/utils/testharness.cpp
index a90402468..24fb4bd41 100644
--- a/src/paimon/testing/utils/testharness.cpp
+++ b/src/paimon/testing/utils/testharness.cpp
@@ -105,6 +105,8 @@ std::map<std::string, std::string> GetJindoTestOptions() {
         {"fs.oss.bucket.paimon-unittest.accessKeyId", access_key_id},
         {"fs.oss.bucket.paimon-unittest.accessKeySecret", access_key_secret},
         {"fs.oss.user", "paimon"},
+        {"enable-object-store-catalog-in-inte-test", ""},
+        {"enable-object-store-commit-in-inte-test", ""},
     };
     return options;
 }
diff --git a/test/inte/CMakeLists.txt b/test/inte/CMakeLists.txt
index 535394809..2c537f9d4 100644
--- a/test/inte/CMakeLists.txt
+++ b/test/inte/CMakeLists.txt
@@ -40,6 +40,10 @@ if(PAIMON_BUILD_TESTS)
                     ${TEST_STATIC_LINK_LIBS}
                     test_utils_static
                     ${GTEST_LINK_TOOLCHAIN})
+    if(PAIMON_ENABLE_NETWORK_TESTS)
+        target_compile_definitions(paimon-write-and-read-inte-test
+                                   PRIVATE PAIMON_ENABLE_NETWORK_TESTS)
+    endif()
 
     add_paimon_test(clean_inte_test
                     STATIC_LINK_LIBS
diff --git a/test/inte/pk_compaction_inte_test.cpp b/test/inte/pk_compaction_inte_test.cpp
index 15b377c37..cc348f25c 100644
--- a/test/inte/pk_compaction_inte_test.cpp
+++ b/test/inte/pk_compaction_inte_test.cpp
@@ -183,9 +183,6 @@ class PkCompactionInteTest : public ::testing::Test,
     Status Commit(const std::string& table_path,
                   const std::vector<std::shared_ptr<CommitMessage>>& commit_msgs) const {
         CommitContextBuilder commit_builder(table_path, "commit_user_1");
-        std::map<std::string, std::string> commit_options = {
-            {"enable-pk-commit-in-inte-test", ""}, {"enable-object-store-commit-in-inte-test", ""}};
-        commit_builder.SetOptions(commit_options);
         PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<CommitContext> commit_context,
                                commit_builder.Finish());
         PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<FileStoreCommit> file_store_commit,
diff --git a/test/inte/write_and_read_inte_test.cpp b/test/inte/write_and_read_inte_test.cpp
index 321e53ee4..80b9d099c 100644
--- a/test/inte/write_and_read_inte_test.cpp
+++ b/test/inte/write_and_read_inte_test.cpp
@@ -1096,7 +1096,9 @@ TEST_P(WriteAndReadInteTest, TestCharVarcharBinaryVarbinaryTypes) {
 
 std::vector<std::pair<std::string, std::string>> GetTestValuesForWriteAndReadInteTest() {
     std::vector<std::pair<std::string, std::string>> values = {{"parquet", "local"}};
-    // values.emplace_back("parquet", "jindo");
+#if defined(PAIMON_ENABLE_NETWORK_TESTS) && defined(PAIMON_ENABLE_JINDO)
+    values.emplace_back("parquet", "jindo");
+#endif
 #ifdef PAIMON_ENABLE_ORC
     values.emplace_back("orc", "local");
 #endif
diff --git a/test/inte/write_inte_test.cpp b/test/inte/write_inte_test.cpp
index 7dd1b114c..bde4ce776 100644
--- a/test/inte/write_inte_test.cpp
+++ b/test/inte/write_inte_test.cpp
@@ -345,11 +345,9 @@ class WriteInteTest : public testing::Test, public ::testing::WithParamInterface
 
     Status CommitMessages(const std::string& table_path,
                           const std::vector<std::shared_ptr<CommitMessage>>& commit_messages,
-                          const std::map<std::string, std::string>& commit_options,
                           bool ignore_empty_commit = true,
                           int64_t commit_identifier = BATCH_WRITE_COMMIT_IDENTIFIER) const {
         CommitContextBuilder commit_builder(table_path, "commit_user_1");
-        commit_builder.SetOptions(commit_options);
         commit_builder.IgnoreEmptyCommit(ignore_empty_commit);
         PAIMON_ASSIGN_OR_RAISE(auto commit_context, commit_builder.Finish());
         PAIMON_ASSIGN_OR_RAISE(auto file_store_commit,
@@ -2546,10 +2544,7 @@ TEST_P(WriteInteTest, TestWriteWithFieldId) {
     ASSERT_OK(file_store_write->Close());
 
     // commit
-    std::map<std::string, std::string> commit_options = {
-        {Options::MANIFEST_TARGET_FILE_SIZE, "8mb"}, {Options::FILE_SYSTEM, "local"}};
-    ASSERT_OK(CommitMessages(table_path, commit_messages, commit_options,
-                             /*ignore_empty_commit=*/false));
+    ASSERT_OK(CommitMessages(table_path, commit_messages, /*ignore_empty_commit=*/false));
 
     // check data file has field id meta
     std::vector<std::unique_ptr<BasicFileStatus>> status_list;
@@ -2631,10 +2626,8 @@ TEST_P(WriteInteTest, TestAppendTableWriteAndReadWithExternalPath) {
     ASSERT_EQ(results.size(), 1);
     auto commit_msg_impl = std::dynamic_pointer_cast<CommitMessageImpl>(results[0]);
     auto meta = commit_msg_impl->data_increment_.new_files_[0];
-    std::map<std::string, std::string> commit_options = {
-        {Options::MANIFEST_TARGET_FILE_SIZE, "8mb"}, {Options::FILE_SYSTEM, "local"}};
-    ASSERT_OK(CommitMessages(root_path, results, commit_options,
-                             /*ignore_empty_commit=*/false, /*commit_identifier=*/1));
+    ASSERT_OK(CommitMessages(root_path, results, /*ignore_empty_commit=*/false,
+                             /*commit_identifier=*/1));
 
     // check external path
     ASSERT_OK_AND_ASSIGN(bool file_exist, file_system_->Exists(meta->external_path.value()));
@@ -2952,10 +2945,8 @@ TEST_P(WriteInteTest, TestWriteAndReadWithSpecialPartitionValue) {
     ASSERT_EQ(results.size(), 3);
     auto commit_msg_impl = std::dynamic_pointer_cast<CommitMessageImpl>(results[0]);
     auto meta = commit_msg_impl->data_increment_.new_files_[0];
-    std::map<std::string, std::string> commit_options = {
-        {Options::MANIFEST_TARGET_FILE_SIZE, "8mb"}, {Options::FILE_SYSTEM, "local"}};
-    ASSERT_OK(CommitMessages(root_path, results, commit_options,
-                             /*ignore_empty_commit=*/false, /*commit_identifier=*/1));
+    ASSERT_OK(CommitMessages(root_path, results, /*ignore_empty_commit=*/false,
+                             /*commit_identifier=*/1));
 
     arrow::FieldVector fields_with_row_kind = fields;
     fields_with_row_kind.insert(fields_with_row_kind.begin(),
@@ -3137,10 +3128,8 @@ TEST_P(WriteInteTest, TestWriteWithNestedSchema) {
     ASSERT_OK_AND_ASSIGN(std::vector<std::shared_ptr<CommitMessage>> results,
                          file_store_write->PrepareCommit());
     ASSERT_EQ(results.size(), 1);
-    std::map<std::string, std::string> commit_options = {
-        {Options::MANIFEST_TARGET_FILE_SIZE, "8mb"}, {Options::FILE_SYSTEM, "local"}};
-    ASSERT_OK(CommitMessages(root_path, results, commit_options,
-                             /*ignore_empty_commit=*/false, /*commit_identifier=*/1));
+    ASSERT_OK(CommitMessages(root_path, results, /*ignore_empty_commit=*/false,
+                             /*commit_identifier=*/1));
 
     // check read result
     ScanContextBuilder scan_context_builder(table_path);
@@ -4080,9 +4069,7 @@ TEST_P(WriteInteTest, TestPkSpillableDiskQuotaExhaustedFallsBackToFlush) {
                          file_store_write->PrepareCommit(/*wait_compaction=*/false,
                                                          /*commit_identifier=*/0));
 
-    std::map<std::string, std::string> pk_commit_options = {
-        {"enable-pk-commit-in-inte-test", ""}, {"enable-object-store-commit-in-inte-test", ""}};
-    ASSERT_OK(CommitMessages(table_path, commit_messages, pk_commit_options));
+    ASSERT_OK(CommitMessages(table_path, commit_messages));
     ASSERT_OK(file_store_write->Close());
 
     std::string expected = R"([
@@ -4152,9 +4139,7 @@ TEST_P(WriteInteTest, TestPkSpillableGlobalMemoryPreemptionDataCorrectness) {
 
     ASSERT_EQ(0, TestHelper::CountChannelFiles(file_system_, tmp_dir));
 
-    std::map<std::string, std::string> pk_commit_options = {
-        {"enable-pk-commit-in-inte-test", ""}, {"enable-object-store-commit-in-inte-test", ""}};
-    ASSERT_OK(CommitMessages(table_path, commit_messages, pk_commit_options));
+    ASSERT_OK(CommitMessages(table_path, commit_messages));
     ASSERT_OK(file_store_write->Close());
 
     // Scan and verify both partitions
@@ -4318,9 +4303,7 @@ TEST_P(WriteInteTest, TestPkSpillableIntermediateMergeWithTempFileTracking) {
                                                          /*commit_identifier=*/0));
     ASSERT_EQ(0, TestHelper::CountChannelFiles(file_system_, tmp_dir));
 
-    std::map<std::string, std::string> pk_commit_options = {
-        {"enable-pk-commit-in-inte-test", ""}, {"enable-object-store-commit-in-inte-test", ""}};
-    ASSERT_OK(CommitMessages(table_path, commit_messages, pk_commit_options));
+    ASSERT_OK(CommitMessages(table_path, commit_messages));
     ASSERT_OK(file_store_write->Close());
 
     // Scan: Alice deduped to f1=3, Bob f1=2
@@ -4417,9 +4400,7 @@ TEST_P(WriteInteTest, TestPkSpillableMultiBucketMultiRoundDataCorrectness) {
                                                          /*commit_identifier=*/0));
     // Spill files should be cleaned after PrepareCommit
     ASSERT_EQ(0, TestHelper::CountChannelFiles(file_system_, tmp_dir));
-    std::map<std::string, std::string> pk_commit_options = {
-        {"enable-pk-commit-in-inte-test", ""}, {"enable-object-store-commit-in-inte-test", ""}};
-    ASSERT_OK(CommitMessages(table_path, commit_messages_1, pk_commit_options));
+    ASSERT_OK(CommitMessages(table_path, commit_messages_1));
 
     // Round 2: Bucket 0 writes Charlie + Bob(overwrite), Bucket 1 writes Frank + Eve(overwrite)
     auto r2_b0_batch1 =
@@ -4450,7 +4431,7 @@ TEST_P(WriteInteTest, TestPkSpillableMultiBucketMultiRoundDataCorrectness) {
                                                          /*commit_identifier=*/1));
     // Spill files should be cleaned after PrepareCommit
     ASSERT_EQ(0, TestHelper::CountChannelFiles(file_system_, tmp_dir));
-    ASSERT_OK(CommitMessages(table_path, commit_messages_2, pk_commit_options));
+    ASSERT_OK(CommitMessages(table_path, commit_messages_2));
     ASSERT_OK(file_store_write->Close());
 
     // Scan and verify per (partition, bucket)
@@ -4594,10 +4575,8 @@ TEST_P(WriteInteTest, TestPkSpillableWithIOException) {
         io_hook->Clear();
 
         // Commit both rounds
-        std::map<std::string, std::string> pk_commit_options = {
-            {"enable-pk-commit-in-inte-test", ""}, {"enable-object-store-commit-in-inte-test", ""}};
-        ASSERT_OK(CommitMessages(root_path, results_1.value(), pk_commit_options));
-        ASSERT_OK(CommitMessages(root_path, results_2.value(), pk_commit_options));
+        ASSERT_OK(CommitMessages(root_path, results_1.value()));
+        ASSERT_OK(CommitMessages(root_path, results_2.value()));
         ASSERT_OK(file_store_write->Close());
 
         // Scan and verify final state after spill: