diff --git a/be/src/cloud/cloud_storage_engine.cpp b/be/src/cloud/cloud_storage_engine.cpp index a8cebc7449da34..c363aecf50f9c9 100644 --- a/be/src/cloud/cloud_storage_engine.cpp +++ b/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) { diff --git a/be/test/cloud/cloud_compaction_test.cpp b/be/test/cloud/cloud_compaction_test.cpp index f435997e5de89a..f807b4c8678e8b 100644 --- a/be/test/cloud/cloud_compaction_test.cpp +++ b/be/test/cloud/cloud_compaction_test.cpp @@ -21,6 +21,7 @@ #include #include +#include #include #include #include @@ -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(*_tablet_meta); + tablet_meta->_tablet_id = 12000; + auto tablet = std::make_shared(_engine, tablet_meta); + + std::vector 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(outcome); + result->second = true; + result->first = Status::InternalError("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); diff --git a/regression-test/suites/fault_injection_p0/cloud/test_cloud_cumu_compaction_global_lock_thread_count.groovy b/regression-test/suites/fault_injection_p0/cloud/test_cloud_cumu_compaction_global_lock_thread_count.groovy new file mode 100644 index 00000000000000..40ce148850f708 --- /dev/null +++ b/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) + } + } +}