database/doris
2
0
Fork 0
Code Issues Pull Requests Actions 3 Packages Projects Releases Wiki Activity

[Bug](exchange) init _instance_to_rpc_ctx on register_sink (#22976)

Browse Source 
init _instance_to_rpc_ctx on register_sink
This commit is contained in:
Pxl
2023-08-15 13:02:28 +08:00
committed by GitHub
parent 13d24297a7
commit 34399e2965
1 changed files with 3 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
be/src/pipeline/exec/exchange_sink_buffer.cpp
View File

@ -113,6 +113,7 @@ void ExchangeSinkBuffer::register_sink(TUniqueId fragment_instance_id) {
finst_id.set_hi(fragment_instance_id.hi);
finst_id.set_lo(fragment_instance_id.lo);
_instance_to_sending_by_pipeline[low_id] = true;
_instance_to_rpc_ctx[low_id] = {};
_instance_to_receiver_eof[low_id] = false;
_instance_to_rpc_time[low_id] = 0;
_construct_request(low_id, finst_id);
@ -191,10 +192,8 @@ Status ExchangeSinkBuffer::_send_rpc(InstanceLoId id) {
}
auto* closure = request.channel->get_closure(id, request.eos, nullptr);
ExchangeRpcContext rpc_ctx;
rpc_ctx._closure = closure;
rpc_ctx.is_cancelled = false;
_instance_to_rpc_ctx[id] = rpc_ctx;
_instance_to_rpc_ctx[id]._closure = closure;
_instance_to_rpc_ctx[id].is_cancelled = false;
closure->cntl.set_timeout_ms(request.channel->_brpc_timeout_ms);
if (config::exchange_sink_ignore_eovercrowded) {