This is an automated email from the ASF dual-hosted git repository.
panxiaolei pushed a commit to branch branch-2.0
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/branch-2.0 by this push:
new 189b90a7b3 [Bug](exchange) init _instance_to_rpc_ctx on register_sink
(#22976) (#22983)
189b90a7b3 is described below
commit 189b90a7b33a9629785abec22b30813078fbf446
Author: Pxl <[email protected]>
AuthorDate: Tue Aug 15 13:12:50 2023 +0800
[Bug](exchange) init _instance_to_rpc_ctx on register_sink (#22976) (#22983)
init _instance_to_rpc_ctx on register_sink
---
be/src/pipeline/exec/exchange_sink_buffer.cpp | 7 +++----
1 file changed, 3 insertions(+), 4 deletions(-)
diff --git a/be/src/pipeline/exec/exchange_sink_buffer.cpp
b/be/src/pipeline/exec/exchange_sink_buffer.cpp
index 7ff0f7b503..bf43c3b860 100644
--- a/be/src/pipeline/exec/exchange_sink_buffer.cpp
+++ b/be/src/pipeline/exec/exchange_sink_buffer.cpp
@@ -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) {
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
