[fix](compaction): avoid decrementing uncounted cumulative compaction threads

be/src/cloud/cloud_storage_engine.cpp

@@ -1184,12 +1184,13 @@ Status CloudStorageEngine::_submit_cumulative_compaction_task(const CloudTabletS
         signal::tablet_id = tablet->tablet_id();
         g_cumu_compaction_running_task_count << 1;
         bool is_large_task = true;
+        bool cumu_thread_counted = false;
         Defer defer {[&]() {
             DBUG_EXECUTE_IF("CloudStorageEngine._submit_cumulative_compaction_task.sleep",
                             { sleep(5); })
             // Idempotent cleanup: remove task from tracker
             CompactionTaskTracker::instance()->remove_task(compaction_id);
-            if (compaction_type == CompactionType::CUMULATIVE_COMPACTION) {
+            if (cumu_thread_counted) {
                 std::lock_guard lock(_cumu_compaction_delay_mtx);
                 _cumu_compaction_thread_pool_used_threads--;
                 if (!is_large_task) {
@@ -1224,6 +1225,7 @@ Status CloudStorageEngine::_submit_cumulative_compaction_task(const CloudTabletS
             }
             std::lock_guard lock(_cumu_compaction_delay_mtx);
             _cumu_compaction_thread_pool_used_threads++;
+            cumu_thread_counted = true;
             if (config::large_cumu_compaction_task_min_thread_num > 1 &&
                 _cumu_compaction_thread_pool->max_threads() >=
                         config::large_cumu_compaction_task_min_thread_num) {

be/test/cloud/cloud_compaction_test.cpp

@@ -21,6 +21,7 @@
 #include <gtest/gtest-test-part.h>
 #include <gtest/gtest.h>
 
+#include <chrono>
 #include <memory>
 #include <mutex>
 #include <string_view>
@@ -46,6 +47,7 @@
 #include "storage/tablet/tablet_meta.h"
 #include "util/defer_op.h"
 #include "util/s3_util.h"
+#include "util/threadpool.h"
 #include "util/time.h"
 #include "util/uid_util.h"
 
@@ -411,6 +413,51 @@ static RowsetSharedPtr create_rowset(Version version, int num_segments, bool ove
     return rowset;
 }
 
+TEST_F(CloudCompactionTest, cumulative_global_lock_failure_keeps_thread_count_balanced) {
+    auto tablet_meta = std::make_shared<TabletMeta>(*_tablet_meta);
+    tablet_meta->_tablet_id = 12000;
+    auto tablet = std::make_shared<CloudTablet>(_engine, tablet_meta);
+
+    std::vector<RowsetSharedPtr> rowsets;
+    for (int64_t version = 0; version < config::cumulative_compaction_min_deltas; ++version) {
+        auto rowset = create_rowset(Version(version, version), 3, false, 16 * 1024 * 1024);
+        ASSERT_NE(rowset, nullptr);
+        rowsets.push_back(std::move(rowset));
+    }
+    {
+        std::unique_lock lock(tablet->get_header_lock());
+        tablet->add_rowsets(std::move(rowsets), false, lock);
+    }
+    tablet->last_sync_time_s = 1;
+    tablet->_approximate_num_rowsets = 0;
+
+    ASSERT_TRUE(ThreadPoolBuilder("CumuCompactionTaskThreadPoolTest")
+                        .set_min_threads(1)
+                        .set_max_threads(1)
+                        .build(&_engine._cumu_compaction_thread_pool)
+                        .ok());
+
+    auto* sync_point = SyncPoint::get_instance();
+    sync_point->enable_processing();
+    sync_point->set_call_back("CloudMetaMgr::prepare_tablet_job", [](auto&& outcome) {
+        auto* result = try_any_cast_ret<Status>(outcome);
+        result->second = true;
+        result->first = Status::InternalError<false>("mock global compaction lock failure");
+    });
+    Defer clear_sync_point {[&] {
+        sync_point->clear_all_call_backs();
+        sync_point->disable_processing();
+    }};
+
+    ASSERT_EQ(_engine._cumu_compaction_thread_pool_used_threads, 0);
+    Status status = _engine.submit_compaction_task(tablet, CompactionType::CUMULATIVE_COMPACTION);
+    ASSERT_TRUE(status.ok()) << status.to_string();
+    ASSERT_TRUE(_engine._cumu_compaction_thread_pool->wait_for(std::chrono::seconds(5)));
+
+    EXPECT_FALSE(_engine.has_cumu_compaction(tablet->tablet_id()));
+    EXPECT_EQ(_engine._cumu_compaction_thread_pool_used_threads, 0);
+}
+
 static RowsetSharedPtr create_delete_rowset(Version version) {
     auto rowset = create_rowset(version, 0, false, 0);
     DORIS_CHECK(rowset != nullptr);

regression-test/suites/fault_injection_p0/cloud/test_cloud_cumu_compaction_global_lock_thread_count.groovy

@@ -0,0 +1,186 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+import groovy.json.JsonSlurper
+import org.apache.doris.regression.suite.ClusterOptions
+import org.apache.doris.regression.util.DebugPoint
+import org.apache.doris.regression.util.NodeType
+
+suite('test_cloud_cumu_compaction_global_lock_thread_count', 'docker') {
+    def options = new ClusterOptions()
+    options.cloudMode = true
+    options.enableDebugPoints()
+    options.beConfigs += [
+        'enable_java_support=false',
+        'cumulative_compaction_min_deltas=2',
+        'cumulative_compaction_max_deltas=3',
+        'max_cumu_compaction_threads=2',
+        'large_cumu_compaction_task_min_thread_num=2',
+        'large_cumu_compaction_task_row_num_threshold=1',
+        'disable_auto_compaction=true',
+    ]
+    options.beNum = 3
+
+    docker(options) {
+        GetDebugPoint().clearDebugPointsForAllBEs()
+        def backends = sql_return_maparray('SHOW BACKENDS')
+        def lockHolderBe = backends[0]
+        def lockFailedBe = backends[1]
+        assertNotNull(lockHolderBe)
+        assertNotNull(lockFailedBe)
+
+        def getCompactionStatus = { be, tabletId ->
+            def (code, out, err) = be_get_compaction_status(be.Host, be.HttpPort, tabletId)
+            assertEquals(0, code, "failed to get compaction status: ${out}, ${err}")
+            return new JsonSlurper().parseText(out.trim())
+        }
+
+        def waitForCompactionRunning = { be, tabletId ->
+            def deadline = System.currentTimeMillis() + 30000
+            while (System.currentTimeMillis() < deadline) {
+                if (getCompactionStatus(be, tabletId).run_status) {
+                    return
+                }
+                sleep(100)
+            }
+            assertTrue(false, "compaction did not start for tablet ${tabletId}")
+        }
+
+        def waitForCompactionFinished = { be, tabletId ->
+            def deadline = System.currentTimeMillis() + 30000
+            while (System.currentTimeMillis() < deadline) {
+                if (!getCompactionStatus(be, tabletId).run_status) {
+                    return
+                }
+                sleep(100)
+            }
+            assertTrue(false, "compaction did not finish for tablet ${tabletId}")
+        }
+
+        def getTabletStatus = { be, tabletId ->
+            def (code, out, err) = be_show_tablet_status(be.Host, be.HttpPort, tabletId)
+            assertEquals(0, code, "failed to get tablet status: ${out}, ${err}")
+            return new JsonSlurper().parseText(out.trim())
+        }
+
+        def getRowsetCount = { be, tabletId ->
+            def status = getTabletStatus(be, tabletId)
+            assertTrue(status.rowsets instanceof List)
+            return status.rowsets.size()
+        }
+
+        def createTable = { table ->
+            sql "DROP TABLE IF EXISTS ${table} FORCE"
+            sql """
+                CREATE TABLE ${table} (
+                    k INT,
+                    v INT
+                ) DUPLICATE KEY(k)
+                DISTRIBUTED BY HASH(k) BUCKETS 1
+                PROPERTIES (
+                    "replication_num" = "1",
+                    "disable_auto_compaction" = "true"
+                )
+            """
+            for (int value = 0; value < 5; value++) {
+                sql "INSERT INTO ${table} VALUES (${value}, ${value})"
+            }
+            sql "SELECT COUNT(*) FROM ${table}"
+            def tablet = sql_return_maparray("SHOW TABLETS FROM ${table}")[0]
+            return tablet.TabletId
+        }
+
+        def runCumulativeCompaction = { be, tabletId ->
+            def (code, out, err) = be_run_cumulative_compaction(be.Host, be.HttpPort, tabletId)
+            assertEquals(0, code, "failed to submit cumulative compaction: ${out}, ${err}")
+        }
+
+        def conflictTabletId = createTable('test_cloud_cumu_compaction_global_lock_conflict')
+        def conflictRowsetsBefore = getRowsetCount(lockFailedBe, conflictTabletId)
+        def holderRowsetsBefore = getRowsetCount(lockHolderBe, conflictTabletId)
+        def blockModifyRowsets = 'CloudCumulativeCompaction::modify_rowsets.enable_spin_wait'
+        def blockModifyRowsetsSwitch = 'CloudCumulativeCompaction::modify_rowsets.block'
+        def holdTaskAfterExecution = 'CloudStorageEngine._submit_cumulative_compaction_task.sleep'
+
+        try {
+            DebugPoint.enableDebugPoint(lockHolderBe.Host, lockHolderBe.HttpPort.toInteger(),
+                    NodeType.BE, blockModifyRowsets)
+            DebugPoint.enableDebugPoint(lockHolderBe.Host, lockHolderBe.HttpPort.toInteger(),
+                    NodeType.BE, blockModifyRowsetsSwitch)
+
+            runCumulativeCompaction(lockHolderBe, conflictTabletId)
+            waitForCompactionRunning(lockHolderBe, conflictTabletId)
+
+            runCumulativeCompaction(lockFailedBe, conflictTabletId)
+            sleep(1000)
+            waitForCompactionFinished(lockFailedBe, conflictTabletId)
+            assertEquals(conflictRowsetsBefore, getRowsetCount(lockFailedBe, conflictTabletId))
+
+            DebugPoint.disableDebugPoint(lockHolderBe.Host, lockHolderBe.HttpPort.toInteger(),
+                    NodeType.BE, blockModifyRowsetsSwitch)
+            DebugPoint.disableDebugPoint(lockHolderBe.Host, lockHolderBe.HttpPort.toInteger(),
+                    NodeType.BE, blockModifyRowsets)
+            waitForCompactionFinished(lockHolderBe, conflictTabletId)
+            assertTrue(getRowsetCount(lockHolderBe, conflictTabletId) < holderRowsetsBefore)
+
+            def runningTabletId = createTable('test_cloud_cumu_compaction_global_lock_thread_holder')
+            def candidateTabletId = createTable('test_cloud_cumu_compaction_global_lock_thread_candidate')
+            def candidateRowsetsBefore = getRowsetCount(lockFailedBe, candidateTabletId)
+            def runningRowsetsBefore = getRowsetCount(lockFailedBe, runningTabletId)
+
+            DebugPoint.enableDebugPoint(lockFailedBe.Host, lockFailedBe.HttpPort.toInteger(),
+                    NodeType.BE, holdTaskAfterExecution)
+            DebugPoint.enableDebugPoint(lockFailedBe.Host, lockFailedBe.HttpPort.toInteger(),
+                    NodeType.BE, blockModifyRowsets)
+            DebugPoint.enableDebugPoint(lockFailedBe.Host, lockFailedBe.HttpPort.toInteger(),
+                    NodeType.BE, blockModifyRowsetsSwitch)
+
+            runCumulativeCompaction(lockFailedBe, runningTabletId)
+            waitForCompactionRunning(lockFailedBe, runningTabletId)
+
+            DebugPoint.disableDebugPoint(lockFailedBe.Host, lockFailedBe.HttpPort.toInteger(),
+                    NodeType.BE, blockModifyRowsetsSwitch)
+            DebugPoint.disableDebugPoint(lockFailedBe.Host, lockFailedBe.HttpPort.toInteger(),
+                    NodeType.BE, blockModifyRowsets)
+            def deadline = System.currentTimeMillis() + 30000
+            while (System.currentTimeMillis() < deadline &&
+                    getRowsetCount(lockFailedBe, runningTabletId) >= runningRowsetsBefore) {
+                sleep(100)
+            }
+            assertTrue(getRowsetCount(lockFailedBe, runningTabletId) < runningRowsetsBefore)
+            assertTrue(getCompactionStatus(lockFailedBe, runningTabletId).run_status)
+
+            runCumulativeCompaction(lockFailedBe, candidateTabletId)
+            sleep(1000)
+            assertEquals(candidateRowsetsBefore, getRowsetCount(lockFailedBe, candidateTabletId),
+                    'a second large compaction must be delayed while the first task is active')
+            waitForCompactionFinished(lockFailedBe, candidateTabletId)
+            waitForCompactionFinished(lockFailedBe, runningTabletId)
+        } finally {
+            DebugPoint.disableDebugPoint(lockHolderBe.Host, lockHolderBe.HttpPort.toInteger(),
+                    NodeType.BE, blockModifyRowsetsSwitch)
+            DebugPoint.disableDebugPoint(lockHolderBe.Host, lockHolderBe.HttpPort.toInteger(),
+                    NodeType.BE, blockModifyRowsets)
+            DebugPoint.disableDebugPoint(lockFailedBe.Host, lockFailedBe.HttpPort.toInteger(),
+                    NodeType.BE, holdTaskAfterExecution)
+            DebugPoint.disableDebugPoint(lockFailedBe.Host, lockFailedBe.HttpPort.toInteger(),
+                    NodeType.BE, blockModifyRowsetsSwitch)
+            DebugPoint.disableDebugPoint(lockFailedBe.Host, lockFailedBe.HttpPort.toInteger(),
+                    NodeType.BE, blockModifyRowsets)
+        }
+    }
+}