[fix](be) Fix row binlog memtable flush backpressure

be/src/cloud/cloud_delta_writer.cpp

@@ -99,8 +99,9 @@ Status CloudDeltaWriter::write(const Block* block, const TabletAddRowsPayload& r
     CHECK(_is_init || _is_cancelled);
     {
         SCOPED_TIMER(_wait_flush_limit_timer);
-        while (_memtable_writer->flush_running_count() >=
-               config::memtable_flush_running_count_limit) {
+        auto* adaptive_controller =
+                ExecEnv::GetInstance()->storage_engine().adaptive_thread_controller();
+        while (adaptive_controller->is_io_busy()) {
             std::this_thread::sleep_for(std::chrono::milliseconds(10));
         }
     }

be/src/load/delta_writer/delta_writer.cpp

@@ -178,8 +178,10 @@ Status DeltaWriter::write(const Block* block, const TabletAddRowsPayload& rows,
     }
     {
         SCOPED_TIMER(_wait_flush_limit_timer);
-        while (_memtable_writer->flush_running_count() >=
-               config::memtable_flush_running_count_limit) {
+        const auto effective_flush_running_count_limit =
+                config::memtable_flush_running_count_limit *
+                (_req.write_req_type == WriteRequestType::GROUP ? 2 : 1);
+        while (_memtable_writer->flush_running_count() >= effective_flush_running_count_limit) {
             std::this_thread::sleep_for(std::chrono::milliseconds(10));
         }
     }

be/src/storage/storage_engine.cpp

@@ -140,15 +140,17 @@ int64_t BaseStorageEngine::memory_limitation_bytes_per_thread_for_schema_change(
 }
 
 void BaseStorageEngine::_start_adaptive_thread_controller() {
-    if (!config::enable_adaptive_flush_threads) {
-        return;
-    }
-
     auto* system_metrics = DorisMetrics::instance()->system_metrics();
     auto* s3_upload_pool = ExecEnv::GetInstance()->s3_file_upload_thread_pool();
 
     _adaptive_thread_controller.init(system_metrics, s3_upload_pool);
 
+    // Cloud write backpressure also uses is_io_busy(), so initialize the IO signal even when
+    // adaptive flush-thread adjustment is disabled.
+    if (!config::enable_adaptive_flush_threads) {
+        return;
+    }
+
     if (_memtable_flush_executor) {
         auto* flush_pool = _memtable_flush_executor->flush_pool();
         auto* high_prio_pool = _memtable_flush_executor->high_prio_flush_pool();

be/test/storage/adaptive_thread_pool_controller_test.cpp

@@ -22,10 +22,12 @@
 #include <algorithm>
 #include <thread>
 
+#include "cloud/config.h"
 #include "common/config.h"
 #include "common/metrics/metrics.h"
 #include "common/metrics/system_metrics.h"
 #include "testutil/test_util.h"
+#include "util/countdown_latch.h"
 #include "util/threadpool.h"
 
 namespace doris {
@@ -44,6 +46,8 @@ class AdaptiveThreadPoolControllerTest : public testing::Test {
 protected:
     void SetUp() override {
         _original_enable_adaptive = config::enable_adaptive_flush_threads;
+        _original_deploy_mode = config::deploy_mode;
+        _original_cloud_unique_id = config::cloud_unique_id;
 
         int num_cpus = std::thread::hardware_concurrency();
         if (num_cpus <= 0) num_cpus = 1;
@@ -64,11 +68,15 @@ class AdaptiveThreadPoolControllerTest : public testing::Test {
 
     void TearDown() override {
         config::enable_adaptive_flush_threads = _original_enable_adaptive;
+        config::deploy_mode = _original_deploy_mode;
+        config::cloud_unique_id = _original_cloud_unique_id;
         if (_pool) _pool->shutdown();
         if (_pool2) _pool2->shutdown();
     }
 
     bool _original_enable_adaptive;
+    std::string _original_deploy_mode;
+    std::string _original_cloud_unique_id;
     std::unique_ptr<ThreadPool> _pool;
     std::unique_ptr<ThreadPool> _pool2;
 };
@@ -260,6 +268,44 @@ TEST_F(AdaptiveThreadPoolControllerTest, TestIoBusyCpuBusyWithNullMetrics) {
     EXPECT_FALSE(controller.is_cpu_busy());
 }
 
+TEST_F(AdaptiveThreadPoolControllerTest, TestCloudIoBusyUsesS3UploadQueue) {
+    config::deploy_mode = "cloud";
+    config::cloud_unique_id.clear();
+
+    std::unique_ptr<ThreadPool> upload_pool;
+    ASSERT_TRUE(ThreadPoolBuilder("TestS3UploadPool")
+                        .set_min_threads(1)
+                        .set_max_threads(1)
+                        .set_max_queue_size(AdaptiveThreadPoolController::kS3QueueBusyThreshold + 1)
+                        .build(&upload_pool)
+                        .ok());
+
+    CountDownLatch task_started(1);
+    CountDownLatch release_task(1);
+    ASSERT_TRUE(upload_pool
+                        ->submit_func([&]() {
+                            task_started.count_down();
+                            release_task.wait();
+                        })
+                        .ok());
+    EXPECT_TRUE(task_started.wait_for(std::chrono::seconds(5)));
+
+    for (int i = 0; i < AdaptiveThreadPoolController::kS3QueueBusyThreshold; ++i) {
+        EXPECT_TRUE(upload_pool->submit_func([]() {}).ok());
+    }
+
+    AdaptiveThreadPoolController controller;
+    controller.init(nullptr, upload_pool.get());
+    EXPECT_FALSE(controller.is_io_busy());
+
+    EXPECT_TRUE(upload_pool->submit_func([]() {}).ok());
+    EXPECT_TRUE(controller.is_io_busy());
+
+    release_task.count_down();
+    upload_pool->shutdown();
+    upload_pool->wait();
+}
+
 TEST_F(AdaptiveThreadPoolControllerTest, TestCpuBusyUsesCpuMetricsDelta) {
     MetricRegistry registry("test");
     const std::string before_path = get_stat_test_data_path("stat_cpu_busy_before");