github-actions[bot] commented on code in PR #16639:
URL: https://github.com/apache/doris/pull/16639#discussion_r1103645984
##########
be/src/service/internal_service.cpp:
##########
@@ -247,15 +292,18 @@
const
PTabletWriterCancelRequest* request,
PTabletWriterCancelResult*
response,
google::protobuf::Closure*
done) {
- VLOG_RPC << "tablet writer cancel, id=" << request->id() << ", index_id="
<< request->index_id()
- << ", sender_id=" << request->sender_id();
- brpc::ClosureGuard closure_guard(done);
- auto st = _exec_env->load_channel_mgr()->cancel(*request);
- if (!st.ok()) {
- LOG(WARNING) << "tablet writer cancel failed, id=" << request->id()
- << ", index_id=" << request->index_id()
- << ", sender_id=" << request->sender_id();
- }
+ DorisMetrics::instance()->tablet_writer_cancel->increment(1);
+ _light_work_pool.offer([this, controller, request, response, done]() {
Review Comment:
warning: lambda capture 'response' is not used
[clang-diagnostic-unused-lambda-capture]
```suggestion
_light_work_pool.offer([this, controller, request, done]() {
```
##########
be/src/service/internal_service.cpp:
##########
@@ -286,125 +334,134 @@
}
}
-void
PInternalServiceImpl::cancel_plan_fragment(google::protobuf::RpcController*
cntl_base,
+void
PInternalServiceImpl::cancel_plan_fragment(google::protobuf::RpcController*
controller,
const
PCancelPlanFragmentRequest* request,
PCancelPlanFragmentResult*
result,
google::protobuf::Closure*
done) {
- auto span = telemetry::start_rpc_server_span("exec_plan_fragment_start",
cntl_base);
- auto scope = OpentelemetryScope {span};
- brpc::ClosureGuard closure_guard(done);
- TUniqueId tid;
- tid.__set_hi(request->finst_id().hi());
- tid.__set_lo(request->finst_id().lo());
-
- Status st = Status::OK();
- if (request->has_cancel_reason()) {
- LOG(INFO) << "cancel fragment, fragment_instance_id=" << print_id(tid)
- << ", reason: " << request->cancel_reason();
- _exec_env->fragment_mgr()->cancel(tid, request->cancel_reason());
- } else {
- LOG(INFO) << "cancel fragment, fragment_instance_id=" << print_id(tid);
- _exec_env->fragment_mgr()->cancel(tid);
- }
-
- // TODO: the logic seems useless, cancel only return Status::OK. remove it
- st.to_protobuf(result->mutable_status());
+ DorisMetrics::instance()->cancel_plan_fragment->increment(1);
+ _light_work_pool.offer([this, controller, request, result, done]() {
+ auto span =
telemetry::start_rpc_server_span("exec_plan_fragment_start", controller);
+ auto scope = OpentelemetryScope {span};
+ brpc::ClosureGuard closure_guard(done);
+ TUniqueId tid;
+ tid.__set_hi(request->finst_id().hi());
+ tid.__set_lo(request->finst_id().lo());
+
+ Status st = Status::OK();
+ if (request->has_cancel_reason()) {
+ LOG(INFO) << "cancel fragment, fragment_instance_id=" <<
print_id(tid)
+ << ", reason: " << request->cancel_reason();
+ _exec_env->fragment_mgr()->cancel(tid, request->cancel_reason());
+ } else {
+ LOG(INFO) << "cancel fragment, fragment_instance_id=" <<
print_id(tid);
+ _exec_env->fragment_mgr()->cancel(tid);
+ }
+ // TODO: the logic seems useless, cancel only return Status::OK.
remove it
+ st.to_protobuf(result->mutable_status());
+ });
}
-void PInternalServiceImpl::fetch_data(google::protobuf::RpcController*
cntl_base,
+void PInternalServiceImpl::fetch_data(google::protobuf::RpcController*
controller,
const PFetchDataRequest* request,
PFetchDataResult* result,
google::protobuf::Closure* done) {
- brpc::Controller* cntl = static_cast<brpc::Controller*>(cntl_base);
- GetResultBatchCtx* ctx = new GetResultBatchCtx(cntl, result, done);
- _exec_env->result_mgr()->fetch_data(request->finst_id(), ctx);
+ DorisMetrics::instance()->fetch_data->increment(1);
+ _heavy_work_pool.offer([this, controller, request, result, done]() {
+ brpc::Controller* cntl = static_cast<brpc::Controller*>(controller);
+ GetResultBatchCtx* ctx = new GetResultBatchCtx(cntl, result, done);
+ _exec_env->result_mgr()->fetch_data(request->finst_id(), ctx);
+ });
}
void PInternalServiceImpl::fetch_table_schema(google::protobuf::RpcController*
controller,
const PFetchTableSchemaRequest*
request,
PFetchTableSchemaResult* result,
google::protobuf::Closure* done)
{
- VLOG_RPC << "fetch table schema";
- brpc::ClosureGuard closure_guard(done);
- TFileScanRange file_scan_range;
- Status st = Status::OK();
- {
- const uint8_t* buf = (const
uint8_t*)(request->file_scan_range().data());
- uint32_t len = request->file_scan_range().size();
- st = deserialize_thrift_msg(buf, &len, false, &file_scan_range);
+ DorisMetrics::instance()->fetch_table_schema->increment(1);
+ _light_work_pool.offer([this, controller, request, result, done]() {
Review Comment:
warning: lambda capture 'controller' is not used
[clang-diagnostic-unused-lambda-capture]
```suggestion
_light_work_pool.offer([this, request, result, done]() {
```
##########
be/src/service/internal_service.cpp:
##########
@@ -423,200 +480,233 @@
const PTabletKeyLookupRequest*
request,
PTabletKeyLookupResponse*
response,
google::protobuf::Closure* done) {
- [[maybe_unused]] brpc::Controller* cntl =
static_cast<brpc::Controller*>(controller);
- brpc::ClosureGuard guard(done);
- Status st = _tablet_fetch_data(request, response);
- st.to_protobuf(response->mutable_status());
+ DorisMetrics::instance()->tablet_fetch_data->increment(1);
+ _heavy_work_pool.offer([this, controller, request, response, done]() {
+ [[maybe_unused]] brpc::Controller* cntl =
static_cast<brpc::Controller*>(controller);
+ brpc::ClosureGuard guard(done);
+ Status st = _tablet_fetch_data(request, response);
+ st.to_protobuf(response->mutable_status());
+ });
}
void PInternalServiceImpl::get_info(google::protobuf::RpcController*
controller,
const PProxyRequest* request,
PProxyResult* response,
google::protobuf::Closure* done) {
- brpc::ClosureGuard closure_guard(done);
- // PProxyRequest is defined in gensrc/proto/internal_service.proto
- // Currently it supports 2 kinds of requests:
- // 1. get all kafka partition ids for given topic
- // 2. get all kafka partition offsets for given topic and timestamp.
- if (request->has_kafka_meta_request()) {
- const PKafkaMetaProxyRequest& kafka_request =
request->kafka_meta_request();
- if (!kafka_request.partition_id_for_latest_offsets().empty()) {
- // get latest offsets for specified partition ids
- std::vector<PIntegerPair> partition_offsets;
- Status st = _exec_env->routine_load_task_executor()
- ->get_kafka_latest_offsets_for_partitions(
- request->kafka_meta_request(),
&partition_offsets);
- if (st.ok()) {
- PKafkaPartitionOffsets* part_offsets =
response->mutable_partition_offsets();
- for (const auto& entry : partition_offsets) {
- PIntegerPair* res = part_offsets->add_offset_times();
- res->set_key(entry.key());
- res->set_val(entry.val());
+ DorisMetrics::instance()->get_info->increment(1);
+ _light_work_pool.offer([this, controller, request, response, done]() {
+ brpc::ClosureGuard closure_guard(done);
+ // PProxyRequest is defined in gensrc/proto/internal_service.proto
+ // Currently it supports 2 kinds of requests:
+ // 1. get all kafka partition ids for given topic
+ // 2. get all kafka partition offsets for given topic and timestamp.
+ if (request->has_kafka_meta_request()) {
+ const PKafkaMetaProxyRequest& kafka_request =
request->kafka_meta_request();
+ if (!kafka_request.partition_id_for_latest_offsets().empty()) {
+ // get latest offsets for specified partition ids
+ std::vector<PIntegerPair> partition_offsets;
+ Status st = _exec_env->routine_load_task_executor()
+ ->get_kafka_latest_offsets_for_partitions(
+ request->kafka_meta_request(),
&partition_offsets);
+ if (st.ok()) {
+ PKafkaPartitionOffsets* part_offsets =
response->mutable_partition_offsets();
+ for (const auto& entry : partition_offsets) {
+ PIntegerPair* res = part_offsets->add_offset_times();
+ res->set_key(entry.key());
+ res->set_val(entry.val());
+ }
}
- }
- st.to_protobuf(response->mutable_status());
- return;
- } else if (!kafka_request.offset_times().empty()) {
- // if offset_times() has elements, which means this request is to
get offset by timestamp.
- std::vector<PIntegerPair> partition_offsets;
- Status st =
-
_exec_env->routine_load_task_executor()->get_kafka_partition_offsets_for_times(
- request->kafka_meta_request(), &partition_offsets);
- if (st.ok()) {
- PKafkaPartitionOffsets* part_offsets =
response->mutable_partition_offsets();
- for (const auto& entry : partition_offsets) {
- PIntegerPair* res = part_offsets->add_offset_times();
- res->set_key(entry.key());
- res->set_val(entry.val());
+ st.to_protobuf(response->mutable_status());
+ return;
+ } else if (!kafka_request.offset_times().empty()) {
+ // if offset_times() has elements, which means this request is
to get offset by timestamp.
+ std::vector<PIntegerPair> partition_offsets;
+ Status st = _exec_env->routine_load_task_executor()
+ ->get_kafka_partition_offsets_for_times(
+ request->kafka_meta_request(),
&partition_offsets);
+ if (st.ok()) {
+ PKafkaPartitionOffsets* part_offsets =
response->mutable_partition_offsets();
+ for (const auto& entry : partition_offsets) {
+ PIntegerPair* res = part_offsets->add_offset_times();
+ res->set_key(entry.key());
+ res->set_val(entry.val());
+ }
}
- }
- st.to_protobuf(response->mutable_status());
- return;
- } else {
- // get partition ids of topic
- std::vector<int32_t> partition_ids;
- Status st =
_exec_env->routine_load_task_executor()->get_kafka_partition_meta(
- request->kafka_meta_request(), &partition_ids);
- if (st.ok()) {
- PKafkaMetaProxyResult* kafka_result =
response->mutable_kafka_meta_result();
- for (int32_t id : partition_ids) {
- kafka_result->add_partition_ids(id);
+ st.to_protobuf(response->mutable_status());
+ return;
+ } else {
+ // get partition ids of topic
+ std::vector<int32_t> partition_ids;
+ Status st =
_exec_env->routine_load_task_executor()->get_kafka_partition_meta(
+ request->kafka_meta_request(), &partition_ids);
+ if (st.ok()) {
+ PKafkaMetaProxyResult* kafka_result =
response->mutable_kafka_meta_result();
+ for (int32_t id : partition_ids) {
+ kafka_result->add_partition_ids(id);
+ }
}
+ st.to_protobuf(response->mutable_status());
+ return;
}
- st.to_protobuf(response->mutable_status());
- return;
}
- }
- Status::OK().to_protobuf(response->mutable_status());
+ Status::OK().to_protobuf(response->mutable_status());
+ });
}
void PInternalServiceImpl::update_cache(google::protobuf::RpcController*
controller,
const PUpdateCacheRequest* request,
PCacheResponse* response,
google::protobuf::Closure* done) {
- brpc::ClosureGuard closure_guard(done);
- _exec_env->result_cache()->update(request, response);
+ DorisMetrics::instance()->update_cache->increment(1);
+ _light_work_pool.offer([this, controller, request, response, done]() {
+ brpc::ClosureGuard closure_guard(done);
+ _exec_env->result_cache()->update(request, response);
+ });
}
void PInternalServiceImpl::fetch_cache(google::protobuf::RpcController*
controller,
const PFetchCacheRequest* request,
PFetchCacheResult* result,
google::protobuf::Closure* done) {
- brpc::ClosureGuard closure_guard(done);
- _exec_env->result_cache()->fetch(request, result);
+ DorisMetrics::instance()->fetch_cache->increment(1);
+ _heavy_work_pool.offer([this, controller, request, result, done]() {
+ brpc::ClosureGuard closure_guard(done);
+ _exec_env->result_cache()->fetch(request, result);
+ });
}
void PInternalServiceImpl::clear_cache(google::protobuf::RpcController*
controller,
const PClearCacheRequest* request,
PCacheResponse* response,
google::protobuf::Closure* done) {
- brpc::ClosureGuard closure_guard(done);
- _exec_env->result_cache()->clear(request, response);
+ DorisMetrics::instance()->clear_cache->increment(1);
+ _light_work_pool.offer([this, controller, request, response, done]() {
Review Comment:
warning: lambda capture 'controller' is not used
[clang-diagnostic-unused-lambda-capture]
```suggestion
_light_work_pool.offer([this, request, response, done]() {
```
##########
be/src/service/internal_service.cpp:
##########
@@ -423,200 +480,233 @@
const PTabletKeyLookupRequest*
request,
PTabletKeyLookupResponse*
response,
google::protobuf::Closure* done) {
- [[maybe_unused]] brpc::Controller* cntl =
static_cast<brpc::Controller*>(controller);
- brpc::ClosureGuard guard(done);
- Status st = _tablet_fetch_data(request, response);
- st.to_protobuf(response->mutable_status());
+ DorisMetrics::instance()->tablet_fetch_data->increment(1);
+ _heavy_work_pool.offer([this, controller, request, response, done]() {
+ [[maybe_unused]] brpc::Controller* cntl =
static_cast<brpc::Controller*>(controller);
+ brpc::ClosureGuard guard(done);
+ Status st = _tablet_fetch_data(request, response);
+ st.to_protobuf(response->mutable_status());
+ });
}
void PInternalServiceImpl::get_info(google::protobuf::RpcController*
controller,
const PProxyRequest* request,
PProxyResult* response,
google::protobuf::Closure* done) {
- brpc::ClosureGuard closure_guard(done);
- // PProxyRequest is defined in gensrc/proto/internal_service.proto
- // Currently it supports 2 kinds of requests:
- // 1. get all kafka partition ids for given topic
- // 2. get all kafka partition offsets for given topic and timestamp.
- if (request->has_kafka_meta_request()) {
- const PKafkaMetaProxyRequest& kafka_request =
request->kafka_meta_request();
- if (!kafka_request.partition_id_for_latest_offsets().empty()) {
- // get latest offsets for specified partition ids
- std::vector<PIntegerPair> partition_offsets;
- Status st = _exec_env->routine_load_task_executor()
- ->get_kafka_latest_offsets_for_partitions(
- request->kafka_meta_request(),
&partition_offsets);
- if (st.ok()) {
- PKafkaPartitionOffsets* part_offsets =
response->mutable_partition_offsets();
- for (const auto& entry : partition_offsets) {
- PIntegerPair* res = part_offsets->add_offset_times();
- res->set_key(entry.key());
- res->set_val(entry.val());
+ DorisMetrics::instance()->get_info->increment(1);
+ _light_work_pool.offer([this, controller, request, response, done]() {
+ brpc::ClosureGuard closure_guard(done);
+ // PProxyRequest is defined in gensrc/proto/internal_service.proto
+ // Currently it supports 2 kinds of requests:
+ // 1. get all kafka partition ids for given topic
+ // 2. get all kafka partition offsets for given topic and timestamp.
+ if (request->has_kafka_meta_request()) {
+ const PKafkaMetaProxyRequest& kafka_request =
request->kafka_meta_request();
+ if (!kafka_request.partition_id_for_latest_offsets().empty()) {
+ // get latest offsets for specified partition ids
+ std::vector<PIntegerPair> partition_offsets;
+ Status st = _exec_env->routine_load_task_executor()
+ ->get_kafka_latest_offsets_for_partitions(
+ request->kafka_meta_request(),
&partition_offsets);
+ if (st.ok()) {
+ PKafkaPartitionOffsets* part_offsets =
response->mutable_partition_offsets();
+ for (const auto& entry : partition_offsets) {
+ PIntegerPair* res = part_offsets->add_offset_times();
+ res->set_key(entry.key());
+ res->set_val(entry.val());
+ }
}
- }
- st.to_protobuf(response->mutable_status());
- return;
- } else if (!kafka_request.offset_times().empty()) {
- // if offset_times() has elements, which means this request is to
get offset by timestamp.
- std::vector<PIntegerPair> partition_offsets;
- Status st =
-
_exec_env->routine_load_task_executor()->get_kafka_partition_offsets_for_times(
- request->kafka_meta_request(), &partition_offsets);
- if (st.ok()) {
- PKafkaPartitionOffsets* part_offsets =
response->mutable_partition_offsets();
- for (const auto& entry : partition_offsets) {
- PIntegerPair* res = part_offsets->add_offset_times();
- res->set_key(entry.key());
- res->set_val(entry.val());
+ st.to_protobuf(response->mutable_status());
+ return;
+ } else if (!kafka_request.offset_times().empty()) {
+ // if offset_times() has elements, which means this request is
to get offset by timestamp.
+ std::vector<PIntegerPair> partition_offsets;
+ Status st = _exec_env->routine_load_task_executor()
+ ->get_kafka_partition_offsets_for_times(
+ request->kafka_meta_request(),
&partition_offsets);
+ if (st.ok()) {
+ PKafkaPartitionOffsets* part_offsets =
response->mutable_partition_offsets();
+ for (const auto& entry : partition_offsets) {
+ PIntegerPair* res = part_offsets->add_offset_times();
+ res->set_key(entry.key());
+ res->set_val(entry.val());
+ }
}
- }
- st.to_protobuf(response->mutable_status());
- return;
- } else {
- // get partition ids of topic
- std::vector<int32_t> partition_ids;
- Status st =
_exec_env->routine_load_task_executor()->get_kafka_partition_meta(
- request->kafka_meta_request(), &partition_ids);
- if (st.ok()) {
- PKafkaMetaProxyResult* kafka_result =
response->mutable_kafka_meta_result();
- for (int32_t id : partition_ids) {
- kafka_result->add_partition_ids(id);
+ st.to_protobuf(response->mutable_status());
+ return;
+ } else {
+ // get partition ids of topic
+ std::vector<int32_t> partition_ids;
+ Status st =
_exec_env->routine_load_task_executor()->get_kafka_partition_meta(
+ request->kafka_meta_request(), &partition_ids);
+ if (st.ok()) {
+ PKafkaMetaProxyResult* kafka_result =
response->mutable_kafka_meta_result();
+ for (int32_t id : partition_ids) {
+ kafka_result->add_partition_ids(id);
+ }
}
+ st.to_protobuf(response->mutable_status());
+ return;
}
- st.to_protobuf(response->mutable_status());
- return;
}
- }
- Status::OK().to_protobuf(response->mutable_status());
+ Status::OK().to_protobuf(response->mutable_status());
+ });
}
void PInternalServiceImpl::update_cache(google::protobuf::RpcController*
controller,
const PUpdateCacheRequest* request,
PCacheResponse* response,
google::protobuf::Closure* done) {
- brpc::ClosureGuard closure_guard(done);
- _exec_env->result_cache()->update(request, response);
+ DorisMetrics::instance()->update_cache->increment(1);
+ _light_work_pool.offer([this, controller, request, response, done]() {
+ brpc::ClosureGuard closure_guard(done);
+ _exec_env->result_cache()->update(request, response);
+ });
}
void PInternalServiceImpl::fetch_cache(google::protobuf::RpcController*
controller,
const PFetchCacheRequest* request,
PFetchCacheResult* result,
google::protobuf::Closure* done) {
- brpc::ClosureGuard closure_guard(done);
- _exec_env->result_cache()->fetch(request, result);
+ DorisMetrics::instance()->fetch_cache->increment(1);
+ _heavy_work_pool.offer([this, controller, request, result, done]() {
Review Comment:
warning: lambda capture 'controller' is not used
[clang-diagnostic-unused-lambda-capture]
```suggestion
_heavy_work_pool.offer([this, request, result, done]() {
```
##########
be/src/service/internal_service.cpp:
##########
@@ -247,15 +292,18 @@ void
PInternalServiceImpl::tablet_writer_cancel(google::protobuf::RpcController*
const
PTabletWriterCancelRequest* request,
PTabletWriterCancelResult*
response,
google::protobuf::Closure*
done) {
- VLOG_RPC << "tablet writer cancel, id=" << request->id() << ", index_id="
<< request->index_id()
- << ", sender_id=" << request->sender_id();
- brpc::ClosureGuard closure_guard(done);
- auto st = _exec_env->load_channel_mgr()->cancel(*request);
- if (!st.ok()) {
- LOG(WARNING) << "tablet writer cancel failed, id=" << request->id()
- << ", index_id=" << request->index_id()
- << ", sender_id=" << request->sender_id();
- }
+ DorisMetrics::instance()->tablet_writer_cancel->increment(1);
+ _light_work_pool.offer([this, controller, request, response, done]() {
Review Comment:
warning: lambda capture 'controller' is not used
[clang-diagnostic-unused-lambda-capture]
```suggestion
_light_work_pool.offer([this, request, response, done]() {
```
##########
be/src/service/internal_service.cpp:
##########
@@ -423,200 +480,233 @@
const PTabletKeyLookupRequest*
request,
PTabletKeyLookupResponse*
response,
google::protobuf::Closure* done) {
- [[maybe_unused]] brpc::Controller* cntl =
static_cast<brpc::Controller*>(controller);
- brpc::ClosureGuard guard(done);
- Status st = _tablet_fetch_data(request, response);
- st.to_protobuf(response->mutable_status());
+ DorisMetrics::instance()->tablet_fetch_data->increment(1);
+ _heavy_work_pool.offer([this, controller, request, response, done]() {
+ [[maybe_unused]] brpc::Controller* cntl =
static_cast<brpc::Controller*>(controller);
+ brpc::ClosureGuard guard(done);
+ Status st = _tablet_fetch_data(request, response);
+ st.to_protobuf(response->mutable_status());
+ });
}
void PInternalServiceImpl::get_info(google::protobuf::RpcController*
controller,
const PProxyRequest* request,
PProxyResult* response,
google::protobuf::Closure* done) {
- brpc::ClosureGuard closure_guard(done);
- // PProxyRequest is defined in gensrc/proto/internal_service.proto
- // Currently it supports 2 kinds of requests:
- // 1. get all kafka partition ids for given topic
- // 2. get all kafka partition offsets for given topic and timestamp.
- if (request->has_kafka_meta_request()) {
- const PKafkaMetaProxyRequest& kafka_request =
request->kafka_meta_request();
- if (!kafka_request.partition_id_for_latest_offsets().empty()) {
- // get latest offsets for specified partition ids
- std::vector<PIntegerPair> partition_offsets;
- Status st = _exec_env->routine_load_task_executor()
- ->get_kafka_latest_offsets_for_partitions(
- request->kafka_meta_request(),
&partition_offsets);
- if (st.ok()) {
- PKafkaPartitionOffsets* part_offsets =
response->mutable_partition_offsets();
- for (const auto& entry : partition_offsets) {
- PIntegerPair* res = part_offsets->add_offset_times();
- res->set_key(entry.key());
- res->set_val(entry.val());
+ DorisMetrics::instance()->get_info->increment(1);
+ _light_work_pool.offer([this, controller, request, response, done]() {
+ brpc::ClosureGuard closure_guard(done);
+ // PProxyRequest is defined in gensrc/proto/internal_service.proto
+ // Currently it supports 2 kinds of requests:
+ // 1. get all kafka partition ids for given topic
+ // 2. get all kafka partition offsets for given topic and timestamp.
+ if (request->has_kafka_meta_request()) {
+ const PKafkaMetaProxyRequest& kafka_request =
request->kafka_meta_request();
+ if (!kafka_request.partition_id_for_latest_offsets().empty()) {
+ // get latest offsets for specified partition ids
+ std::vector<PIntegerPair> partition_offsets;
+ Status st = _exec_env->routine_load_task_executor()
+ ->get_kafka_latest_offsets_for_partitions(
+ request->kafka_meta_request(),
&partition_offsets);
+ if (st.ok()) {
+ PKafkaPartitionOffsets* part_offsets =
response->mutable_partition_offsets();
+ for (const auto& entry : partition_offsets) {
+ PIntegerPair* res = part_offsets->add_offset_times();
+ res->set_key(entry.key());
+ res->set_val(entry.val());
+ }
}
- }
- st.to_protobuf(response->mutable_status());
- return;
- } else if (!kafka_request.offset_times().empty()) {
- // if offset_times() has elements, which means this request is to
get offset by timestamp.
- std::vector<PIntegerPair> partition_offsets;
- Status st =
-
_exec_env->routine_load_task_executor()->get_kafka_partition_offsets_for_times(
- request->kafka_meta_request(), &partition_offsets);
- if (st.ok()) {
- PKafkaPartitionOffsets* part_offsets =
response->mutable_partition_offsets();
- for (const auto& entry : partition_offsets) {
- PIntegerPair* res = part_offsets->add_offset_times();
- res->set_key(entry.key());
- res->set_val(entry.val());
+ st.to_protobuf(response->mutable_status());
+ return;
+ } else if (!kafka_request.offset_times().empty()) {
+ // if offset_times() has elements, which means this request is
to get offset by timestamp.
+ std::vector<PIntegerPair> partition_offsets;
+ Status st = _exec_env->routine_load_task_executor()
+ ->get_kafka_partition_offsets_for_times(
+ request->kafka_meta_request(),
&partition_offsets);
+ if (st.ok()) {
+ PKafkaPartitionOffsets* part_offsets =
response->mutable_partition_offsets();
+ for (const auto& entry : partition_offsets) {
+ PIntegerPair* res = part_offsets->add_offset_times();
+ res->set_key(entry.key());
+ res->set_val(entry.val());
+ }
}
- }
- st.to_protobuf(response->mutable_status());
- return;
- } else {
- // get partition ids of topic
- std::vector<int32_t> partition_ids;
- Status st =
_exec_env->routine_load_task_executor()->get_kafka_partition_meta(
- request->kafka_meta_request(), &partition_ids);
- if (st.ok()) {
- PKafkaMetaProxyResult* kafka_result =
response->mutable_kafka_meta_result();
- for (int32_t id : partition_ids) {
- kafka_result->add_partition_ids(id);
+ st.to_protobuf(response->mutable_status());
+ return;
+ } else {
+ // get partition ids of topic
+ std::vector<int32_t> partition_ids;
+ Status st =
_exec_env->routine_load_task_executor()->get_kafka_partition_meta(
+ request->kafka_meta_request(), &partition_ids);
+ if (st.ok()) {
+ PKafkaMetaProxyResult* kafka_result =
response->mutable_kafka_meta_result();
+ for (int32_t id : partition_ids) {
+ kafka_result->add_partition_ids(id);
+ }
}
+ st.to_protobuf(response->mutable_status());
+ return;
}
- st.to_protobuf(response->mutable_status());
- return;
}
- }
- Status::OK().to_protobuf(response->mutable_status());
+ Status::OK().to_protobuf(response->mutable_status());
+ });
}
void PInternalServiceImpl::update_cache(google::protobuf::RpcController*
controller,
const PUpdateCacheRequest* request,
PCacheResponse* response,
google::protobuf::Closure* done) {
- brpc::ClosureGuard closure_guard(done);
- _exec_env->result_cache()->update(request, response);
+ DorisMetrics::instance()->update_cache->increment(1);
+ _light_work_pool.offer([this, controller, request, response, done]() {
Review Comment:
warning: lambda capture 'controller' is not used
[clang-diagnostic-unused-lambda-capture]
```suggestion
_light_work_pool.offer([this, request, response, done]() {
```
##########
be/src/service/internal_service.cpp:
##########
@@ -423,200 +480,233 @@
const PTabletKeyLookupRequest*
request,
PTabletKeyLookupResponse*
response,
google::protobuf::Closure* done) {
- [[maybe_unused]] brpc::Controller* cntl =
static_cast<brpc::Controller*>(controller);
- brpc::ClosureGuard guard(done);
- Status st = _tablet_fetch_data(request, response);
- st.to_protobuf(response->mutable_status());
+ DorisMetrics::instance()->tablet_fetch_data->increment(1);
+ _heavy_work_pool.offer([this, controller, request, response, done]() {
+ [[maybe_unused]] brpc::Controller* cntl =
static_cast<brpc::Controller*>(controller);
+ brpc::ClosureGuard guard(done);
+ Status st = _tablet_fetch_data(request, response);
+ st.to_protobuf(response->mutable_status());
+ });
}
void PInternalServiceImpl::get_info(google::protobuf::RpcController*
controller,
const PProxyRequest* request,
PProxyResult* response,
google::protobuf::Closure* done) {
- brpc::ClosureGuard closure_guard(done);
- // PProxyRequest is defined in gensrc/proto/internal_service.proto
- // Currently it supports 2 kinds of requests:
- // 1. get all kafka partition ids for given topic
- // 2. get all kafka partition offsets for given topic and timestamp.
- if (request->has_kafka_meta_request()) {
- const PKafkaMetaProxyRequest& kafka_request =
request->kafka_meta_request();
- if (!kafka_request.partition_id_for_latest_offsets().empty()) {
- // get latest offsets for specified partition ids
- std::vector<PIntegerPair> partition_offsets;
- Status st = _exec_env->routine_load_task_executor()
- ->get_kafka_latest_offsets_for_partitions(
- request->kafka_meta_request(),
&partition_offsets);
- if (st.ok()) {
- PKafkaPartitionOffsets* part_offsets =
response->mutable_partition_offsets();
- for (const auto& entry : partition_offsets) {
- PIntegerPair* res = part_offsets->add_offset_times();
- res->set_key(entry.key());
- res->set_val(entry.val());
+ DorisMetrics::instance()->get_info->increment(1);
+ _light_work_pool.offer([this, controller, request, response, done]() {
Review Comment:
warning: lambda capture 'controller' is not used
[clang-diagnostic-unused-lambda-capture]
```suggestion
_light_work_pool.offer([this, request, response, done]() {
```
##########
be/src/service/internal_service.cpp:
##########
@@ -286,125 +334,134 @@
}
}
-void
PInternalServiceImpl::cancel_plan_fragment(google::protobuf::RpcController*
cntl_base,
+void
PInternalServiceImpl::cancel_plan_fragment(google::protobuf::RpcController*
controller,
const
PCancelPlanFragmentRequest* request,
PCancelPlanFragmentResult*
result,
google::protobuf::Closure*
done) {
- auto span = telemetry::start_rpc_server_span("exec_plan_fragment_start",
cntl_base);
- auto scope = OpentelemetryScope {span};
- brpc::ClosureGuard closure_guard(done);
- TUniqueId tid;
- tid.__set_hi(request->finst_id().hi());
- tid.__set_lo(request->finst_id().lo());
-
- Status st = Status::OK();
- if (request->has_cancel_reason()) {
- LOG(INFO) << "cancel fragment, fragment_instance_id=" << print_id(tid)
- << ", reason: " << request->cancel_reason();
- _exec_env->fragment_mgr()->cancel(tid, request->cancel_reason());
- } else {
- LOG(INFO) << "cancel fragment, fragment_instance_id=" << print_id(tid);
- _exec_env->fragment_mgr()->cancel(tid);
- }
-
- // TODO: the logic seems useless, cancel only return Status::OK. remove it
- st.to_protobuf(result->mutable_status());
+ DorisMetrics::instance()->cancel_plan_fragment->increment(1);
+ _light_work_pool.offer([this, controller, request, result, done]() {
+ auto span =
telemetry::start_rpc_server_span("exec_plan_fragment_start", controller);
+ auto scope = OpentelemetryScope {span};
+ brpc::ClosureGuard closure_guard(done);
+ TUniqueId tid;
+ tid.__set_hi(request->finst_id().hi());
+ tid.__set_lo(request->finst_id().lo());
+
+ Status st = Status::OK();
+ if (request->has_cancel_reason()) {
+ LOG(INFO) << "cancel fragment, fragment_instance_id=" <<
print_id(tid)
+ << ", reason: " << request->cancel_reason();
+ _exec_env->fragment_mgr()->cancel(tid, request->cancel_reason());
+ } else {
+ LOG(INFO) << "cancel fragment, fragment_instance_id=" <<
print_id(tid);
+ _exec_env->fragment_mgr()->cancel(tid);
+ }
+ // TODO: the logic seems useless, cancel only return Status::OK.
remove it
+ st.to_protobuf(result->mutable_status());
+ });
}
-void PInternalServiceImpl::fetch_data(google::protobuf::RpcController*
cntl_base,
+void PInternalServiceImpl::fetch_data(google::protobuf::RpcController*
controller,
const PFetchDataRequest* request,
PFetchDataResult* result,
google::protobuf::Closure* done) {
- brpc::Controller* cntl = static_cast<brpc::Controller*>(cntl_base);
- GetResultBatchCtx* ctx = new GetResultBatchCtx(cntl, result, done);
- _exec_env->result_mgr()->fetch_data(request->finst_id(), ctx);
+ DorisMetrics::instance()->fetch_data->increment(1);
+ _heavy_work_pool.offer([this, controller, request, result, done]() {
+ brpc::Controller* cntl = static_cast<brpc::Controller*>(controller);
+ GetResultBatchCtx* ctx = new GetResultBatchCtx(cntl, result, done);
+ _exec_env->result_mgr()->fetch_data(request->finst_id(), ctx);
+ });
}
void PInternalServiceImpl::fetch_table_schema(google::protobuf::RpcController*
controller,
const PFetchTableSchemaRequest*
request,
PFetchTableSchemaResult* result,
google::protobuf::Closure* done)
{
- VLOG_RPC << "fetch table schema";
- brpc::ClosureGuard closure_guard(done);
- TFileScanRange file_scan_range;
- Status st = Status::OK();
- {
- const uint8_t* buf = (const
uint8_t*)(request->file_scan_range().data());
- uint32_t len = request->file_scan_range().size();
- st = deserialize_thrift_msg(buf, &len, false, &file_scan_range);
+ DorisMetrics::instance()->fetch_table_schema->increment(1);
+ _light_work_pool.offer([this, controller, request, result, done]() {
Review Comment:
warning: lambda capture 'this' is not used
[clang-diagnostic-unused-lambda-capture]
```suggestion
_light_work_pool.offer([ controller, request, result, done]() {
```
##########
be/src/service/internal_service.cpp:
##########
@@ -693,75 +790,86 @@
const PCheckRPCChannelRequest*
request,
PCheckRPCChannelResponse*
response,
google::protobuf::Closure* done) {
- brpc::ClosureGuard closure_guard(done);
- response->mutable_status()->set_status_code(0);
- if (request->data().size() != request->size()) {
- std::stringstream ss;
- ss << "data size not same, expected: " << request->size()
- << ", actual: " << request->data().size();
- response->mutable_status()->add_error_msgs(ss.str());
- response->mutable_status()->set_status_code(1);
-
- } else {
- Md5Digest digest;
- digest.update(static_cast<const void*>(request->data().c_str()),
request->data().size());
- digest.digest();
- if (!iequal(digest.hex(), request->md5())) {
+ DorisMetrics::instance()->check_rpc_channel->increment(1);
+ _light_work_pool.offer([controller, request, response, done]() {
Review Comment:
warning: lambda capture 'controller' is not used
[clang-diagnostic-unused-lambda-capture]
```suggestion
_light_work_pool.offer([ request, response, done]() {
```
##########
be/src/service/internal_service.cpp:
##########
@@ -693,75 +790,86 @@
const PCheckRPCChannelRequest*
request,
PCheckRPCChannelResponse*
response,
google::protobuf::Closure* done) {
- brpc::ClosureGuard closure_guard(done);
- response->mutable_status()->set_status_code(0);
- if (request->data().size() != request->size()) {
- std::stringstream ss;
- ss << "data size not same, expected: " << request->size()
- << ", actual: " << request->data().size();
- response->mutable_status()->add_error_msgs(ss.str());
- response->mutable_status()->set_status_code(1);
-
- } else {
- Md5Digest digest;
- digest.update(static_cast<const void*>(request->data().c_str()),
request->data().size());
- digest.digest();
- if (!iequal(digest.hex(), request->md5())) {
+ DorisMetrics::instance()->check_rpc_channel->increment(1);
+ _light_work_pool.offer([controller, request, response, done]() {
+ brpc::ClosureGuard closure_guard(done);
+ response->mutable_status()->set_status_code(0);
+ if (request->data().size() != request->size()) {
std::stringstream ss;
- ss << "md5 not same, expected: " << request->md5() << ", actual: "
<< digest.hex();
+ ss << "data size not same, expected: " << request->size()
+ << ", actual: " << request->data().size();
response->mutable_status()->add_error_msgs(ss.str());
response->mutable_status()->set_status_code(1);
+
+ } else {
+ Md5Digest digest;
+ digest.update(static_cast<const void*>(request->data().c_str()),
+ request->data().size());
+ digest.digest();
+ if (!iequal(digest.hex(), request->md5())) {
+ std::stringstream ss;
+ ss << "md5 not same, expected: " << request->md5() << ",
actual: " << digest.hex();
+ response->mutable_status()->add_error_msgs(ss.str());
+ response->mutable_status()->set_status_code(1);
+ }
}
- }
+ });
}
void PInternalServiceImpl::reset_rpc_channel(google::protobuf::RpcController*
controller,
const PResetRPCChannelRequest*
request,
PResetRPCChannelResponse*
response,
google::protobuf::Closure* done) {
- brpc::ClosureGuard closure_guard(done);
- response->mutable_status()->set_status_code(0);
- if (request->all()) {
- int size =
ExecEnv::GetInstance()->brpc_internal_client_cache()->size();
- if (size > 0) {
- std::vector<std::string> endpoints;
-
ExecEnv::GetInstance()->brpc_internal_client_cache()->get_all(&endpoints);
- ExecEnv::GetInstance()->brpc_internal_client_cache()->clear();
- *response->mutable_channels() = {endpoints.begin(),
endpoints.end()};
- }
- } else {
- for (const std::string& endpoint : request->endpoints()) {
- if
(!ExecEnv::GetInstance()->brpc_internal_client_cache()->exist(endpoint)) {
- response->mutable_status()->add_error_msgs(endpoint + ": not
found.");
- continue;
+ DorisMetrics::instance()->reset_rpc_channel->increment(1);
+ _light_work_pool.offer([this, controller, request, response, done]() {
Review Comment:
warning: lambda capture 'controller' is not used
[clang-diagnostic-unused-lambda-capture]
```suggestion
_light_work_pool.offer([this, request, response, done]() {
```
##########
be/src/service/internal_service.cpp:
##########
@@ -693,75 +790,86 @@
const PCheckRPCChannelRequest*
request,
PCheckRPCChannelResponse*
response,
google::protobuf::Closure* done) {
- brpc::ClosureGuard closure_guard(done);
- response->mutable_status()->set_status_code(0);
- if (request->data().size() != request->size()) {
- std::stringstream ss;
- ss << "data size not same, expected: " << request->size()
- << ", actual: " << request->data().size();
- response->mutable_status()->add_error_msgs(ss.str());
- response->mutable_status()->set_status_code(1);
-
- } else {
- Md5Digest digest;
- digest.update(static_cast<const void*>(request->data().c_str()),
request->data().size());
- digest.digest();
- if (!iequal(digest.hex(), request->md5())) {
+ DorisMetrics::instance()->check_rpc_channel->increment(1);
+ _light_work_pool.offer([controller, request, response, done]() {
+ brpc::ClosureGuard closure_guard(done);
+ response->mutable_status()->set_status_code(0);
+ if (request->data().size() != request->size()) {
std::stringstream ss;
- ss << "md5 not same, expected: " << request->md5() << ", actual: "
<< digest.hex();
+ ss << "data size not same, expected: " << request->size()
+ << ", actual: " << request->data().size();
response->mutable_status()->add_error_msgs(ss.str());
response->mutable_status()->set_status_code(1);
+
+ } else {
+ Md5Digest digest;
+ digest.update(static_cast<const void*>(request->data().c_str()),
+ request->data().size());
+ digest.digest();
+ if (!iequal(digest.hex(), request->md5())) {
+ std::stringstream ss;
+ ss << "md5 not same, expected: " << request->md5() << ",
actual: " << digest.hex();
+ response->mutable_status()->add_error_msgs(ss.str());
+ response->mutable_status()->set_status_code(1);
+ }
}
- }
+ });
}
void PInternalServiceImpl::reset_rpc_channel(google::protobuf::RpcController*
controller,
const PResetRPCChannelRequest*
request,
PResetRPCChannelResponse*
response,
google::protobuf::Closure* done) {
- brpc::ClosureGuard closure_guard(done);
- response->mutable_status()->set_status_code(0);
- if (request->all()) {
- int size =
ExecEnv::GetInstance()->brpc_internal_client_cache()->size();
- if (size > 0) {
- std::vector<std::string> endpoints;
-
ExecEnv::GetInstance()->brpc_internal_client_cache()->get_all(&endpoints);
- ExecEnv::GetInstance()->brpc_internal_client_cache()->clear();
- *response->mutable_channels() = {endpoints.begin(),
endpoints.end()};
- }
- } else {
- for (const std::string& endpoint : request->endpoints()) {
- if
(!ExecEnv::GetInstance()->brpc_internal_client_cache()->exist(endpoint)) {
- response->mutable_status()->add_error_msgs(endpoint + ": not
found.");
- continue;
+ DorisMetrics::instance()->reset_rpc_channel->increment(1);
+ _light_work_pool.offer([this, controller, request, response, done]() {
Review Comment:
warning: lambda capture 'this' is not used
[clang-diagnostic-unused-lambda-capture]
```suggestion
_light_work_pool.offer([ controller, request, response, done]() {
```
##########
be/src/service/internal_service.cpp:
##########
@@ -693,75 +790,86 @@
const PCheckRPCChannelRequest*
request,
PCheckRPCChannelResponse*
response,
google::protobuf::Closure* done) {
- brpc::ClosureGuard closure_guard(done);
- response->mutable_status()->set_status_code(0);
- if (request->data().size() != request->size()) {
- std::stringstream ss;
- ss << "data size not same, expected: " << request->size()
- << ", actual: " << request->data().size();
- response->mutable_status()->add_error_msgs(ss.str());
- response->mutable_status()->set_status_code(1);
-
- } else {
- Md5Digest digest;
- digest.update(static_cast<const void*>(request->data().c_str()),
request->data().size());
- digest.digest();
- if (!iequal(digest.hex(), request->md5())) {
+ DorisMetrics::instance()->check_rpc_channel->increment(1);
+ _light_work_pool.offer([controller, request, response, done]() {
+ brpc::ClosureGuard closure_guard(done);
+ response->mutable_status()->set_status_code(0);
+ if (request->data().size() != request->size()) {
std::stringstream ss;
- ss << "md5 not same, expected: " << request->md5() << ", actual: "
<< digest.hex();
+ ss << "data size not same, expected: " << request->size()
+ << ", actual: " << request->data().size();
response->mutable_status()->add_error_msgs(ss.str());
response->mutable_status()->set_status_code(1);
+
+ } else {
+ Md5Digest digest;
+ digest.update(static_cast<const void*>(request->data().c_str()),
+ request->data().size());
+ digest.digest();
+ if (!iequal(digest.hex(), request->md5())) {
+ std::stringstream ss;
+ ss << "md5 not same, expected: " << request->md5() << ",
actual: " << digest.hex();
+ response->mutable_status()->add_error_msgs(ss.str());
+ response->mutable_status()->set_status_code(1);
+ }
}
- }
+ });
}
void PInternalServiceImpl::reset_rpc_channel(google::protobuf::RpcController*
controller,
const PResetRPCChannelRequest*
request,
PResetRPCChannelResponse*
response,
google::protobuf::Closure* done) {
- brpc::ClosureGuard closure_guard(done);
- response->mutable_status()->set_status_code(0);
- if (request->all()) {
- int size =
ExecEnv::GetInstance()->brpc_internal_client_cache()->size();
- if (size > 0) {
- std::vector<std::string> endpoints;
-
ExecEnv::GetInstance()->brpc_internal_client_cache()->get_all(&endpoints);
- ExecEnv::GetInstance()->brpc_internal_client_cache()->clear();
- *response->mutable_channels() = {endpoints.begin(),
endpoints.end()};
- }
- } else {
- for (const std::string& endpoint : request->endpoints()) {
- if
(!ExecEnv::GetInstance()->brpc_internal_client_cache()->exist(endpoint)) {
- response->mutable_status()->add_error_msgs(endpoint + ": not
found.");
- continue;
+ DorisMetrics::instance()->reset_rpc_channel->increment(1);
+ _light_work_pool.offer([this, controller, request, response, done]() {
+ brpc::ClosureGuard closure_guard(done);
+ response->mutable_status()->set_status_code(0);
+ if (request->all()) {
+ int size =
ExecEnv::GetInstance()->brpc_internal_client_cache()->size();
+ if (size > 0) {
+ std::vector<std::string> endpoints;
+
ExecEnv::GetInstance()->brpc_internal_client_cache()->get_all(&endpoints);
+ ExecEnv::GetInstance()->brpc_internal_client_cache()->clear();
+ *response->mutable_channels() = {endpoints.begin(),
endpoints.end()};
}
+ } else {
+ for (const std::string& endpoint : request->endpoints()) {
+ if
(!ExecEnv::GetInstance()->brpc_internal_client_cache()->exist(endpoint)) {
+ response->mutable_status()->add_error_msgs(endpoint + ":
not found.");
+ continue;
+ }
- if
(ExecEnv::GetInstance()->brpc_internal_client_cache()->erase(endpoint)) {
- response->add_channels(endpoint);
- } else {
- response->mutable_status()->add_error_msgs(endpoint + ": reset
failed.");
+ if
(ExecEnv::GetInstance()->brpc_internal_client_cache()->erase(endpoint)) {
+ response->add_channels(endpoint);
+ } else {
+ response->mutable_status()->add_error_msgs(endpoint + ":
reset failed.");
+ }
+ }
+ if (request->endpoints_size() != response->channels_size()) {
+ response->mutable_status()->set_status_code(1);
}
}
- if (request->endpoints_size() != response->channels_size()) {
- response->mutable_status()->set_status_code(1);
- }
- }
+ });
}
-void PInternalServiceImpl::hand_shake(google::protobuf::RpcController*
cntl_base,
+void PInternalServiceImpl::hand_shake(google::protobuf::RpcController*
controller,
const PHandShakeRequest* request,
PHandShakeResponse* response,
google::protobuf::Closure* done) {
- brpc::ClosureGuard closure_guard(done);
- if (request->has_hello()) {
- response->set_hello(request->hello());
- }
- response->mutable_status()->set_status_code(0);
+ DorisMetrics::instance()->hand_shake->increment(1);
+ _light_work_pool.offer([this, request, response, done]() {
Review Comment:
warning: lambda capture 'this' is not used
[clang-diagnostic-unused-lambda-capture]
```suggestion
_light_work_pool.offer([ request, response, done]() {
```
--
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
