[Chore](runtime-filter) catch Exception from runtime filter rpc (apache#57085)

catch Exception from runtime filter rpc

```cpp
12340; stack trace: ***
  what():  [E6] consumer meet invalid state, Consumer: ([id: 1, state: [DISABLED], type: BLOOM_FILTER, column_type: INT, reason: get disabled from remote], mode: GLOBAL, state: READY, reached_timeout: false, timeout_limit: 600000ms), assumed_states is [NOT_READY, TIMEOUT]

	0#  doris::Exception::Exception(int, std::basic_string_view > const&, bool) at /usr/local/ldb-toolchain-v0.26/bin/../lib/gcc/x86_64-pc-linux-gnu/15/include/g++-v15/bits/unique_ptr.h:193
	1#  doris::Exception::Exception(int, std::basic_string_view > const&) at /usr/local/ldb-toolchain-v0.26/bin/../lib/gcc/x86_64-pc-linux-gnu/15/include/g++-v15/bits/basic_string.h:239
	2#  doris::Exception::Exception, std::allocator >, std::__cxx11::basic_string, std::allocator > >(int, std::basic_string_view > const&, std::__cxx11::basic_string, std::allocator >&&, std::__cxx11::basic_string, std::allocator >&&) at /usr/local/ldb-toolchain-v0.26/bin/../lib/gcc/x86_64-pc-linux-gnu/15/include/g++-v15/bits/basic_string.h:239
	3#  doris::RuntimeFilterConsumer::_check_state(std::vector >) at /home/zcp/repo_center/doris_branch-4.0/doris/be/src/runtime_filter/runtime_filter_consumer.h:0
	4#  doris::RuntimeFilterConsumer::_set_state(doris::RuntimeFilterConsumer::State, std::shared_ptr) at /usr/local/ldb-toolchain-v0.26/bin/../lib/gcc/x86_64-pc-linux-gnu/15/include/g++-v15/bits/stl_vector.h:375
	5#  doris::RuntimeFilterConsumer::signal(doris::RuntimeFilter*) at /usr/local/ldb-toolchain-v0.26/bin/../lib/gcc/x86_64-pc-linux-gnu/15/include/g++-v15/bits/shared_ptr_base.h:1068
	6#  doris::FragmentMgr::apply_filterv2(doris::PPublishFilterRequestV2 const*, butil::IOBufAsZeroCopyInputStream*) at /usr/local/ldb-toolchain-v0.26/bin/../lib/gcc/x86_64-pc-linux-gnu/15/include/g++-v15/bits/stl_iterator.h:1103
	7#  std::_Function_handler::_M_invoke(std::_Any_data const&) at /home/zcp/repo_center/doris_branch-4.0/doris/be/src/common/status.h:524
	8#  doris::WorkThreadPool::work_thread(int) at /usr/local/ldb-toolchain-v0.26/bin/../lib/gcc/x86_64-pc-linux-gnu/15/include/g++-v15/bits/atomic_base.h:641
	9#  execute_native_thread_routine
```

Author: BiteTheDDDDt

be/src/service/internal_service.cpp

@@ -1389,7 +1389,12 @@ void PInternalService::merge_filter(::google::protobuf::RpcController* controlle
         brpc::ClosureGuard closure_guard(done);
         auto attachment = static_cast<brpc::Controller*>(controller)->request_attachment();
         butil::IOBufAsZeroCopyInputStream zero_copy_input_stream(attachment);
-        Status st = _exec_env->fragment_mgr()->merge_filter(request, &zero_copy_input_stream);
+        Status st;
+        try {
+            st = _exec_env->fragment_mgr()->merge_filter(request, &zero_copy_input_stream);
+        } catch (Exception& e) {
+            st = e.to_status();
+        }
         st.to_protobuf(response->mutable_status());
     });
     if (!ret) {
@@ -1405,7 +1410,12 @@ void PInternalService::send_filter_size(::google::protobuf::RpcController* contr
     bool ret = _light_work_pool.try_offer([this, request, response, done]() {
         signal::SignalTaskIdKeeper keeper(request->query_id());
         brpc::ClosureGuard closure_guard(done);
-        Status st = _exec_env->fragment_mgr()->send_filter_size(request);
+        Status st;
+        try {
+            st = _exec_env->fragment_mgr()->send_filter_size(request);
+        } catch (Exception& e) {
+            st = e.to_status();
+        }
         st.to_protobuf(response->mutable_status());
     });
     if (!ret) {
@@ -1421,7 +1431,12 @@ void PInternalService::sync_filter_size(::google::protobuf::RpcController* contr
     bool ret = _light_work_pool.try_offer([this, request, response, done]() {
         signal::SignalTaskIdKeeper keeper(request->query_id());
         brpc::ClosureGuard closure_guard(done);
-        Status st = _exec_env->fragment_mgr()->sync_filter_size(request);
+        Status st;
+        try {
+            st = _exec_env->fragment_mgr()->sync_filter_size(request);
+        } catch (Exception& e) {
+            st = e.to_status();
+        }
         st.to_protobuf(response->mutable_status());
     });
     if (!ret) {
@@ -1440,7 +1455,12 @@ void PInternalService::apply_filterv2(::google::protobuf::RpcController* control
         auto attachment = static_cast<brpc::Controller*>(controller)->request_attachment();
         butil::IOBufAsZeroCopyInputStream zero_copy_input_stream(attachment);
         VLOG_NOTICE << "rpc apply_filterv2 recv";
-        Status st = _exec_env->fragment_mgr()->apply_filterv2(request, &zero_copy_input_stream);
+        Status st;
+        try {
+            st = _exec_env->fragment_mgr()->apply_filterv2(request, &zero_copy_input_stream);
+        } catch (Exception& e) {
+            st = e.to_status();
+        }
         if (!st.ok()) {
             LOG(WARNING) << "apply filter meet error: " << st.to_string();
         }