[fix](cloud) Drain txn lazy committer workers before destruction (#63876)

## What
Fix shutdown ordering in `TxnLazyCommitter` by explicitly stopping
worker pools before member destruction can invalidate state used by
worker callbacks.

## Why
Lazy commit worker jobs keep a back pointer to `TxnLazyCommitter` and
call back into `remove()`. They can also access the parallel commit pool
and resource manager during `commit()`. With the default destructor,
`running_tasks_`, `mutex_`, and `parallel_commit_pool_` are destroyed
before `worker_pool_` is joined, which can lead to shutdown-time
use-after-destruction.

## How
- Add an explicit `TxnLazyCommitter` destructor.
- Mark the committer as stopped before draining workers.
- Stop and join the lazy commit worker pool before destroying task
tracking state.
- Stop the parallel commit pool after lazy workers are quiesced.
- Make failed or post-shutdown submissions complete with an error
instead of leaving waiters blocked.

## Tests
- `sh format_code.sh cloud/src/meta-service/txn_lazy_committer.h`
- `sh format_code.sh cloud/src/meta-service/txn_lazy_committer.cpp`
- `sh run-cloud-ut.sh --run --fdb
"fdb_cluster0:cluster0@10.26.20.4:4500"`
  - Build passed.
  - `txn_lazy_commit_test` passed 24/24 in the full run.
- The full run had unrelated storage vault/HDFS failures in
`meta_service_test`.
- After tightening the submit/shutdown race:
- `sh run-cloud-ut.sh --run --fdb
"fdb_cluster0:cluster0@10.26.20.4:4500" --filter
"txn_lazy_commit_test:*.*"`
- Build passed; 22/24 passed, 2 tests failed due FDB `Timeout` while
committing setup transactions.

Co-authored-by: gavinchou <gavinchou@apache.org>

Author: gavinchou

cloud/src/meta-service/txn_lazy_committer.cpp

@@ -633,14 +633,20 @@ TxnLazyCommitTask::TxnLazyCommitTask(const std::string& instance_id, int64_t txn
     DCHECK(txn_id > 0);
 }
 
+void TxnLazyCommitTask::finish(MetaServiceCode code, std::string msg) {
+    {
+        std::unique_lock lock(mutex_);
+        finished_ = true;
+        code_ = code;
+        msg_ = std::move(msg);
+    }
+    cond_.notify_all();
+}
+
 void TxnLazyCommitTask::commit() {
     StopWatch sw;
     DORIS_CLOUD_DEFER {
-        {
-            std::unique_lock lock(mutex_);
-            this->finished_ = true;
-        }
-        this->cond_.notify_all();
+        finish(code_, msg_);
         g_bvar_txn_lazy_committer_committing_duration << sw.elapsed_us();
     };
 
@@ -965,6 +971,25 @@ TxnLazyCommitter::TxnLazyCommitter(std::shared_ptr<TxnKv> txn_kv,
     parallel_commit_pool_->start();
 }
 
+TxnLazyCommitter::~TxnLazyCommitter() {
+    {
+        std::unique_lock<std::mutex> lock(mutex_);
+        stopped_ = true;
+    }
+
+    if (worker_pool_ != nullptr) {
+        worker_pool_->stop();
+    }
+    if (parallel_commit_pool_ != nullptr) {
+        parallel_commit_pool_->stop();
+    }
+
+    {
+        std::unique_lock<std::mutex> lock(mutex_);
+        running_tasks_.clear();
+    }
+}
+
 /**
  * @brief Submit a lazy commit txn task
  * 
@@ -978,6 +1003,12 @@ std::shared_ptr<TxnLazyCommitTask> TxnLazyCommitter::submit(const std::string& i
     std::shared_ptr<TxnLazyCommitTask> task;
     {
         std::unique_lock<std::mutex> lock(mutex_);
+        if (stopped_) {
+            task = std::make_shared<TxnLazyCommitTask>(instance_id, txn_id, txn_kv_, this);
+            task->finish(MetaServiceCode::UNDEFINED_ERR, "txn lazy committer is stopped");
+            return task;
+        }
+
         auto iter = running_tasks_.find(txn_id);
         if (iter != running_tasks_.end()) {
             return iter->second;
@@ -986,13 +1017,17 @@ std::shared_ptr<TxnLazyCommitTask> TxnLazyCommitter::submit(const std::string& i
         task = std::make_shared<TxnLazyCommitTask>(instance_id, txn_id, txn_kv_, this);
         running_tasks_.emplace(txn_id, task);
         g_bvar_txn_lazy_committer_submitted << 1;
-    }
 
-    worker_pool_->submit([task]() {
-        task->commit();
-        task->txn_lazy_committer_->remove(task->txn_id_);
-        g_bvar_txn_lazy_committer_finished << 1;
-    });
+        int ret = worker_pool_->submit([task]() {
+            task->commit();
+            task->txn_lazy_committer_->remove(task->txn_id_);
+            g_bvar_txn_lazy_committer_finished << 1;
+        });
+        if (ret != 0) {
+            running_tasks_.erase(txn_id);
+            task->finish(MetaServiceCode::UNDEFINED_ERR, "failed to submit txn lazy commit task");
+        }
+    }
     DCHECK(task != nullptr);
     return task;
 }

cloud/src/meta-service/txn_lazy_committer.h

@@ -45,6 +45,11 @@ class TxnLazyCommitTask {
 private:
     friend class TxnLazyCommitter;
 
+    // Marks the task as finished with the final result and wakes all waiters.
+    // `code` is returned by wait() as the task status, and `msg` carries the
+    // corresponding error detail or an empty string on success.
+    void finish(MetaServiceCode code, std::string msg);
+
     std::pair<MetaServiceCode, std::string> commit_partition(
             int64_t db_id, int64_t partition_id,
             const std::vector<std::pair<std::string, doris::RowsetMetaCloudPB>>& tmp_rowset_metas,
@@ -66,6 +71,7 @@ class TxnLazyCommitter {
 public:
     TxnLazyCommitter(std::shared_ptr<TxnKv> txn_kv);
     TxnLazyCommitter(std::shared_ptr<TxnKv> txn_kv, std::shared_ptr<ResourceManager> resource_mgr);
+    ~TxnLazyCommitter();
     std::shared_ptr<TxnLazyCommitTask> submit(const std::string& instance_id, int64_t txn_id);
     void remove(int64_t txn_id);
 
@@ -82,5 +88,6 @@ class TxnLazyCommitter {
     std::mutex mutex_;
     // <txn_id, TxnLazyCommitTask>
     std::unordered_map<int64_t, std::shared_ptr<TxnLazyCommitTask>> running_tasks_;
+    bool stopped_ = false;
 };
-} // namespace doris::cloud
+} // namespace doris::cloud