github-actions[bot] commented on code in PR #33268:
URL: https://github.com/apache/doris/pull/33268#discussion_r1552934349
##########
be/src/cloud/cloud_compaction_action.cpp:
##########
@@ -121,58 +121,29 @@ Status
CloudCompactionAction::_handle_run_compaction(HttpRequest* req, std::stri
RETURN_NOT_OK_STATUS_WITH_WARN(_check_param(req, &tablet_id, &table_id),
"check param failed");
// check compaction_type equals 'base' or 'cumulative'
- std::string compaction_type = req->param(PARAM_COMPACTION_TYPE);
+ auto& compaction_type = req->param(PARAM_COMPACTION_TYPE);
Review Comment:
warning: 'auto &compaction_type' can be declared as 'const auto
&compaction_type' [readability-qualified-auto]
```suggestion
const auto& compaction_type = req->param(PARAM_COMPACTION_TYPE);
```
##########
be/src/cloud/cloud_storage_engine.h:
##########
@@ -95,6 +95,21 @@
Status get_compaction_status_json(std::string* result);
+ bool has_base_compaction(int64_t tablet_id) const {
+ std::lock_guard lock(_compaction_mtx);
+ return _submitted_base_compactions.count(tablet_id);
+ }
+
+ bool has_cumu_compaction(int64_t tablet_id) const {
Review Comment:
warning: method 'has_cumu_compaction' can be made static
[readability-convert-member-functions-to-static]
```suggestion
static bool has_cumu_compaction(int64_t tablet_id) {
```
##########
be/src/cloud/injection_point_action.cpp:
##########
@@ -0,0 +1,196 @@
+#include "cloud/injection_point_action.h"
+
+#include <glog/logging.h>
+
+#include <chrono>
+#include <mutex>
+
+#include "common/status.h"
+#include "common/sync_point.h"
+#include "http/http_channel.h"
+#include "http/http_request.h"
+#include "http/http_status.h"
+#include "olap/rowset/rowset.h"
+#include "util/stack_util.h"
+
+namespace doris {
+namespace {
+
+// TODO(cyx): Provide an object pool
+// `suite_map` won't be modified after `register_suites`
+std::map<std::string, std::function<void()>> suite_map;
+std::once_flag register_suites_once;
+
+// only call once
+void register_suites() {
+ suite_map.emplace("test_compaction", [] {
+ auto sp = SyncPoint::get_instance();
Review Comment:
warning: 'auto sp' can be declared as 'auto *sp' [readability-qualified-auto]
```suggestion
auto *sp = SyncPoint::get_instance();
```
##########
be/src/cloud/cloud_compaction_action.cpp:
##########
@@ -121,58 +121,29 @@
RETURN_NOT_OK_STATUS_WITH_WARN(_check_param(req, &tablet_id, &table_id),
"check param failed");
// check compaction_type equals 'base' or 'cumulative'
- std::string compaction_type = req->param(PARAM_COMPACTION_TYPE);
+ auto& compaction_type = req->param(PARAM_COMPACTION_TYPE);
if (compaction_type != PARAM_COMPACTION_BASE &&
compaction_type != PARAM_COMPACTION_CUMULATIVE &&
compaction_type != PARAM_COMPACTION_FULL) {
return Status::NotSupported("The compaction type '{}' is not
supported", compaction_type);
}
- if (tablet_id == 0 && table_id != 0) {
- /*
- std::vector<TabletSharedPtr> tablet_vec =
_engine.tablet_manager()->get_all_tablet(
- [table_id](Tablet* tablet) -> bool { return
tablet->get_table_id() == table_id; });
- */
- auto tablet_vec = _engine.tablet_mgr().get_weak_tablets();
- for (const auto& weak_tablet : tablet_vec) {
- if (auto tablet = weak_tablet.lock()) {
- if (tablet->table_id() != table_id) {
- continue;
- }
- RETURN_IF_ERROR(
- _engine.submit_compaction_task(tablet,
CompactionType::FULL_COMPACTION));
- }
- }
- } else {
- // 2. fetch the tablet by tablet_id
- CloudTabletSPtr tablet =
DORIS_TRY(_engine.tablet_mgr().get_tablet(tablet_id));
- if (tablet == nullptr) {
- return Status::NotFound("Tablet not found. tablet_id={}",
tablet_id);
- }
+ CloudTabletSPtr tablet =
DORIS_TRY(_engine.tablet_mgr().get_tablet(tablet_id));
+ if (tablet == nullptr) {
+ return Status::NotFound("Tablet not found. tablet_id={}", tablet_id);
+ }
- // 3. execute compaction task
- std::packaged_task<Status()> task([this, tablet, compaction_type]() {
- return _execute_compaction_callback(tablet, compaction_type);
- });
- std::future<Status> future_obj = task.get_future();
- std::thread(std::move(task)).detach();
+ // 3. submit compaction task
+ RETURN_IF_ERROR(_engine.submit_compaction_task(
+ tablet, compaction_type == PARAM_COMPACTION_BASE ?
CompactionType::BASE_COMPACTION
+ : compaction_type == PARAM_COMPACTION_CUMULATIVE
+ ? CompactionType::CUMULATIVE_COMPACTION
+ : CompactionType::FULL_COMPACTION));
- // 4. wait for result for 2 seconds by async
- std::future_status status =
future_obj.wait_for(std::chrono::seconds(2));
- if (status == std::future_status::ready) {
- // fetch execute result
- Status olap_status = future_obj.get();
- if (!olap_status.ok()) {
- return olap_status;
- }
- } else {
- LOG(INFO) << "Manual compaction task is timeout for waiting "
- << (status == std::future_status::timeout);
- }
- }
LOG(INFO) << "Manual compaction task is successfully triggered";
*json_result =
- R"({"status": "Success", "msg": "compaction task is successfully
triggered. Table id: )" +
+ "{\"status\": \"Success\", \"msg\": \"compaction task is
successfully triggered. Table "
+ "id: " +
Review Comment:
warning: escaped string literal can be written as a raw string literal
[modernize-raw-string-literal]
```suggestion
R"({"status": "Success", "msg": "compaction task is successfully
triggered. Table id: )" +
```
##########
be/src/cloud/injection_point_action.cpp:
##########
@@ -0,0 +1,196 @@
+#include "cloud/injection_point_action.h"
+
+#include <glog/logging.h>
+
+#include <chrono>
+#include <mutex>
+
+#include "common/status.h"
+#include "common/sync_point.h"
+#include "http/http_channel.h"
+#include "http/http_request.h"
+#include "http/http_status.h"
+#include "olap/rowset/rowset.h"
+#include "util/stack_util.h"
+
+namespace doris {
+namespace {
+
+// TODO(cyx): Provide an object pool
+// `suite_map` won't be modified after `register_suites`
+std::map<std::string, std::function<void()>> suite_map;
+std::once_flag register_suites_once;
+
+// only call once
+void register_suites() {
+ suite_map.emplace("test_compaction", [] {
+ auto sp = SyncPoint::get_instance();
+ sp->set_call_back("new_cumulative_point", [](auto&& args) {
+ auto output_rowset = try_any_cast<Rowset*>(args[0]);
+ auto last_cumulative_point = try_any_cast<int64_t>(args[1]);
+ auto pair = try_any_cast<std::pair<int64_t, bool>*>(args.back());
+ pair->first = output_rowset->start_version() ==
last_cumulative_point
+ ? output_rowset->end_version() + 1
+ : last_cumulative_point;
+ pair->second = true;
+ });
+ });
+}
+
+void set_sleep(const std::string& point, HttpRequest* req) {
+ int duration = 0;
+ auto& duration_str = req->param("duration");
+ if (!duration_str.empty()) {
+ try {
+ duration = std::stoi(duration_str);
+ } catch (const std::exception&) {
+ HttpChannel::send_reply(req, HttpStatus::BAD_REQUEST,
+ "invalid duration: " + duration_str);
+ return;
+ }
+ }
+ auto sp = SyncPoint::get_instance();
+ sp->set_call_back(point, [duration](auto&& args) {
+ std::this_thread::sleep_for(std::chrono::milliseconds(duration));
+ });
+ HttpChannel::send_reply(req, HttpStatus::OK, "OK");
+}
+
+void set_return(const std::string& point, HttpRequest* req) {
+ auto sp = SyncPoint::get_instance();
+ sp->set_call_back(point, [](auto&& args) {
+ try {
+ auto pred = try_any_cast<bool*>(args.back());
+ *pred = true;
+ } catch (const std::bad_any_cast&) {
+ LOG_EVERY_N(ERROR, 10) << "failed to process `return` callback\n"
<< get_stack_trace();
+ }
+ });
+ HttpChannel::send_reply(req, HttpStatus::OK, "OK");
+}
+
+void set_return_ok(const std::string& point, HttpRequest* req) {
+ auto sp = SyncPoint::get_instance();
+ sp->set_call_back(point, [](auto&& args) {
+ try {
+ auto* pair = try_any_cast_ret<Status>(args);
+ pair->first = Status::OK();
+ pair->second = true;
+ } catch (const std::bad_any_cast&) {
+ LOG_EVERY_N(ERROR, 10) << "failed to process `return_ok`
callback\n"
+ << get_stack_trace();
+ }
+ });
+ HttpChannel::send_reply(req, HttpStatus::OK, "OK");
+}
+
+void set_return_error(const std::string& point, HttpRequest* req) {
+ const std::string CODE_PARAM = "code";
+ int code = ErrorCode::INTERNAL_ERROR;
+ auto& code_str = req->param(CODE_PARAM);
+ if (!code_str.empty()) {
+ try {
+ code = std::stoi(code_str);
+ } catch (const std::exception& e) {
+ HttpChannel::send_reply(req, HttpStatus::BAD_REQUEST,
+ fmt::format("convert topn failed, {}",
e.what()));
+ return;
+ }
+ }
+
+ auto sp = SyncPoint::get_instance();
+ sp->set_call_back(point, [code](auto&& args) {
+ try {
+ auto* pair = try_any_cast_ret<Status>(args);
+ pair->first = Status::Error<false>(code, "injected error");
+ pair->second = true;
+ } catch (const std::bad_any_cast&) {
+ LOG_EVERY_N(ERROR, 10) << "failed to process `return_error`
callback\n"
+ << get_stack_trace();
+ }
+ });
+ HttpChannel::send_reply(req, HttpStatus::OK, "OK");
+}
+
+void handle_set(HttpRequest* req) {
+ auto& point = req->param("name");
+ if (point.empty()) {
+ HttpChannel::send_reply(req, HttpStatus::BAD_REQUEST, "empty point
name");
+ return;
+ }
+ auto& behavior = req->param("behavior");
+ if (behavior.empty()) {
+ HttpChannel::send_reply(req, HttpStatus::BAD_REQUEST, "empty
behavior");
+ return;
+ }
+ if (behavior == "sleep") {
+ set_sleep(point, req);
+ return;
+ } else if (behavior == "return") {
+ set_return(point, req);
+ return;
+ } else if (behavior == "return_ok") {
+ set_return_ok(point, req);
+ return;
+ } else if (behavior == "return_error") {
+ set_return_error(point, req);
+ return;
+ }
+ HttpChannel::send_reply(req, HttpStatus::BAD_REQUEST, "unknown behavior: "
+ behavior);
+}
+
+void handle_clear(HttpRequest* req) {
+ auto& point = req->param("name");
+ if (point.empty()) {
+ HttpChannel::send_reply(req, HttpStatus::BAD_REQUEST, "empty point
name");
+ return;
+ }
+ auto sp = SyncPoint::get_instance();
Review Comment:
warning: 'auto sp' can be declared as 'auto *sp' [readability-qualified-auto]
```suggestion
auto *sp = SyncPoint::get_instance();
```
##########
be/src/cloud/injection_point_action.cpp:
##########
@@ -0,0 +1,196 @@
+#include "cloud/injection_point_action.h"
+
+#include <glog/logging.h>
+
+#include <chrono>
+#include <mutex>
+
+#include "common/status.h"
+#include "common/sync_point.h"
+#include "http/http_channel.h"
+#include "http/http_request.h"
+#include "http/http_status.h"
+#include "olap/rowset/rowset.h"
+#include "util/stack_util.h"
+
+namespace doris {
+namespace {
+
+// TODO(cyx): Provide an object pool
+// `suite_map` won't be modified after `register_suites`
+std::map<std::string, std::function<void()>> suite_map;
+std::once_flag register_suites_once;
+
+// only call once
+void register_suites() {
+ suite_map.emplace("test_compaction", [] {
+ auto sp = SyncPoint::get_instance();
+ sp->set_call_back("new_cumulative_point", [](auto&& args) {
+ auto output_rowset = try_any_cast<Rowset*>(args[0]);
+ auto last_cumulative_point = try_any_cast<int64_t>(args[1]);
+ auto pair = try_any_cast<std::pair<int64_t, bool>*>(args.back());
+ pair->first = output_rowset->start_version() ==
last_cumulative_point
+ ? output_rowset->end_version() + 1
+ : last_cumulative_point;
+ pair->second = true;
+ });
+ });
+}
+
+void set_sleep(const std::string& point, HttpRequest* req) {
+ int duration = 0;
+ auto& duration_str = req->param("duration");
+ if (!duration_str.empty()) {
+ try {
+ duration = std::stoi(duration_str);
+ } catch (const std::exception&) {
+ HttpChannel::send_reply(req, HttpStatus::BAD_REQUEST,
+ "invalid duration: " + duration_str);
+ return;
+ }
+ }
+ auto sp = SyncPoint::get_instance();
+ sp->set_call_back(point, [duration](auto&& args) {
+ std::this_thread::sleep_for(std::chrono::milliseconds(duration));
+ });
+ HttpChannel::send_reply(req, HttpStatus::OK, "OK");
+}
+
+void set_return(const std::string& point, HttpRequest* req) {
+ auto sp = SyncPoint::get_instance();
+ sp->set_call_back(point, [](auto&& args) {
+ try {
+ auto pred = try_any_cast<bool*>(args.back());
+ *pred = true;
+ } catch (const std::bad_any_cast&) {
+ LOG_EVERY_N(ERROR, 10) << "failed to process `return` callback\n"
<< get_stack_trace();
+ }
+ });
+ HttpChannel::send_reply(req, HttpStatus::OK, "OK");
+}
+
+void set_return_ok(const std::string& point, HttpRequest* req) {
+ auto sp = SyncPoint::get_instance();
+ sp->set_call_back(point, [](auto&& args) {
+ try {
+ auto* pair = try_any_cast_ret<Status>(args);
+ pair->first = Status::OK();
+ pair->second = true;
+ } catch (const std::bad_any_cast&) {
+ LOG_EVERY_N(ERROR, 10) << "failed to process `return_ok`
callback\n"
+ << get_stack_trace();
+ }
+ });
+ HttpChannel::send_reply(req, HttpStatus::OK, "OK");
+}
+
+void set_return_error(const std::string& point, HttpRequest* req) {
+ const std::string CODE_PARAM = "code";
+ int code = ErrorCode::INTERNAL_ERROR;
+ auto& code_str = req->param(CODE_PARAM);
Review Comment:
warning: 'auto &code_str' can be declared as 'const auto &code_str'
[readability-qualified-auto]
```suggestion
const auto& code_str = req->param(CODE_PARAM);
```
##########
be/src/cloud/injection_point_action.cpp:
##########
@@ -0,0 +1,196 @@
+#include "cloud/injection_point_action.h"
+
+#include <glog/logging.h>
+
+#include <chrono>
+#include <mutex>
+
+#include "common/status.h"
+#include "common/sync_point.h"
+#include "http/http_channel.h"
+#include "http/http_request.h"
+#include "http/http_status.h"
+#include "olap/rowset/rowset.h"
+#include "util/stack_util.h"
+
+namespace doris {
+namespace {
+
+// TODO(cyx): Provide an object pool
+// `suite_map` won't be modified after `register_suites`
+std::map<std::string, std::function<void()>> suite_map;
+std::once_flag register_suites_once;
+
+// only call once
+void register_suites() {
+ suite_map.emplace("test_compaction", [] {
+ auto sp = SyncPoint::get_instance();
+ sp->set_call_back("new_cumulative_point", [](auto&& args) {
+ auto output_rowset = try_any_cast<Rowset*>(args[0]);
+ auto last_cumulative_point = try_any_cast<int64_t>(args[1]);
+ auto pair = try_any_cast<std::pair<int64_t, bool>*>(args.back());
+ pair->first = output_rowset->start_version() ==
last_cumulative_point
+ ? output_rowset->end_version() + 1
+ : last_cumulative_point;
+ pair->second = true;
+ });
+ });
+}
+
+void set_sleep(const std::string& point, HttpRequest* req) {
+ int duration = 0;
+ auto& duration_str = req->param("duration");
+ if (!duration_str.empty()) {
+ try {
+ duration = std::stoi(duration_str);
+ } catch (const std::exception&) {
+ HttpChannel::send_reply(req, HttpStatus::BAD_REQUEST,
+ "invalid duration: " + duration_str);
+ return;
+ }
+ }
+ auto sp = SyncPoint::get_instance();
+ sp->set_call_back(point, [duration](auto&& args) {
+ std::this_thread::sleep_for(std::chrono::milliseconds(duration));
+ });
+ HttpChannel::send_reply(req, HttpStatus::OK, "OK");
+}
+
+void set_return(const std::string& point, HttpRequest* req) {
+ auto sp = SyncPoint::get_instance();
+ sp->set_call_back(point, [](auto&& args) {
+ try {
+ auto pred = try_any_cast<bool*>(args.back());
+ *pred = true;
+ } catch (const std::bad_any_cast&) {
+ LOG_EVERY_N(ERROR, 10) << "failed to process `return` callback\n"
<< get_stack_trace();
+ }
+ });
+ HttpChannel::send_reply(req, HttpStatus::OK, "OK");
+}
+
+void set_return_ok(const std::string& point, HttpRequest* req) {
+ auto sp = SyncPoint::get_instance();
+ sp->set_call_back(point, [](auto&& args) {
+ try {
+ auto* pair = try_any_cast_ret<Status>(args);
+ pair->first = Status::OK();
+ pair->second = true;
+ } catch (const std::bad_any_cast&) {
+ LOG_EVERY_N(ERROR, 10) << "failed to process `return_ok`
callback\n"
+ << get_stack_trace();
+ }
+ });
+ HttpChannel::send_reply(req, HttpStatus::OK, "OK");
+}
+
+void set_return_error(const std::string& point, HttpRequest* req) {
+ const std::string CODE_PARAM = "code";
+ int code = ErrorCode::INTERNAL_ERROR;
+ auto& code_str = req->param(CODE_PARAM);
+ if (!code_str.empty()) {
+ try {
+ code = std::stoi(code_str);
+ } catch (const std::exception& e) {
+ HttpChannel::send_reply(req, HttpStatus::BAD_REQUEST,
+ fmt::format("convert topn failed, {}",
e.what()));
+ return;
+ }
+ }
+
+ auto sp = SyncPoint::get_instance();
Review Comment:
warning: 'auto sp' can be declared as 'auto *sp' [readability-qualified-auto]
```suggestion
auto *sp = SyncPoint::get_instance();
```
##########
be/src/cloud/cloud_storage_engine.h:
##########
@@ -95,6 +95,21 @@
Status get_compaction_status_json(std::string* result);
+ bool has_base_compaction(int64_t tablet_id) const {
+ std::lock_guard lock(_compaction_mtx);
+ return _submitted_base_compactions.count(tablet_id);
+ }
+
+ bool has_cumu_compaction(int64_t tablet_id) const {
+ std::lock_guard lock(_compaction_mtx);
+ return _submitted_cumu_compactions.count(tablet_id);
+ }
+
+ bool has_full_compaction(int64_t tablet_id) const {
Review Comment:
warning: method 'has_full_compaction' can be made static
[readability-convert-member-functions-to-static]
```suggestion
static bool has_full_compaction(int64_t tablet_id) {
```
##########
be/src/cloud/injection_point_action.cpp:
##########
@@ -0,0 +1,196 @@
+#include "cloud/injection_point_action.h"
+
+#include <glog/logging.h>
Review Comment:
warning: 'glog/logging.h' file not found [clang-diagnostic-error]
```cpp
#include <glog/logging.h>
^
```
##########
be/src/cloud/injection_point_action.cpp:
##########
@@ -0,0 +1,196 @@
+#include "cloud/injection_point_action.h"
+
+#include <glog/logging.h>
+
+#include <chrono>
+#include <mutex>
+
+#include "common/status.h"
+#include "common/sync_point.h"
+#include "http/http_channel.h"
+#include "http/http_request.h"
+#include "http/http_status.h"
+#include "olap/rowset/rowset.h"
+#include "util/stack_util.h"
+
+namespace doris {
+namespace {
+
+// TODO(cyx): Provide an object pool
+// `suite_map` won't be modified after `register_suites`
+std::map<std::string, std::function<void()>> suite_map;
+std::once_flag register_suites_once;
+
+// only call once
+void register_suites() {
+ suite_map.emplace("test_compaction", [] {
+ auto sp = SyncPoint::get_instance();
+ sp->set_call_back("new_cumulative_point", [](auto&& args) {
+ auto output_rowset = try_any_cast<Rowset*>(args[0]);
+ auto last_cumulative_point = try_any_cast<int64_t>(args[1]);
+ auto pair = try_any_cast<std::pair<int64_t, bool>*>(args.back());
+ pair->first = output_rowset->start_version() ==
last_cumulative_point
+ ? output_rowset->end_version() + 1
+ : last_cumulative_point;
+ pair->second = true;
+ });
+ });
+}
+
+void set_sleep(const std::string& point, HttpRequest* req) {
+ int duration = 0;
+ auto& duration_str = req->param("duration");
+ if (!duration_str.empty()) {
+ try {
+ duration = std::stoi(duration_str);
+ } catch (const std::exception&) {
+ HttpChannel::send_reply(req, HttpStatus::BAD_REQUEST,
+ "invalid duration: " + duration_str);
+ return;
+ }
+ }
+ auto sp = SyncPoint::get_instance();
Review Comment:
warning: 'auto sp' can be declared as 'auto *sp' [readability-qualified-auto]
```suggestion
auto *sp = SyncPoint::get_instance();
```
##########
be/src/cloud/cloud_storage_engine.h:
##########
@@ -95,6 +95,21 @@ class CloudStorageEngine final : public BaseStorageEngine {
Status get_compaction_status_json(std::string* result);
+ bool has_base_compaction(int64_t tablet_id) const {
Review Comment:
warning: method 'has_base_compaction' can be made static
[readability-convert-member-functions-to-static]
```suggestion
static bool has_base_compaction(int64_t tablet_id) {
```
##########
be/src/cloud/injection_point_action.cpp:
##########
@@ -0,0 +1,196 @@
+#include "cloud/injection_point_action.h"
+
+#include <glog/logging.h>
+
+#include <chrono>
+#include <mutex>
+
+#include "common/status.h"
+#include "common/sync_point.h"
+#include "http/http_channel.h"
+#include "http/http_request.h"
+#include "http/http_status.h"
+#include "olap/rowset/rowset.h"
+#include "util/stack_util.h"
+
+namespace doris {
+namespace {
+
+// TODO(cyx): Provide an object pool
+// `suite_map` won't be modified after `register_suites`
+std::map<std::string, std::function<void()>> suite_map;
+std::once_flag register_suites_once;
+
+// only call once
+void register_suites() {
+ suite_map.emplace("test_compaction", [] {
+ auto sp = SyncPoint::get_instance();
+ sp->set_call_back("new_cumulative_point", [](auto&& args) {
+ auto output_rowset = try_any_cast<Rowset*>(args[0]);
+ auto last_cumulative_point = try_any_cast<int64_t>(args[1]);
+ auto pair = try_any_cast<std::pair<int64_t, bool>*>(args.back());
+ pair->first = output_rowset->start_version() ==
last_cumulative_point
+ ? output_rowset->end_version() + 1
+ : last_cumulative_point;
+ pair->second = true;
+ });
+ });
+}
+
+void set_sleep(const std::string& point, HttpRequest* req) {
+ int duration = 0;
+ auto& duration_str = req->param("duration");
+ if (!duration_str.empty()) {
+ try {
+ duration = std::stoi(duration_str);
+ } catch (const std::exception&) {
+ HttpChannel::send_reply(req, HttpStatus::BAD_REQUEST,
+ "invalid duration: " + duration_str);
+ return;
+ }
+ }
+ auto sp = SyncPoint::get_instance();
+ sp->set_call_back(point, [duration](auto&& args) {
+ std::this_thread::sleep_for(std::chrono::milliseconds(duration));
+ });
+ HttpChannel::send_reply(req, HttpStatus::OK, "OK");
+}
+
+void set_return(const std::string& point, HttpRequest* req) {
+ auto sp = SyncPoint::get_instance();
Review Comment:
warning: 'auto sp' can be declared as 'auto *sp' [readability-qualified-auto]
```suggestion
auto *sp = SyncPoint::get_instance();
```
##########
be/src/cloud/injection_point_action.cpp:
##########
@@ -0,0 +1,196 @@
+#include "cloud/injection_point_action.h"
+
+#include <glog/logging.h>
+
+#include <chrono>
+#include <mutex>
+
+#include "common/status.h"
+#include "common/sync_point.h"
+#include "http/http_channel.h"
+#include "http/http_request.h"
+#include "http/http_status.h"
+#include "olap/rowset/rowset.h"
+#include "util/stack_util.h"
+
+namespace doris {
+namespace {
+
+// TODO(cyx): Provide an object pool
+// `suite_map` won't be modified after `register_suites`
+std::map<std::string, std::function<void()>> suite_map;
+std::once_flag register_suites_once;
+
+// only call once
+void register_suites() {
+ suite_map.emplace("test_compaction", [] {
+ auto sp = SyncPoint::get_instance();
+ sp->set_call_back("new_cumulative_point", [](auto&& args) {
+ auto output_rowset = try_any_cast<Rowset*>(args[0]);
+ auto last_cumulative_point = try_any_cast<int64_t>(args[1]);
+ auto pair = try_any_cast<std::pair<int64_t, bool>*>(args.back());
+ pair->first = output_rowset->start_version() ==
last_cumulative_point
+ ? output_rowset->end_version() + 1
+ : last_cumulative_point;
+ pair->second = true;
+ });
+ });
+}
+
+void set_sleep(const std::string& point, HttpRequest* req) {
+ int duration = 0;
+ auto& duration_str = req->param("duration");
+ if (!duration_str.empty()) {
+ try {
+ duration = std::stoi(duration_str);
+ } catch (const std::exception&) {
+ HttpChannel::send_reply(req, HttpStatus::BAD_REQUEST,
+ "invalid duration: " + duration_str);
+ return;
+ }
+ }
+ auto sp = SyncPoint::get_instance();
+ sp->set_call_back(point, [duration](auto&& args) {
+ std::this_thread::sleep_for(std::chrono::milliseconds(duration));
+ });
+ HttpChannel::send_reply(req, HttpStatus::OK, "OK");
+}
+
+void set_return(const std::string& point, HttpRequest* req) {
+ auto sp = SyncPoint::get_instance();
+ sp->set_call_back(point, [](auto&& args) {
+ try {
+ auto pred = try_any_cast<bool*>(args.back());
+ *pred = true;
+ } catch (const std::bad_any_cast&) {
+ LOG_EVERY_N(ERROR, 10) << "failed to process `return` callback\n"
<< get_stack_trace();
+ }
+ });
+ HttpChannel::send_reply(req, HttpStatus::OK, "OK");
+}
+
+void set_return_ok(const std::string& point, HttpRequest* req) {
+ auto sp = SyncPoint::get_instance();
Review Comment:
warning: 'auto sp' can be declared as 'auto *sp' [readability-qualified-auto]
```suggestion
auto *sp = SyncPoint::get_instance();
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscr...@doris.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscr...@doris.apache.org
For additional commands, e-mail: commits-h...@doris.apache.org
