[Chore](runtime-filter) enlarge sync filter size rpc timeout limit

be/src/common/config.cpp

@@ -255,6 +255,7 @@ DEFINE_Int32(doris_scanner_thread_pool_queue_size, "102400");
 // default thrift client connect timeout(in seconds)
 DEFINE_mInt32(thrift_connect_timeout_seconds, "3");
 DEFINE_mInt32(fetch_rpc_timeout_seconds, "30");
+
 // default thrift client retry interval (in milliseconds)
 DEFINE_mInt64(thrift_client_retry_interval_ms, "1000");
 // max message size of thrift request

be/src/exprs/runtime_filter.cpp

@@ -1054,7 +1054,7 @@ class SyncSizeClosure : public AutoReleaseClosure<PSendFilterSizeRequest,
             : Base(req, callback), _dependency(std::move(dependency)) {}
 };
 
-Status IRuntimeFilter::send_filter_size(uint64_t local_filter_size) {
+Status IRuntimeFilter::send_filter_size(RuntimeState* state, uint64_t local_filter_size) {
     DCHECK(is_producer());
 
     if (_need_local_merge) {
@@ -1105,7 +1105,7 @@ Status IRuntimeFilter::send_filter_size(uint64_t local_filter_size) {
 
     request->set_filter_size(local_filter_size);
     request->set_filter_id(_filter_id);
-    callback->cntl_->set_timeout_ms(wait_time_ms());
+    callback->cntl_->set_timeout_ms(std::min(3600, state->execution_timeout()) * 1000);
 
     stub->send_filter_size(closure->cntl_.get(), closure->request_.get(), closure->response_.get(),
                            closure.get());

be/src/exprs/runtime_filter.h

@@ -229,7 +229,7 @@ class IRuntimeFilter {
     // push filter to remote node or push down it to scan_node
     Status publish(bool publish_local = false);
 
-    Status send_filter_size(uint64_t local_filter_size);
+    Status send_filter_size(RuntimeState* state, uint64_t local_filter_size);
 
     RuntimeFilterType type() const { return _runtime_filter_type; }
 

be/src/exprs/runtime_filter_slots.h

@@ -55,7 +55,7 @@ class VRuntimeFilterSlots {
         // send_filter_size may call dependency->sub(), so we call set_dependency firstly for all rf to avoid dependency set_ready repeatedly
         for (auto* runtime_filter : _runtime_filters) {
             if (runtime_filter->need_sync_filter_size()) {
-                RETURN_IF_ERROR(runtime_filter->send_filter_size(hash_table_size));
+                RETURN_IF_ERROR(runtime_filter->send_filter_size(state, hash_table_size));
             }
         }
         return Status::OK();

regression-test/suites/query_p0/join/test_join5.groovy

@@ -16,6 +16,7 @@
 // under the License.
 
 suite("test_join5", "query,p0") {
+    sql "set runtime_filter_wait_time_ms = 5"
     def DBname = "regression_test_join5"
     sql "DROP DATABASE IF EXISTS ${DBname}"
     sql "CREATE DATABASE IF NOT EXISTS ${DBname}"