diff --git a/include/paimon/logging.h b/include/paimon/logging.h
index 0859d4435..95d22b7d2 100644
--- a/include/paimon/logging.h
+++ b/include/paimon/logging.h
@@ -55,8 +55,24 @@ class PAIMON_EXPORT Logger {
  public:
     using LoggerCreator = std::function<std::unique_ptr<Logger>(const std::string&)>;
 
+    /// Register a creator that `GetLogger()` uses instead of the default glog logger.
+    ///
+    /// @param creator Creates a logger from the `path` passed to `GetLogger()`. All `GetLogger()`
+    /// calls share it, so it must be safe to call from multiple threads simultaneously, and its
+    /// state persists across calls. It may call `RegisterLogger()`. It replaces any previously
+    /// registered creator, which stays alive until running `GetLogger()` calls finish with it.
+    ///
+    /// @note This method is thread-safe and can be called concurrently with `GetLogger()`.
     static void RegisterLogger(LoggerCreator creator);
 
+    /// Get a logger from the registered creator, or a glog logger if no creator is registered.
+    ///
+    /// @param path Name of the logger, such as the class that logs. It is passed to the
+    /// registered creator, and the glog logger ignores it.
+    ///
+    /// @return A unique pointer to the logger.
+    ///
+    /// @note This method is thread-safe.
     static std::unique_ptr<Logger> GetLogger(const std::string& path);
 
     virtual void LogV(PaimonLogLevel level, const char* fname, int lineno, const char* function,
diff --git a/src/paimon/common/logging/logging.cpp b/src/paimon/common/logging/logging.cpp
index c91d8fc96..bfd649f68 100644
--- a/src/paimon/common/logging/logging.cpp
+++ b/src/paimon/common/logging/logging.cpp
@@ -23,9 +23,10 @@
 #include <cstdarg>
 #include <cstdio>
 #include <cstdlib>
+#include <memory>
 #include <mutex>
-#include <optional>
 #include <shared_mutex>
+#include <utility>
 
 #include "glog/log_severity.h"
 #include "glog/logging.h"
@@ -43,8 +44,8 @@ static const char* GetProgramName() {
 #endif
 }
 
-static std::optional<Logger::LoggerCreator>& getLoggerCreator() {
-    static std::optional<Logger::LoggerCreator> _loggerCreator;
+static std::shared_ptr<Logger::LoggerCreator>& getLoggerCreator() {
+    static std::shared_ptr<Logger::LoggerCreator> _loggerCreator;
     return _loggerCreator;
 }
 
@@ -55,7 +56,7 @@ static std::shared_mutex& getRegistryLock() {
 
 void Logger::RegisterLogger(LoggerCreator creator) {
     std::unique_lock<std::shared_mutex> lock(getRegistryLock());
-    getLoggerCreator() = creator;
+    getLoggerCreator() = std::make_shared<LoggerCreator>(std::move(creator));
 }
 
 static google::LogSeverity ToGlogLevel(PaimonLogLevel level) {
@@ -99,15 +100,22 @@ class GlogAdaptor : public Logger {
 };
 
 std::unique_ptr<Logger> Logger::GetLogger(const std::string& path) {
-    auto& creator = getLoggerCreator();
-    if (creator) {
+    // Share the registered creator under the lock and call it after unlocking, so it keeps
+    // its state, stays alive if replaced, and may call RegisterLogger() without deadlocking.
+    std::shared_ptr<LoggerCreator> creator;
+    {
         std::shared_lock<std::shared_mutex> lock(getRegistryLock());
-        return creator.value()(path);
+        creator = getLoggerCreator();
     }
-    std::unique_lock<std::shared_mutex> ulock(getRegistryLock());
-    if (!google::IsGoogleLoggingInitialized()) {
-        google::InitGoogleLogging(GetProgramName());
+    if (creator) {
+        return (*creator)(path);
     }
+    static std::once_flag glog_init_flag;
+    std::call_once(glog_init_flag, []() {
+        if (!google::IsGoogleLoggingInitialized()) {
+            google::InitGoogleLogging(GetProgramName());
+        }
+    });
     return std::make_unique<GlogAdaptor>();
 }
 
diff --git a/src/paimon/common/logging/logging_test.cpp b/src/paimon/common/logging/logging_test.cpp
index 3c391a4bf..d0122b31d 100644
--- a/src/paimon/common/logging/logging_test.cpp
+++ b/src/paimon/common/logging/logging_test.cpp
@@ -19,6 +19,7 @@
 #include "paimon/logging.h"
 
 #include <atomic>
+#include <chrono>
 #include <cstdlib>
 #include <functional>
 #include <future>
@@ -66,6 +67,55 @@ void RegisterCustomLoggerCreatorAndExit() {
     std::_Exit(used_creator ? 0 : 1);
 }
 
+void RegisterStatefulLoggerCreatorAndExit() {
+    g_creator_calls.store(0);
+    Logger::RegisterLogger([calls = 0](const std::string&) mutable -> std::unique_ptr<Logger> {
+        g_creator_calls.store(++calls);
+        return nullptr;
+    });
+    Logger::GetLogger("first");
+    Logger::GetLogger("second");
+    std::_Exit(g_creator_calls.load() == 2 ? 0 : 1);
+}
+
+void RegisterLoggerCreatorWhileGettingLoggerAndExit() {
+    g_creator_calls.store(0);
+    std::thread getter([]() {
+        while (g_creator_calls.load() == 0) {
+            Logger::GetLogger("getter");
+        }
+    });
+    Logger::RegisterLogger([](const std::string&) -> std::unique_ptr<Logger> {
+        g_creator_calls.fetch_add(1);
+        return nullptr;
+    });
+    getter.join();
+    // std::exit runs ThreadSanitizer's exit check, which fails the process on a race.
+    std::exit(0);
+}
+
+void RegisterLoggerFromLoggerCreatorAndExit() {
+    static std::weak_ptr<int> creator_state;
+    static bool kept_alive = false;
+    auto state = std::make_shared<int>(0);
+    creator_state = state;
+    Logger::RegisterLogger([state](const std::string&) -> std::unique_ptr<Logger> {
+        Logger::RegisterLogger(
+            [](const std::string&) -> std::unique_ptr<Logger> { return nullptr; });
+        kept_alive = !creator_state.expired();
+        return nullptr;
+    });
+    state.reset();
+    std::future<void> getter =
+        std::async(std::launch::async, []() { Logger::GetLogger("getter"); });
+    if (getter.wait_for(std::chrono::seconds(30)) != std::future_status::ready) {
+        std::cerr << "GetLogger() did not finish within 30 seconds" << std::endl;
+        std::_Exit(1);
+    }
+    getter.get();
+    std::_Exit(kept_alive && creator_state.expired() ? 0 : 1);
+}
+
 }  // namespace
 TEST(LoggerTest, TestMultiThreadGetLogger) {
     ASSERT_OK_AND_ASSIGN(auto executor, CreateDefaultExecutor(/*thread_count=*/4));
@@ -192,4 +242,19 @@ TEST(LoggerTest, TestRegisterCustomLoggerCreator) {
     ::testing::FLAGS_gtest_death_test_style = "threadsafe";
     ASSERT_EXIT(RegisterCustomLoggerCreatorAndExit(), ::testing::ExitedWithCode(0), "");
 }
+
+TEST(LoggerTest, TestRegisterStatefulLoggerCreator) {
+    ::testing::FLAGS_gtest_death_test_style = "threadsafe";
+    ASSERT_EXIT(RegisterStatefulLoggerCreatorAndExit(), ::testing::ExitedWithCode(0), "");
+}
+
+TEST(LoggerTest, TestRegisterLoggerCreatorWhileGettingLogger) {
+    ::testing::FLAGS_gtest_death_test_style = "threadsafe";
+    ASSERT_EXIT(RegisterLoggerCreatorWhileGettingLoggerAndExit(), ::testing::ExitedWithCode(0), "");
+}
+
+TEST(LoggerTest, TestRegisterLoggerFromLoggerCreator) {
+    ::testing::FLAGS_gtest_death_test_style = "threadsafe";
+    ASSERT_EXIT(RegisterLoggerFromLoggerCreatorAndExit(), ::testing::ExitedWithCode(0), "");
+}
 }  // namespace paimon::test