diff --git a/be/src/util/brpc_stub_cache.cpp b/be/src/util/brpc_stub_cache.cpp
index 2371ff695b4db3..b30e16bc3b1406 100644
--- a/be/src/util/brpc_stub_cache.cpp
+++ b/be/src/util/brpc_stub_cache.cpp
@@ -14,7 +14,14 @@
 
 #include "util/brpc_stub_cache.h"
 
+<<<<<<< HEAD:be/src/util/brpc_stub_cache.cpp
 #include "common/config.h"
+=======
+#include "base/failpoint/fail_point.h"
+#include "base/metrics.h"
+#include "base/time/time.h"
+#include "common/config_network_fwd.h"
+>>>>>>> 2bbca67281 ([BugFix] Fix bRPC stub cache clean timer leak (#75973)):be/src/common/brpc/brpc_stub_cache.cpp
 #include "gen_cpp/internal_service.pb.h"
 #include "gen_cpp/lake_service.pb.h"
 #include "runtime/exec_env.h"
@@ -23,7 +30,82 @@
 
 namespace starrocks {
 
+<<<<<<< HEAD:be/src/util/brpc_stub_cache.cpp
 BrpcStubCache::BrpcStubCache(ExecEnv* exec_env) : _pipeline_timer(exec_env->pipeline_timer()) {
+=======
+namespace {
+
+const char* const kBrpcEndpointStubCountMetric = "brpc_endpoint_stub_count";
+
+template <typename Cache>
+Cache*& singleton_cache() {
+    static Cache* cache = nullptr;
+    return cache;
+}
+
+template <typename Cache>
+std::mutex& singleton_cache_mutex() {
+    static std::mutex mutex;
+    return mutex;
+}
+
+inline int64_t absolute_deadline_us(int64_t ttl_seconds) {
+    return butil::gettimeofday_us() + ttl_seconds * 1000 * 1000;
+}
+} // namespace
+
+template <typename CacheT, typename ExtractFn>
+void wait_clean_tasks_terminate(CacheT* cache, ExtractFn extract) {
+    std::vector<std::shared_ptr<EndpointCleanupTask<CacheT>>> tasks;
+    BthreadTimer* timer = nullptr;
+    {
+        std::lock_guard<SpinLock> l(cache->_lock);
+        cache->_stopping = true;
+        timer = cache->_timer;
+        cache->_timer = nullptr;
+        for (auto& stub : cache->_stub_map) {
+            tasks.push_back(extract(stub.second));
+        }
+        cache->_stub_map.clear();
+    }
+    if (timer != nullptr) {
+        for (auto& task : tasks) {
+            task->unschedule_and_join(timer);
+        }
+    }
+}
+
+template <typename CacheT>
+void reset_state_for_rebind(CacheT* cache, BthreadTimer* timer) {
+    DCHECK(timer != nullptr);
+    std::lock_guard<SpinLock> l(cache->_lock);
+    DCHECK(cache->_stub_map.empty() || cache->_timer == timer);
+    cache->_stopping = false;
+    cache->_timer = timer;
+}
+
+struct BrpcStubCache::Metrics {
+    Metrics(MetricRegistry* metric_registry, BrpcStubCache* cache) : registry(metric_registry), cache(cache) {
+        DCHECK(registry != nullptr);
+        registry->register_metric(kBrpcEndpointStubCountMetric, &brpc_endpoint_stub_count);
+        registry->register_hook(kBrpcEndpointStubCountMetric, [this] {
+            std::lock_guard<SpinLock> l(this->cache->_lock);
+            brpc_endpoint_stub_count.set_value(this->cache->_stub_map.size());
+        });
+    }
+
+    ~Metrics() {
+        registry->deregister_hook(kBrpcEndpointStubCountMetric);
+        brpc_endpoint_stub_count.hide();
+    }
+
+    MetricRegistry* registry;
+    BrpcStubCache* cache;
+    UIntGauge brpc_endpoint_stub_count{MetricUnit::NOUNIT};
+};
+
+BrpcStubCache::BrpcStubCache(BthreadTimer* timer, MetricRegistry* metric_registry) : _timer(timer) {
+>>>>>>> 2bbca67281 ([BugFix] Fix bRPC stub cache clean timer leak (#75973)):be/src/common/brpc/brpc_stub_cache.cpp
     _stub_map.init(239);
     REGISTER_GAUGE_STARROCKS_METRIC(brpc_endpoint_stub_count, [this]() {
         std::lock_guard<SpinLock> l(_lock);
@@ -32,6 +114,7 @@ BrpcStubCache::BrpcStubCache(ExecEnv* exec_env) : _pipeline_timer(exec_env->pipe
 }
 
 BrpcStubCache::~BrpcStubCache() {
+<<<<<<< HEAD:be/src/util/brpc_stub_cache.cpp
     std::vector<std::shared_ptr<StubPool>> pools_to_cleanup;
     {
         std::lock_guard<SpinLock> l(_lock);
@@ -46,6 +129,20 @@ BrpcStubCache::~BrpcStubCache() {
     }
     std::lock_guard<SpinLock> l(_lock);
     _stub_map.clear();
+=======
+    _metrics.reset();
+    wait_clean_tasks_terminate(this, [](const std::shared_ptr<StubPool>& pool) { return pool->_cleanup_task; });
+}
+
+bool BrpcStubCache::replace_cleanup_task_locked(const butil::EndPoint& endpoint,
+                                                std::shared_ptr<EndpointCleanupTask<BrpcStubCache>> task) {
+    auto pool = _stub_map.seek(endpoint);
+    if (pool != nullptr) {
+        (*pool)->_cleanup_task = std::move(task);
+        return true;
+    }
+    return false;
+>>>>>>> 2bbca67281 ([BugFix] Fix bRPC stub cache clean timer leak (#75973)):be/src/common/brpc/brpc_stub_cache.cpp
 }
 
 std::shared_ptr<PInternalService_RecoverableStub> BrpcStubCache::get_stub(const butil::EndPoint& endpoint) {
@@ -54,17 +151,27 @@ std::shared_ptr<PInternalService_RecoverableStub> BrpcStubCache::get_stub(const
     auto stub_pool = _stub_map.seek(endpoint);
     if (stub_pool == nullptr) {
         auto new_pool = std::make_shared<StubPool>();
-        new_pool->_cleanup_task = std::make_shared<EndpointCleanupTask<BrpcStubCache>>(this, endpoint);
+        new_pool->_cleanup_task =
+                std::make_shared<EndpointCleanupTask<BrpcStubCache>>(this, endpoint, config::brpc_stub_expire_s);
+        new_pool->_cleanup_task->renew_deadline_locked(absolute_deadline_us(config::brpc_stub_expire_s));
         _stub_map.insert(endpoint, new_pool);
         stub_pool = _stub_map.seek(endpoint);
-    }
 
+<<<<<<< HEAD:be/src/util/brpc_stub_cache.cpp
     if (_pipeline_timer->unschedule((*stub_pool)->_cleanup_task.get()) != TIMER_TASK_RUNNING) {
         timespec tm = butil::seconds_from_now(config::brpc_stub_expire_s);
         auto status = _pipeline_timer->schedule((*stub_pool)->_cleanup_task.get(), tm);
+=======
+        timespec tm = butil::microseconds_to_timespec((*stub_pool)->_cleanup_task->deadline_locked());
+        auto status = _timer->schedule((*stub_pool)->_cleanup_task.get(), tm);
+>>>>>>> 2bbca67281 ([BugFix] Fix bRPC stub cache clean timer leak (#75973)):be/src/common/brpc/brpc_stub_cache.cpp
         if (!status.ok()) {
             LOG(WARNING) << "Failed to schedule brpc cleanup task: " << endpoint;
+            _stub_map.erase(endpoint);
+            return new_pool->get_or_create(endpoint);
         }
+    } else {
+        (*stub_pool)->_cleanup_task->renew_deadline_locked(absolute_deadline_us(config::brpc_stub_expire_s));
     }
 
     return (*stub_pool)->get_or_create(endpoint);
@@ -94,6 +201,7 @@ std::shared_ptr<PInternalService_RecoverableStub> BrpcStubCache::get_stub(const
     return get_stub(endpoint);
 }
 
+<<<<<<< HEAD:be/src/util/brpc_stub_cache.cpp
 void BrpcStubCache::cleanup_expired(const butil::EndPoint& endpoint) {
     std::lock_guard<SpinLock> l(_lock);
 
@@ -102,6 +210,9 @@ void BrpcStubCache::cleanup_expired(const butil::EndPoint& endpoint) {
 }
 
 BrpcStubCache::StubPool::StubPool() : _idx(-1) {
+=======
+BrpcStubCache::StubPool::StubPool() {
+>>>>>>> 2bbca67281 ([BugFix] Fix bRPC stub cache clean timer leak (#75973)):be/src/common/brpc/brpc_stub_cache.cpp
     _stubs.reserve(config::brpc_max_connections_per_server);
 }
 
@@ -140,6 +251,7 @@ HttpBrpcStubCache::~HttpBrpcStubCache() {
     shutdown();
 }
 
+<<<<<<< HEAD:be/src/util/brpc_stub_cache.cpp
 void HttpBrpcStubCache::shutdown() {
     std::vector<std::shared_ptr<EndpointCleanupTask<HttpBrpcStubCache>>> task_to_cleanup;
 
@@ -160,6 +272,24 @@ void HttpBrpcStubCache::shutdown() {
         std::lock_guard<SpinLock> l(_lock);
         _stub_map.clear();
     }
+=======
+void HttpBrpcStubCache::bind_timer(BthreadTimer* timer) {
+    reset_state_for_rebind(this, timer);
+}
+
+void HttpBrpcStubCache::shutdown() {
+    wait_clean_tasks_terminate(this, [](const StubEntry& entry) { return entry.cleanup_task; });
+}
+
+bool HttpBrpcStubCache::replace_cleanup_task_locked(const butil::EndPoint& endpoint,
+                                                    std::shared_ptr<EndpointCleanupTask<HttpBrpcStubCache>> task) {
+    auto entry = _stub_map.seek(endpoint);
+    if (entry != nullptr) {
+        entry->cleanup_task = std::move(task);
+        return true;
+    }
+    return false;
+>>>>>>> 2bbca67281 ([BugFix] Fix bRPC stub cache clean timer leak (#75973)):be/src/common/brpc/brpc_stub_cache.cpp
 }
 
 StatusOr<std::shared_ptr<PInternalService_RecoverableStub>> HttpBrpcStubCache::get_http_stub(
@@ -185,33 +315,36 @@ StatusOr<std::shared_ptr<PInternalService_RecoverableStub>> HttpBrpcStubCache::g
     auto stub_pair_ptr = _stub_map.seek(endpoint);
     if (stub_pair_ptr == nullptr) {
         // create
-        auto new_task = std::make_shared<EndpointCleanupTask<HttpBrpcStubCache>>(this, endpoint);
+        auto new_task =
+                std::make_shared<EndpointCleanupTask<HttpBrpcStubCache>>(this, endpoint, config::brpc_stub_expire_s);
         auto stub = std::make_shared<PInternalService_RecoverableStub>(endpoint, "http");
         if (!stub->reset_channel().ok()) {
             return Status::RuntimeError("init http brpc channel error on " + taddr.hostname + ":" +
                                         std::to_string(taddr.port));
         }
-        _stub_map.insert(endpoint, std::make_pair(stub, new_task));
+        new_task->renew_deadline_locked(absolute_deadline_us(config::brpc_stub_expire_s));
+        _stub_map.insert(endpoint, StubEntry{stub, new_task});
         stub_pair_ptr = _stub_map.seek(endpoint);
-    }
 
+<<<<<<< HEAD:be/src/util/brpc_stub_cache.cpp
     // schedule clean up task
     if (_pipeline_timer->unschedule((*stub_pair_ptr).second.get()) != TIMER_TASK_RUNNING) {
         timespec tm = butil::seconds_from_now(config::brpc_stub_expire_s);
         auto status = _pipeline_timer->schedule((*stub_pair_ptr).second.get(), tm);
+=======
+        timespec tm = butil::microseconds_to_timespec(stub_pair_ptr->cleanup_task->deadline_locked());
+        auto status = _timer->schedule(stub_pair_ptr->cleanup_task.get(), tm);
+>>>>>>> 2bbca67281 ([BugFix] Fix bRPC stub cache clean timer leak (#75973)):be/src/common/brpc/brpc_stub_cache.cpp
         if (!status.ok()) {
-            LOG(WARNING) << "Failed to schedule http brpc cleanup task: " << endpoint;
+            LOG(WARNING) << "Failed to schedule brpc cleanup task: " << endpoint;
+            _stub_map.erase(endpoint);
+            return stub;
         }
+    } else {
+        stub_pair_ptr->cleanup_task->renew_deadline_locked(absolute_deadline_us(config::brpc_stub_expire_s));
     }
 
-    return (*stub_pair_ptr).first;
-}
-
-void HttpBrpcStubCache::cleanup_expired(const butil::EndPoint& endpoint) {
-    std::lock_guard<SpinLock> l(_lock);
-
-    LOG(INFO) << "cleanup http brpc stub, endpoint:" << endpoint;
-    _stub_map.erase(endpoint);
+    return stub_pair_ptr->stub;
 }
 
 LakeServiceBrpcStubCache* LakeServiceBrpcStubCache::getInstance() {
@@ -228,6 +361,7 @@ LakeServiceBrpcStubCache::~LakeServiceBrpcStubCache() {
     shutdown();
 }
 
+<<<<<<< HEAD:be/src/util/brpc_stub_cache.cpp
 void LakeServiceBrpcStubCache::shutdown() {
     std::vector<std::shared_ptr<EndpointCleanupTask<LakeServiceBrpcStubCache>>> task_to_cleanup;
 
@@ -248,6 +382,24 @@ void LakeServiceBrpcStubCache::shutdown() {
         std::lock_guard<SpinLock> l(_lock);
         _stub_map.clear();
     }
+=======
+void LakeServiceBrpcStubCache::bind_timer(BthreadTimer* timer) {
+    reset_state_for_rebind(this, timer);
+}
+
+void LakeServiceBrpcStubCache::shutdown() {
+    wait_clean_tasks_terminate(this, [](const StubEntry& entry) { return entry.cleanup_task; });
+}
+
+bool LakeServiceBrpcStubCache::replace_cleanup_task_locked(
+        const butil::EndPoint& endpoint, std::shared_ptr<EndpointCleanupTask<LakeServiceBrpcStubCache>> task) {
+    auto entry = _stub_map.seek(endpoint);
+    if (entry != nullptr) {
+        entry->cleanup_task = std::move(task);
+        return true;
+    }
+    return false;
+>>>>>>> 2bbca67281 ([BugFix] Fix bRPC stub cache clean timer leak (#75973)):be/src/common/brpc/brpc_stub_cache.cpp
 }
 
 DEFINE_FAIL_POINT(get_stub_return_nullptr);
@@ -272,31 +424,45 @@ StatusOr<std::shared_ptr<starrocks::LakeService_RecoverableStub>> LakeServiceBrp
     if (stub_pair_ptr == nullptr) {
         // create
         auto stub = std::make_shared<starrocks::LakeService_RecoverableStub>(endpoint, "");
-        auto new_task = std::make_shared<EndpointCleanupTask<LakeServiceBrpcStubCache>>(this, endpoint);
+        auto new_task = std::make_shared<EndpointCleanupTask<LakeServiceBrpcStubCache>>(this, endpoint,
+                                                                                        config::brpc_stub_expire_s);
         if (!stub->reset_channel().ok()) {
             return Status::RuntimeError("init lakeService brpc channel error on " + host + ":" + std::to_string(port));
         }
-        _stub_map.insert(endpoint, std::make_pair(stub, new_task));
+        new_task->renew_deadline_locked(absolute_deadline_us(config::brpc_stub_expire_s));
+        _stub_map.insert(endpoint, StubEntry{stub, new_task});
         stub_pair_ptr = _stub_map.seek(endpoint);
-    }
 
+<<<<<<< HEAD:be/src/util/brpc_stub_cache.cpp
     // schedule clean up task
     if (_pipeline_timer->unschedule((*stub_pair_ptr).second.get()) != TIMER_TASK_RUNNING) {
         timespec tm = butil::seconds_from_now(config::brpc_stub_expire_s);
         auto status = _pipeline_timer->schedule((*stub_pair_ptr).second.get(), tm);
+=======
+        timespec tm = butil::microseconds_to_timespec(stub_pair_ptr->cleanup_task->deadline_locked());
+        auto status = _timer->schedule(stub_pair_ptr->cleanup_task.get(), tm);
+>>>>>>> 2bbca67281 ([BugFix] Fix bRPC stub cache clean timer leak (#75973)):be/src/common/brpc/brpc_stub_cache.cpp
         if (!status.ok()) {
-            LOG(WARNING) << "Failed to schedule lake brpc cleanup task: " << endpoint;
+            LOG(WARNING) << "Failed to schedule brpc cleanup task: " << endpoint;
+            _stub_map.erase(endpoint);
+            return stub;
         }
+    } else {
+        stub_pair_ptr->cleanup_task->renew_deadline_locked(absolute_deadline_us(config::brpc_stub_expire_s));
     }
 
-    return (*stub_pair_ptr).first;
+    return stub_pair_ptr->stub;
 }
 
+<<<<<<< HEAD:be/src/util/brpc_stub_cache.cpp
 void LakeServiceBrpcStubCache::cleanup_expired(const butil::EndPoint& endpoint) {
     std::lock_guard<SpinLock> l(_lock);
 
     LOG(INFO) << "cleanup lake service brpc stub, endpoint:" << endpoint;
     _stub_map.erase(endpoint);
 }
+=======
+#endif
+>>>>>>> 2bbca67281 ([BugFix] Fix bRPC stub cache clean timer leak (#75973)):be/src/common/brpc/brpc_stub_cache.cpp
 
 } // namespace starrocks
\ No newline at end of file
diff --git a/be/src/util/brpc_stub_cache.h b/be/src/util/brpc_stub_cache.h
index 5b3de3102d8bed..0a76518411e794 100644
--- a/be/src/util/brpc_stub_cache.h
+++ b/be/src/util/brpc_stub_cache.h
@@ -38,6 +38,16 @@
 #include <mutex>
 #include <vector>
 
+<<<<<<< HEAD:be/src/util/brpc_stub_cache.h
+=======
+#include "base/brpc/brpc.h"
+#include "base/concurrency/spinlock.h"
+#include "base/network/network_util.h"
+#include "base/time/time.h"
+#include "common/brpc/internal_service_recoverable_stub.h"
+#include "common/bthread_timer.h"
+#include "common/logging.h"
+>>>>>>> 2bbca67281 ([BugFix] Fix bRPC stub cache clean timer leak (#75973)):be/src/common/brpc/brpc_stub_cache.h
 #include "common/statusor.h"
 #include "exec/pipeline/schedule/pipeline_timer.h"
 #include "gen_cpp/Types_types.h" // TNetworkAddress
@@ -54,14 +64,64 @@ constexpr int TIMER_TASK_RUNNING = 1;
 class ExecEnv;
 
 template <typename StubCacheT>
+<<<<<<< HEAD:be/src/util/brpc_stub_cache.h
 class EndpointCleanupTask : public starrocks::pipeline::LightTimerTask {
+=======
+class EndpointCleanupTask : public BthreadTimerTask {
+>>>>>>> 2bbca67281 ([BugFix] Fix bRPC stub cache clean timer leak (#75973)):be/src/common/brpc/brpc_stub_cache.h
 public:
-    EndpointCleanupTask(StubCacheT* cache, const butil::EndPoint& endpoint) : _cache(cache), _endpoint(endpoint){};
-    void Run() override { _cache->cleanup_expired(_endpoint); }
+    // ttl_seconds is the cache-wide expire window (config::brpc_stub_expire_s).
+    EndpointCleanupTask(StubCacheT* cache, const butil::EndPoint& endpoint, int64_t ttl_seconds)
+            : _cache(cache), _endpoint(endpoint), _ttl_seconds(ttl_seconds) {}
+    // The actual cleanup/renewal decision must run while the cache's _lock is held so
+    // that _stopping and _deadline are observed atomically with the cache state.
+    void Run() override {
+        std::lock_guard<SpinLock> l(_cache->_lock);
+        if (_cache->_stopping) {
+            return;
+        }
+        // Make sure this task is still the authoritative cleanup task for the
+        // endpoint before rescheduling. If shutdown() cleared the cache or a new
+        // entry was created for the same endpoint, this task is stale and must
+        // not schedule anything.
+        if (!_cache->is_cleanup_task_owner_locked(_endpoint, this)) {
+            return;
+        }
+        int64_t now_us = butil::gettimeofday_us();
+        if (now_us >= _deadline) {
+            LOG(INFO) << "cleanup brpc stub, endpoint:" << _endpoint << ", idle for " << (now_us - _deadline) / 1000
+                      << "ms past deadline";
+            _cache->_stub_map.erase(_endpoint);
+            return;
+        }
+        auto new_task = std::make_shared<EndpointCleanupTask<StubCacheT>>(_cache, _endpoint, _ttl_seconds);
+        new_task->_deadline = _deadline;
+        if (!_cache->replace_cleanup_task_locked(_endpoint, new_task)) {
+            return;
+        }
+        timespec tm = butil::microseconds_to_timespec(_deadline);
+        auto status = _cache->_timer->schedule(new_task.get(), tm);
+        if (!status.ok()) {
+            LOG(WARNING) << "Failed to reschedule brpc cleanup task: " << _endpoint;
+            // Drop the entry; the next get_*_stub() will recreate it with a fresh task.
+            _cache->_stub_map.erase(_endpoint);
+        }
+    }
+
+    // Reset the absolute deadline (in butil::gettimeofday_us() units) used by the next
+    // Run() invocation to decide between evict and reschedule. Caller must hold the
+    // cache lock.
+    void renew_deadline_locked(int64_t new_deadline) { _deadline = new_deadline; }
+    int64_t deadline_locked() const { return _deadline; }
 
 private:
     StubCacheT* _cache;
     butil::EndPoint _endpoint;
+    // Absolute deadline (in butil::gettimeofday_us() units) used to decide whether a
+    // firing task should evict the stub or simply reschedule itself. Read/written only
+    // under the cache's _lock, so it does not need to be atomic.
+    int64_t _deadline{0};
+    int64_t _ttl_seconds{0};
 };
 
 class BrpcStubCache {
@@ -72,9 +132,29 @@ class BrpcStubCache {
     std::shared_ptr<PInternalService_RecoverableStub> get_stub(const butil::EndPoint& endpoint);
     std::shared_ptr<PInternalService_RecoverableStub> get_stub(const TNetworkAddress& taddr);
     std::shared_ptr<PInternalService_RecoverableStub> get_stub(const std::string& host, int port);
-    void cleanup_expired(const butil::EndPoint& endpoint);
 
 private:
+<<<<<<< HEAD:be/src/util/brpc_stub_cache.h
+=======
+    friend class EndpointCleanupTask<BrpcStubCache>;
+
+    template <typename CacheT, typename ExtractFn>
+    friend void wait_clean_tasks_terminate(CacheT* cache, ExtractFn extract);
+
+    template <typename CacheT>
+    friend void reset_state_for_rebind(CacheT* cache, BthreadTimer* timer);
+
+    bool is_cleanup_task_owner_locked(const butil::EndPoint& endpoint,
+                                      const EndpointCleanupTask<BrpcStubCache>* task) const {
+        auto pool = _stub_map.seek(endpoint);
+        return pool != nullptr && (*pool)->_cleanup_task.get() == task;
+    }
+
+    bool replace_cleanup_task_locked(const butil::EndPoint& endpoint,
+                                     std::shared_ptr<EndpointCleanupTask<BrpcStubCache>> task);
+
+    struct Metrics;
+>>>>>>> 2bbca67281 ([BugFix] Fix bRPC stub cache clean timer leak (#75973)):be/src/common/brpc/brpc_stub_cache.h
     struct StubPool {
         StubPool();
         ~StubPool();
@@ -87,14 +167,19 @@ class BrpcStubCache {
 
     SpinLock _lock;
     butil::FlatMap<butil::EndPoint, std::shared_ptr<StubPool>> _stub_map;
+<<<<<<< HEAD:be/src/util/brpc_stub_cache.h
     pipeline::PipelineTimer* _pipeline_timer;
+=======
+    BthreadTimer* _timer;
+    std::unique_ptr<Metrics> _metrics;
+    bool _stopping{false};
+>>>>>>> 2bbca67281 ([BugFix] Fix bRPC stub cache clean timer leak (#75973)):be/src/common/brpc/brpc_stub_cache.h
 };
 
 class HttpBrpcStubCache {
 public:
     static HttpBrpcStubCache* getInstance();
     StatusOr<std::shared_ptr<PInternalService_RecoverableStub>> get_http_stub(const TNetworkAddress& taddr);
-    void cleanup_expired(const butil::EndPoint& endpoint);
     void shutdown();
 
 private:
@@ -102,19 +187,48 @@ class HttpBrpcStubCache {
     HttpBrpcStubCache(const HttpBrpcStubCache&) = delete;
     HttpBrpcStubCache& operator=(const HttpBrpcStubCache&) = delete;
     ~HttpBrpcStubCache();
+<<<<<<< HEAD:be/src/util/brpc_stub_cache.h
 
     SpinLock _lock;
     butil::FlatMap<butil::EndPoint, std::pair<std::shared_ptr<PInternalService_RecoverableStub>,
                                               std::shared_ptr<EndpointCleanupTask<HttpBrpcStubCache>>>>
             _stub_map;
     pipeline::PipelineTimer* _pipeline_timer;
+=======
+    void bind_timer(BthreadTimer* timer);
+    friend class EndpointCleanupTask<HttpBrpcStubCache>;
+
+    template <typename CacheT, typename ExtractFn>
+    friend void wait_clean_tasks_terminate(CacheT* cache, ExtractFn extract);
+
+    template <typename CacheT>
+    friend void reset_state_for_rebind(CacheT* cache, BthreadTimer* timer);
+
+    bool is_cleanup_task_owner_locked(const butil::EndPoint& endpoint,
+                                      const EndpointCleanupTask<HttpBrpcStubCache>* task) const {
+        auto entry = _stub_map.seek(endpoint);
+        return entry != nullptr && entry->cleanup_task.get() == task;
+    }
+
+    bool replace_cleanup_task_locked(const butil::EndPoint& endpoint,
+                                     std::shared_ptr<EndpointCleanupTask<HttpBrpcStubCache>> task);
+
+    struct StubEntry {
+        std::shared_ptr<PInternalService_RecoverableStub> stub;
+        std::shared_ptr<EndpointCleanupTask<HttpBrpcStubCache>> cleanup_task;
+    };
+
+    SpinLock _lock;
+    butil::FlatMap<butil::EndPoint, StubEntry> _stub_map;
+    BthreadTimer* _timer;
+    bool _stopping{false};
+>>>>>>> 2bbca67281 ([BugFix] Fix bRPC stub cache clean timer leak (#75973)):be/src/common/brpc/brpc_stub_cache.h
 };
 
 class LakeServiceBrpcStubCache {
 public:
     static LakeServiceBrpcStubCache* getInstance();
     StatusOr<std::shared_ptr<starrocks::LakeService_RecoverableStub>> get_stub(const std::string& host, int port);
-    void cleanup_expired(const butil::EndPoint& endpoint);
     void shutdown();
 
 private:
@@ -122,12 +236,42 @@ class LakeServiceBrpcStubCache {
     LakeServiceBrpcStubCache(const LakeServiceBrpcStubCache&) = delete;
     LakeServiceBrpcStubCache& operator=(const LakeServiceBrpcStubCache&) = delete;
     ~LakeServiceBrpcStubCache();
+<<<<<<< HEAD:be/src/util/brpc_stub_cache.h
 
     SpinLock _lock;
     butil::FlatMap<butil::EndPoint, std::pair<std::shared_ptr<LakeService_RecoverableStub>,
                                               std::shared_ptr<EndpointCleanupTask<LakeServiceBrpcStubCache>>>>
             _stub_map;
     pipeline::PipelineTimer* _pipeline_timer;
+=======
+    void bind_timer(BthreadTimer* timer);
+    friend class EndpointCleanupTask<LakeServiceBrpcStubCache>;
+
+    template <typename CacheT, typename ExtractFn>
+    friend void wait_clean_tasks_terminate(CacheT* cache, ExtractFn extract);
+
+    template <typename CacheT>
+    friend void reset_state_for_rebind(CacheT* cache, BthreadTimer* timer);
+
+    bool is_cleanup_task_owner_locked(const butil::EndPoint& endpoint,
+                                      const EndpointCleanupTask<LakeServiceBrpcStubCache>* task) const {
+        auto entry = _stub_map.seek(endpoint);
+        return entry != nullptr && entry->cleanup_task.get() == task;
+    }
+
+    bool replace_cleanup_task_locked(const butil::EndPoint& endpoint,
+                                     std::shared_ptr<EndpointCleanupTask<LakeServiceBrpcStubCache>> task);
+
+    struct StubEntry {
+        std::shared_ptr<LakeService_RecoverableStub> stub;
+        std::shared_ptr<EndpointCleanupTask<LakeServiceBrpcStubCache>> cleanup_task;
+    };
+
+    SpinLock _lock;
+    butil::FlatMap<butil::EndPoint, StubEntry> _stub_map;
+    BthreadTimer* _timer;
+    bool _stopping{false};
+>>>>>>> 2bbca67281 ([BugFix] Fix bRPC stub cache clean timer leak (#75973)):be/src/common/brpc/brpc_stub_cache.h
 };
 
 } // namespace starrocks
diff --git a/be/test/util/brpc_stub_cache_test.cpp b/be/test/util/brpc_stub_cache_test.cpp
index bf456c3bbbfb8b..ab27f1abb44456 100644
--- a/be/test/util/brpc_stub_cache_test.cpp
+++ b/be/test/util/brpc_stub_cache_test.cpp
@@ -171,4 +171,142 @@ TEST_F(BrpcStubCacheTest, test_http_cleanup) {
     ASSERT_NE(*stub3, *stub1);
 }
 
+<<<<<<< HEAD:be/test/util/brpc_stub_cache_test.cpp
+=======
+// Regression test: destroying BrpcStubCache while a cleanup task is scheduled
+// (and possibly firing) must join the task before the cache state is torn down.
+TEST_F(BrpcStubCacheTest, test_destructor_joins_inflight_cleanup_tasks) {
+    config::brpc_stub_expire_s = 1;
+    auto cache = std::make_unique<BrpcStubCache>(_timer.get());
+    TNetworkAddress address;
+    address.hostname = "127.0.0.1";
+    address.port = 123;
+    auto stub = cache->get_stub(address);
+    ASSERT_NE(nullptr, stub);
+
+    // Trigger ~BrpcStubCache() while the cleanup task is (or will be) in flight.
+    // Drain the cache and assert the unique_ptr release returns cleanly.
+    cache.reset();
+
+    // Reacquire the endpoint through a fresh cache; the slot must have been
+    // cleanly torn down without leaking the previous task.
+    auto cache2 = std::make_unique<BrpcStubCache>(_timer.get());
+    auto fresh_stub = cache2->get_stub(address);
+    ASSERT_NE(nullptr, fresh_stub);
+    cache2.reset();
+}
+
+// Regression test for the lazy-reschedule fix: while an endpoint is being
+// accessed within the expire window, the timer fires, sees the stub is still
+// active (idle < brpc_stub_expire_s), and reschedules instead of evicting, so the
+// stub must survive across multiple timer periods.
+TEST_F(BrpcStubCacheTest, test_active_access_keeps_stub_alive_across_expire_window) {
+    config::brpc_stub_expire_s = 2;
+    BrpcStubCache cache(_timer.get());
+    TNetworkAddress address;
+    address.hostname = "127.0.0.1";
+    address.port = 123;
+    auto stub1 = cache.get_stub(address);
+    ASSERT_NE(nullptr, stub1);
+
+    // Repeatedly access within the window; each access refreshes the last-access
+    // time, and the single scheduled timer reschedules instead of evicting.
+    for (int i = 0; i < 3; ++i) {
+        sleep(1);
+        auto stub = cache.get_stub(address);
+        ASSERT_EQ(stub1, stub) << "stub must not be evicted while being accessed";
+    }
+}
+
+TEST_F(BrpcStubCacheTest, test_http_active_access_keeps_stub_alive_across_expire_window) {
+    config::brpc_stub_expire_s = 2;
+    HttpBrpcStubCache cache(_timer.get());
+    TNetworkAddress address;
+    address.hostname = "127.0.0.1";
+    address.port = 123;
+    auto stub1 = cache.get_http_stub(address);
+    ASSERT_NE(nullptr, *stub1);
+
+    for (int i = 0; i < 3; ++i) {
+        sleep(1);
+        auto stub = cache.get_http_stub(address);
+        ASSERT_NE(nullptr, *stub);
+        ASSERT_EQ(*stub1, *stub) << "http stub must not be evicted while being accessed";
+    }
+}
+
+#ifndef __APPLE__
+TEST_F(BrpcStubCacheTest, test_lake_active_access_keeps_stub_alive_across_expire_window) {
+    config::brpc_stub_expire_s = 2;
+    LakeServiceBrpcStubCache cache(_timer.get());
+    std::string hostname = "127.0.0.1";
+    int32_t port = 123;
+    auto stub1 = cache.get_stub(hostname, port);
+    ASSERT_TRUE(stub1.ok());
+    ASSERT_NE(nullptr, *stub1);
+
+    for (int i = 0; i < 3; ++i) {
+        sleep(1);
+        auto stub = cache.get_stub(hostname, port);
+        ASSERT_TRUE(stub.ok());
+        ASSERT_NE(nullptr, *stub);
+        ASSERT_EQ(*stub1, *stub) << "lake stub must not be evicted while being accessed";
+    }
+}
+#endif
+
+TEST_F(BrpcStubCacheTest, http_singleton_reinitialize_rebinds_pipeline_timer) {
+    auto timer2 = std::make_unique<BthreadTimer>();
+    ASSERT_OK(timer2->start());
+
+    HttpBrpcStubCache::initialize(_timer.get());
+    auto* cache = HttpBrpcStubCache::getInstance();
+    ASSERT_NE(nullptr, cache);
+
+    TNetworkAddress address;
+    address.hostname = "127.0.0.1";
+    address.port = 123;
+    auto stub = cache->get_http_stub(address);
+    ASSERT_TRUE(stub.ok());
+    ASSERT_NE(nullptr, *stub);
+
+    cache->shutdown();
+    ASSERT_FALSE(cache->get_http_stub(address).ok());
+
+    HttpBrpcStubCache::initialize(timer2.get());
+
+    auto rebound_stub = cache->get_http_stub(address);
+    ASSERT_TRUE(rebound_stub.ok());
+    ASSERT_NE(nullptr, *rebound_stub);
+
+    cache->shutdown();
+}
+
+#ifndef __APPLE__
+TEST_F(BrpcStubCacheTest, lake_singleton_reinitialize_rebinds_pipeline_timer) {
+    auto timer2 = std::make_unique<BthreadTimer>();
+    ASSERT_OK(timer2->start());
+
+    LakeServiceBrpcStubCache::initialize(_timer.get());
+    auto* cache = LakeServiceBrpcStubCache::getInstance();
+    ASSERT_NE(nullptr, cache);
+
+    auto stub = cache->get_stub("127.0.0.1", 123);
+    ASSERT_TRUE(stub.ok());
+    ASSERT_NE(nullptr, *stub);
+
+    cache->shutdown();
+    ASSERT_FALSE(cache->get_stub("127.0.0.1", 123).ok());
+
+    LakeServiceBrpcStubCache::initialize(timer2.get());
+
+    auto rebound_stub = cache->get_stub("127.0.0.1", 123);
+    ASSERT_TRUE(rebound_stub.ok());
+    ASSERT_NE(nullptr, *rebound_stub);
+
+    cache->shutdown();
+}
+#endif
+
+>>>>>>> 2bbca67281 ([BugFix] Fix bRPC stub cache clean timer leak (#75973)):be/test/common/brpc/brpc_stub_cache_test.cpp
 } // namespace starrocks