diff --git a/cloud/src/recycler/recycler.cpp b/cloud/src/recycler/recycler.cpp index bd1b9bff878cee..6a87684b8f5b1e 100644 --- a/cloud/src/recycler/recycler.cpp +++ b/cloud/src/recycler/recycler.cpp @@ -6382,7 +6382,7 @@ int InstanceRecycler::recycle_expired_txn_label() { std::string end_recycle_txn_key; recycle_txn_key(recycle_txn_key_info0, &begin_recycle_txn_key); recycle_txn_key(recycle_txn_key_info1, &end_recycle_txn_key); - std::vector recycle_txn_info_keys; + std::unordered_map> recycle_txn_keys_by_label; LOG_WARNING("begin to recycle expired txn").tag("instance_id", instance_id_); @@ -6423,7 +6423,17 @@ int InstanceRecycler::recycle_expired_txn_label() { current_time_ms)) { VLOG_DEBUG << "found recycle txn, key=" << hex(k); num_expired++; - recycle_txn_info_keys.emplace_back(k); + + std::string_view k1 = k; + k1.remove_prefix(1); // Remove key space + std::vector, int, int>> out; + if (decode_key(&k1, &out) != 0) { + LOG_ERROR("failed to decode key").tag("key", hex(k)); + return -1; + } + int64_t db_id = std::get(std::get<0>(out[3])); + auto label_key = txn_label_key({instance_id_, db_id, recycle_txn_pb.label()}); + recycle_txn_keys_by_label[label_key].emplace_back(k); } return 0; }; @@ -6532,35 +6542,48 @@ int InstanceRecycler::recycle_expired_txn_label() { auto loop_done = [&]() -> int { DORIS_CLOUD_DEFER { - recycle_txn_info_keys.clear(); + recycle_txn_keys_by_label.clear(); }; TEST_SYNC_POINT_CALLBACK( - "InstanceRecycler::recycle_expired_txn_label.check_recycle_txn_info_keys", - &recycle_txn_info_keys); - for (const auto& k : recycle_txn_info_keys) { - concurrent_delete_executor.add([&]() { - int ret = delete_recycle_txn_kv(k); - if (ret == 1) { - const int max_retry = std::max(1, config::recycle_txn_delete_max_retry_times); - for (int i = 1; i <= max_retry; ++i) { - LOG(WARNING) << "txn conflict, retry times=" << i << " key=" << hex(k); - ret = delete_recycle_txn_kv(k); - // clang-format off - TEST_SYNC_POINT_CALLBACK( - "InstanceRecycler::recycle_expired_txn_label.delete_recycle_txn_kv_error", &ret); - // clang-format off - if (ret != 1) { - break; - } - // random sleep 0-100 ms to retry - std::this_thread::sleep_for(std::chrono::milliseconds(rand() % 100)); + "InstanceRecycler::recycle_expired_txn_label.check_recycle_txn_keys_by_label", + &recycle_txn_keys_by_label); + auto delete_recycle_txn_kv_with_retry = [&](const std::string& k) -> int { + int ret = delete_recycle_txn_kv(k); + TEST_SYNC_POINT_CALLBACK( + "InstanceRecycler::recycle_expired_txn_label.delete_recycle_txn_kv_error", + &ret); + if (ret == 1) { + const int max_retry = std::max(1, config::recycle_txn_delete_max_retry_times); + for (int i = 1; i <= max_retry; ++i) { + LOG(WARNING) << "txn conflict, retry times=" << i << " key=" << hex(k); + ret = delete_recycle_txn_kv(k); + TEST_SYNC_POINT_CALLBACK( + "InstanceRecycler::recycle_expired_txn_label.delete_recycle_txn_kv_" + "error", + &ret); + if (ret != 1) { + break; } + // random sleep 0-100 ms to retry + std::this_thread::sleep_for(std::chrono::milliseconds(rand() % 100)); } - if (ret != 0) { - LOG_WARNING("failed to delete recycle txn kv") - .tag("instance id", instance_id_) - .tag("key", hex(k)); - return -1; + } + return ret; + }; + + for (auto& [label_key, txn_keys] : recycle_txn_keys_by_label) { + concurrent_delete_executor.add([&, txn_keys = std::move(txn_keys), label_key]() { + VLOG_DEBUG << "recycle txn label group, key=" << hex(label_key) + << " txn_count=" << txn_keys.size(); + for (const auto& k : txn_keys) { + int ret = delete_recycle_txn_kv_with_retry(k); + if (ret != 0) { + LOG_WARNING("failed to delete recycle txn kv") + .tag("instance id", instance_id_) + .tag("key", hex(k)) + .tag("label_key", hex(label_key)); + return -1; + } } return 0; }); @@ -7322,8 +7345,8 @@ int InstanceRecycler::scan_and_statistics_rowsets() { std::string recyc_rs_key0; std::string recyc_rs_key1; recycle_rowset_key(recyc_rs_key_info0, &recyc_rs_key0); - recycle_rowset_key(recyc_rs_key_info1, &recyc_rs_key1); - int64_t earlest_ts = std::numeric_limits::max(); + recycle_rowset_key(recyc_rs_key_info1, &recyc_rs_key1); + int64_t earlest_ts = std::numeric_limits::max(); auto handle_rowset_kv = [&, this](std::string_view k, std::string_view v) -> int { RecycleRowsetPB rowset; @@ -7347,11 +7370,12 @@ int InstanceRecycler::scan_and_statistics_rowsets() { metrics_context.total_need_recycle_num++; metrics_context.total_need_recycle_data_size += rowset.rowset_meta().total_disk_size(); segment_metrics_context_.total_need_recycle_num += rowset.rowset_meta().num_segments(); - segment_metrics_context_.total_need_recycle_data_size += rowset.rowset_meta().total_disk_size(); + segment_metrics_context_.total_need_recycle_data_size += + rowset.rowset_meta().total_disk_size(); return 0; } - if(!rowset_meta->has_is_recycled() || !rowset_meta->is_recycled()) { + if (!rowset_meta->has_is_recycled() || !rowset_meta->is_recycled()) { return 0; } @@ -7398,7 +7422,7 @@ int InstanceRecycler::scan_and_statistics_tmp_rowsets() { DCHECK_GT(rowset.txn_id(), 0) << "txn_id=" << rowset.txn_id() << " rowset=" << rowset.ShortDebugString(); - if(!rowset.has_is_recycled() || !rowset.is_recycled()) { + if (!rowset.has_is_recycled() || !rowset.is_recycled()) { return 0; } @@ -7475,7 +7499,8 @@ int InstanceRecycler::scan_and_statistics_abort_timeout_txn() { return 0; }; - int ret = scan_and_recycle(begin_txn_running_key, end_txn_running_key, std::move(handle_abort_timeout_txn_kv)); + int ret = scan_and_recycle(begin_txn_running_key, end_txn_running_key, + std::move(handle_abort_timeout_txn_kv)); metrics_context.report(true); return ret; } @@ -7509,7 +7534,8 @@ int InstanceRecycler::scan_and_statistics_expired_txn_label() { return 0; }; - int ret = scan_and_recycle(begin_recycle_txn_key, end_recycle_txn_key, std::move(handle_expired_txn_label_kv)); + int ret = scan_and_recycle(begin_recycle_txn_key, end_recycle_txn_key, + std::move(handle_expired_txn_label_kv)); metrics_context.report(true); return ret; } @@ -7740,7 +7766,7 @@ int InstanceRecycler::scan_and_statistics_restore_jobs() { return 0; } metrics_context.total_need_recycle_num++; - if(restore_job_pb.need_recycle_data()) { + if (restore_job_pb.need_recycle_data()) { scan_tablet_and_statistics(restore_job_pb.tablet_id(), metrics_context); } return 0; @@ -7782,7 +7808,7 @@ void InstanceRecycler::scan_and_statistics_operation_logs() { OperationLogReferenceInfo ref_info; if (recycle_checker.can_recycle(log_versionstamp, operation_log.min_timestamp(), - &ref_info)) { + &ref_info)) { metrics_context.total_need_recycle_num++; metrics_context.total_need_recycle_data_size += operation_log.ByteSizeLong(); } @@ -7938,8 +7964,8 @@ int InstanceRecycler::cleanup_rowset_metadata(const std::vector::max(), std::numeric_limits::max()}); + {reference_instance_id, tablet_id, rowset_id, std::numeric_limits::max(), + std::numeric_limits::max()}); txn->remove(dbm_start_key, dbm_end_key); LOG_INFO("remove delete bitmap kv in cleanup phase") .tag("instance_id", instance_id_) @@ -7962,8 +7988,8 @@ int InstanceRecycler::cleanup_rowset_metadata(const std::vector( - txn.get(), task.versioned_rowset_key, task.versionstamp); + versioned::document_remove(txn.get(), task.versioned_rowset_key, + task.versionstamp); LOG_INFO("remove versioned meta rowset key in cleanup phase") .tag("instance_id", instance_id_) .tag("tablet_id", tablet_id) diff --git a/cloud/test/recycler_test.cpp b/cloud/test/recycler_test.cpp index ad8075cec9c55f..bd55f6eb6727c3 100644 --- a/cloud/test/recycler_test.cpp +++ b/cloud/test/recycler_test.cpp @@ -8245,20 +8245,24 @@ TEST(RecyclerTest, concurrent_recycle_txn_label_failure_test) { auto txn_kv = mem_txn_kv; ASSERT_TRUE(txn_kv.get()) << "exit get MemTxnKv error" << std::endl; - make_multiple_txn_info_kvs(txn_kv, 20000, 15000); - check_multiple_txn_info_kvs(txn_kv, 20000); + make_multiple_txn_info_kvs(txn_kv, 40000, 30000); + check_multiple_txn_info_kvs(txn_kv, 40000); auto* sp = SyncPoint::get_instance(); DORIS_CLOUD_DEFER { SyncPoint::get_instance()->clear_all_call_backs(); }; - size_t recycle_txn_info_keys_cnt = 0; - sp->set_call_back("InstanceRecycler::recycle_expired_txn_label.check_recycle_txn_info_keys", - [&](auto&& args) { - auto* recycle_txn_info_keys = - try_any_cast*>(args[0]); - recycle_txn_info_keys_cnt += recycle_txn_info_keys->size(); - }); + size_t recycle_txn_keys_cnt = 0; + sp->set_call_back( + "InstanceRecycler::recycle_expired_txn_label.check_recycle_txn_keys_by_label", + [&](auto&& args) { + auto* recycle_txn_keys_by_label = + try_any_cast>*>( + args[0]); + for (const auto& entry : *recycle_txn_keys_by_label) { + recycle_txn_keys_cnt += entry.second.size(); + } + }); sp->set_call_back("InstanceRecycler::recycle_expired_txn_label.failure", [](auto&& args) { auto* ret = try_any_cast(args[0]); *ret = -1; @@ -8276,7 +8280,7 @@ TEST(RecyclerTest, concurrent_recycle_txn_label_failure_test) { std::cout << "recycle expired txn label cost=" << std::chrono::duration_cast(finish - start).count() << "ms" << std::endl; - check_multiple_txn_info_kvs(txn_kv, (20000 - recycle_txn_info_keys_cnt)); + check_multiple_txn_info_kvs(txn_kv, (40000 - recycle_txn_keys_cnt)); } TEST(RecyclerTest, concurrent_recycle_txn_label_conflict_test) { config::label_keep_max_second = 0; @@ -8427,7 +8431,7 @@ TEST(RecyclerTest, concurrent_recycle_txn_label_conflict_test) { std::cout << "Update label after count: " << update_label_after_count << std::endl; std::cout << "Transaction conflict count: " << txn_conflict_count << std::endl; - EXPECT_GT(txn_conflict_count, 0) << "txn_conflict sync point should be triggered"; + EXPECT_EQ(txn_conflict_count, 0) << "txn conflicts should not occur within one label group"; std::unique_ptr verify_txn; ASSERT_EQ(mem_txn_kv->create_txn(&verify_txn), TxnErrorCode::TXN_OK); @@ -8455,7 +8459,7 @@ TEST(RecyclerTest, concurrent_recycle_txn_label_conflict_test) { } } -TEST(RecyclerTest, recycle_txn_label_deal_with_conflict_error_test) { +TEST(RecyclerTest, recycle_txn_label_propagate_delete_error_test) { config::label_keep_max_second = 0; config::recycle_pool_parallelism = 20; @@ -8601,10 +8605,47 @@ TEST(RecyclerTest, recycle_txn_label_deal_with_conflict_error_test) { std::make_shared(mem_txn_kv)); ASSERT_EQ(recycler.init(), 0); - // deal with conflict but error during recycle + // Propagate a recycle error without relying on an internal label conflict. ASSERT_EQ(recycler.recycle_expired_txn_label(), -1); - EXPECT_GT(txn_conflict_count, 0) << "txn_conflict sync point should be triggered"; + EXPECT_EQ(txn_conflict_count, 0) << "txn conflicts should not occur within one label group"; +} + +TEST(RecyclerTest, recycle_txn_label_retry_after_conflict_test) { + config::label_keep_max_second = 0; + + auto mem_txn_kv = std::make_shared(); + ASSERT_EQ(mem_txn_kv->init(), 0); + make_single_txn_related_kvs(mem_txn_kv, 0, 1); + + auto* sp = SyncPoint::get_instance(); + DORIS_CLOUD_DEFER { + SyncPoint::get_instance()->clear_all_call_backs(); + }; + + std::atomic delete_attempt_count {0}; + sp->set_call_back("InstanceRecycler::recycle_expired_txn_label.delete_recycle_txn_kv_error", + [&](auto&& args) { + auto* ret = try_any_cast(args[0]); + const int attempt = delete_attempt_count.fetch_add(1); + *ret = attempt == 0 ? 1 : 0; + }); + sp->enable_processing(); + + InstanceInfoPB instance; + instance.set_instance_id(instance_id); + InstanceRecycler recycler(mem_txn_kv, instance, thread_group, + std::make_shared(mem_txn_kv)); + ASSERT_EQ(recycler.init(), 0); + + ASSERT_EQ(recycler.recycle_expired_txn_label(), 0); + EXPECT_EQ(delete_attempt_count, 2); + + std::unique_ptr verify_txn; + ASSERT_EQ(mem_txn_kv->create_txn(&verify_txn), TxnErrorCode::TXN_OK); + const std::string recycle_key = recycle_txn_key({instance_id, 0, 1000000}); + std::string recycle_value; + EXPECT_EQ(verify_txn->get(recycle_key, &recycle_value), TxnErrorCode::TXN_KEY_NOT_FOUND); } TEST(RecyclerTest, recycle_restore_job_complete_state) {