[Bug] 5.x版本cpp客户端，同时producer和push_consumer，持续运行，crashed4

[41405465]

Before Creating the Bug Report

• I found a bug, not just asking a question, which should be created in GitHub Discussions.

• I have searched the GitHub Issues and GitHub Discussions of this repository and believe that this is not a duplicate.

• I have confirmed that this bug belongs to the current repository, not other repositories of RocketMQ.

Programming Language of the Client

C++

Runtime Platform Environment

Red Hat Enterprise Linux Server release 7.9 (Maipo)

RocketMQ Version of the Client/Server

客户端：2024.07.05拉取master分支代码
服务端：rocketmq 5.1.4

Run or Compiler Version

Red Hat Enterprise Linux Server release 7.9 (Maipo)

Describe the Bug

Program terminated with signal 11, Segmentation fault.
#0 0x00007fb2c8fef255 in grpc_transport_perform_stream_op(grpc_transport*, grpc_stream*, grpc_transport_stream_op_batch*) ()
from TestRocketMQModule-runtime/libs/librocketmq.so
Missing separate debuginfos, use: debuginfo-install glibc-2.17-326.0.1.el7_9.x86_64
(gdb) bt
#0 0x00007fb2c8fef255 in grpc_transport_perform_stream_op(grpc_transport*, grpc_stream*, grpc_transport_stream_op_batch*) ()
from TestRocketMQModule-runtime/libs/librocketmq.so
#1 0x00007fb2c9236b02 in connected_channel_start_transport_stream_op_batch(grpc_call_element*, grpc_transport_stream_op_batch*) () from TestRocketMQModule-runtime/libs/librocketmq.so
#2 0x00007fb2c8f0902e in grpc_call_next_op(grpc_call_element*, grpc_transport_stream_op_batch*) ()
from TestRocketMQModule-runtime/libs/librocketmq.so
#3 0x00007fb2c91221c0 in (anonymous namespace)::CallData::CompressStartTransportStreamOpBatch(grpc_call_element*, grpc_transport_stream_op_batch*) () from TestRocketMQModule-runtime/libs/librocketmq.so
#4 0x00007fb2c91221fc in (anonymous namespace)::CompressStartTransportStreamOpBatch(grpc_call_element*, grpc_transport_stream_op_batch*) () from TestRocketMQModule-runtime/libs/librocketmq.so
#5 0x00007fb2c8f0902e in grpc_call_next_op(grpc_call_element*, grpc_transport_stream_op_batch*) ()
from TestRocketMQModule-runtime/libs/librocketmq.so
#6 0x00007fb2c91232bd in grpc_core::(anonymous namespace)::CallData::DecompressStartTransportStreamOpBatch(grpc_call_element*, grpc_transport_stream_op_batch*) () from TestRocketMQModule-runtime/libs/librocketmq.so
#7 0x00007fb2c91232f3 in grpc_core::(anonymous namespace)::DecompressStartTransportStreamOpBatch(grpc_call_element*, grpc_transport_stream_op_batch*) () from TestRocketMQModule-runtime/libs/librocketmq.so
#8 0x00007fb2c8f0902e in grpc_call_next_op(grpc_call_element*, grpc_transport_stream_op_batch*) ()
from TestRocketMQModule-runtime/libs/librocketmq.so
#9 0x00007fb2c91cd0ca in grpc_core::promise_filter_detail::ClientCallData::StartBatch(grpc_transport_stream_op_batch*) ()
from TestRocketMQModule-runtime/libs/librocketmq.so
#10 0x00007fb2c922f8c8 in std::enable_if<std::is_base_of<grpc_core::ChannelFilter, grpc_core::HttpClientFilter>::value, grpc_channel_filter>::type grpc_core::MakePromiseBasedFilter<grpc_core::HttpClientFilter, (grpc_core::FilterEndpoint)0, (unsigned char)1>(char const*)::{lambda(grpc_call_element*, grpc_transport_stream_op_batch*)#1}::operator()(grpc_call_element*, grpc_transport_stream_op_batch*) const () from TestRocketMQModule-runtime/libs/librocketmq.so
#11 0x00007fb2c922f8f0 in std::enable_if<std::is_base_of<grpc_core::ChannelFilter, grpc_core::HttpClientFilter>::value, grpc_channel_filter>::type grpc_core::MakePromiseBasedFilter<grpc_core::HttpClientFilter, (grpc_core::FilterEndpoint)0, (unsigned char)1>(char const*)::{lambda(grpc_call_element*, grpc_transport_stream_op_batch*)#1}::_FUN(grpc_call_element*, grpc_transport_stream_op_batch*) () from TestRocketMQModule-runtime/libs/librocketmq.so
#12 0x00007fb2c8f0902e in grpc_call_next_op(grpc_call_element*, grpc_transport_stream_op_batch*) ()
from TestRocketMQModule-runtime/libs/librocketmq.so
#13 0x00007fb2c9125f16 in message_size_start_transport_stream_op_batch(grpc_call_element*, grpc_transport_stream_op_batch*)
() from TestRocketMQModule-runtime/libs/librocketmq.so
#14 0x00007fb2c8f0902e in grpc_call_next_op(grpc_call_element*, grpc_transport_stream_op_batch*) ()
from TestRocketMQModule-runtime/libs/librocketmq.so
#15 0x00007fb2c91cd0ca in grpc_core::promise_filter_detail::ClientCallData::StartBatch(grpc_transport_stream_op_batch*) ()
from TestRocketMQModule-runtime/libs/librocketmq.so
#16 0x00007fb2c8faabcc in std::enable_if<std::is_base_of<grpc_core::ChannelFilter, grpc_core::ClientAuthFilter>::value, grpc_channel_filter>::type grpc_core::MakePromiseBasedFilter<grpc_core::ClientAuthFilter, (grpc_core::FilterEndpoint)0, (unsigned char)0>(char const*)::{lambda(grpc_call_element*, grpc_transport_stream_op_batch*)#1}::operator()(grpc_call_element*, grpc_trans---Type to continue, or q to quit---
port_stream_op_batch*) const () from TestRocketMQModule-runtime/libs/librocketmq.so
#17 0x00007fb2c8faabf4 in std::enable_if<std::is_base_of<grpc_core::ChannelFilter, grpc_core::ClientAuthFilter>::value, grpc_channel_filter>::type grpc_core::MakePromiseBasedFilter<grpc_core::ClientAuthFilter, (grpc_core::FilterEndpoint)0, (unsigned char)0>(char const*)::{lambda(grpc_call_element*, grpc_transport_stream_op_batch*)#1}::_FUN(grpc_call_element*, grpc_transport_stream_op_batch*) () from TestRocketMQModule-runtime/libs/librocketmq.so
#18 0x00007fb2c8f0902e in grpc_call_next_op(grpc_call_element*, grpc_transport_stream_op_batch*) ()
from TestRocketMQModule-runtime/libs/librocketmq.so
#19 0x00007fb2c91cd0ca in grpc_core::promise_filter_detail::ClientCallData::StartBatch(grpc_transport_stream_op_batch*) ()
from TestRocketMQModule-runtime/libs/librocketmq.so
#20 0x00007fb2c911fd0c in std::enable_if<std::is_base_of<grpc_core::ChannelFilter, grpc_core::ClientAuthorityFilter>::value, grpc_channel_filter>::type grpc_core::MakePromiseBasedFilter<grpc_core::ClientAuthorityFilter, (grpc_core::FilterEndpoint)0, (unsigned char)0>(char const*)::{lambda(grpc_call_element*, grpc_transport_stream_op_batch*)#1}::operator()(grpc_call_element*, grpc_transport_stream_op_batch*) const () from TestRocketMQModule-runtime/libs/librocketmq.so
#21 0x00007fb2c911fd34 in std::enable_if<std::is_base_of<grpc_core::ChannelFilter, grpc_core::ClientAuthorityFilter>::value, grpc_channel_filter>::type grpc_core::MakePromiseBasedFilter<grpc_core::ClientAuthorityFilter, (grpc_core::FilterEndpoint)0, (unsigned char)0>(char const*)::{lambda(grpc_call_element*, grpc_transport_stream_op_batch*)#1}::_FUN(grpc_call_element*, grpc_transport_stream_op_batch*) () from TestRocketMQModule-runtime/libs/librocketmq.so
#22 0x00007fb2c8e9ca06 in grpc_core::SubchannelCall::StartTransportStreamOpBatch(grpc_transport_stream_op_batch*) ()
from TestRocketMQModule-runtime/libs/librocketmq.so
#23 0x00007fb2c8e344bb in grpc_core::ClientChannel::LoadBalancedCall::StartTransportStreamOpBatch(grpc_transport_stream_op_batch*) () from TestRocketMQModule-runtime/libs/librocketmq.so
#24 0x00007fb2c8e2bda3 in grpc_core::(anonymous namespace)::DynamicTerminationFilter::CallData::StartTransportStreamOpBatch(grpc_call_element*, grpc_transport_stream_op_batch*) () from TestRocketMQModule-runtime/libs/librocketmq.so
#25 0x00007fb2c8e6f623 in grpc_core::DynamicFilters::Call::StartTransportStreamOpBatch(grpc_transport_stream_op_batch*) ()
from TestRocketMQModule-runtime/libs/librocketmq.so
#26 0x00007fb2c8e31662 in grpc_core::ClientChannel::CallData::StartTransportStreamOpBatch(grpc_call_element*, grpc_transport_stream_op_batch*) () from TestRocketMQModule-runtime/libs/librocketmq.so
#27 0x00007fb2c8fc1033 in grpc_core::FilterStackCall::ExecuteBatch(grpc_transport_stream_op_batch*, grpc_closure*)::{lambda(void*, grpc_error*)#1}::operator()(void*, grpc_error*) const () from TestRocketMQModule-runtime/libs/librocketmq.so
#28 0x00007fb2c8fc105b in grpc_core::FilterStackCall::ExecuteBatch(grpc_transport_stream_op_batch*, grpc_closure*)::{lambda(void*, grpc_error*)#1}::_FUN(void*, grpc_error*) () from TestRocketMQModule-runtime/libs/librocketmq.so
#29 0x00007fb2c8f32147 in exec_ctx_run(grpc_closure*) () from TestRocketMQModule-runtime/libs/librocketmq.so
#30 0x00007fb2c8f32244 in grpc_core::ExecCtx::Flush() () from TestRocketMQModule-runtime/libs/librocketmq.so
#31 0x00007fb2c8b2b233 in grpc_core::ExecCtx::~ExecCtx() () from TestRocketMQModule-runtime/libs/librocketmq.so
#32 0x00007fb2c8fc5049 in grpc_call_start_batch () from TestRocketMQModule-runtime/libs/librocketmq.so
#33 0x00007fb2c8b38fa3 in grpc::CoreCodegen::grpc_call_start_batch(grpc_call*, grpc_op const*, unsigned long, void*, void*)
() from TestRocketMQModule-runtime/libs/librocketmq.so
#34 0x00007fb2c8bd5bac in grpc::internal::CallOpSet<grpc::internal::CallOpRecvMessageapache::rocketmq::v2::TelemetryCommand, grpc::internal::CallNoOp<2>, grpc::internal::CallNoOp<3>, grpc::internal::CallNoOp<4>, grpc::internal::CallNoOp<5>, grpc::int---Type to continue, or q to quit---
ernal::CallNoOp<6> >::ContinueFillOpsAfterInterception() () from TestRocketMQModule-runtime/libs/librocketmq.so
#35 0x00007fb2c8980d53 in grpc::internal::InterceptorBatchMethodsImpl::ProceedClient() ()
from TestRocketMQModule-runtime/libs/librocketmq.so
#36 0x00007fb2c8980040 in grpc::internal::InterceptorBatchMethodsImpl::Proceed() ()
from TestRocketMQModule-runtime/libs/librocketmq.so
#37 0x00007fb2c8a4e1b8 in rocketmq::InterceptorContinuation::~InterceptorContinuation() ()
from TestRocketMQModule-runtime/libs/librocketmq.so
#38 0x00007fb2c8a4df7f in rocketmq::LogInterceptor::Intercept(grpc::experimental::InterceptorBatchMethods*) ()
from TestRocketMQModule-runtime/libs/librocketmq.so
#39 0x00007fb2c897f6fe in grpc::experimental::ClientRpcInfo::RunInterceptor(grpc::experimental::InterceptorBatchMethods*, unsigned long) () from TestRocketMQModule-runtime/libs/librocketmq.so
#40 0x00007fb2c8980b56 in grpc::internal::InterceptorBatchMethodsImpl::RunClientInterceptors() ()
from TestRocketMQModule-runtime/libs/librocketmq.so
#41 0x00007fb2c8980a63 in grpc::internal::InterceptorBatchMethodsImpl::RunInterceptors() ()
from TestRocketMQModule-runtime/libs/librocketmq.so
#42 0x00007fb2c8bdcc6d in grpc::internal::CallOpSet<grpc::internal::CallOpRecvMessageapache::rocketmq::v2::TelemetryCommand, grpc::internal::CallNoOp<2>, grpc::internal::CallNoOp<3>, grpc::internal::CallNoOp<4>, grpc::internal::CallNoOp<5>, grpc::internal::CallNoOp<6> >::RunInterceptors() () from TestRocketMQModule-runtime/libs/librocketmq.so
#43 0x00007fb2c8bd5976 in grpc::internal::CallOpSet<grpc::internal::CallOpRecvMessageapache::rocketmq::v2::TelemetryCommand, grpc::internal::CallNoOp<2>, grpc::internal::CallNoOp<3>, grpc::internal::CallNoOp<4>, grpc::internal::CallNoOp<5>, grpc::internal::CallNoOp<6> >::FillOps(grpc::internal::Call*) () from TestRocketMQModule-runtime/libs/librocketmq.so
#44 0x00007fb2c8b293be in grpc::Channel::PerformOpsOnCall(grpc::internal::CallOpSetInterface*, grpc::internal::Call*) ()
from TestRocketMQModule-runtime/libs/librocketmq.so
#45 0x00007fb2c897ec03 in grpc::internal::Call::PerformOps(grpc::internal::CallOpSetInterface*) ()
from TestRocketMQModule-runtime/libs/librocketmq.so
#46 0x00007fb2c8bd2b9c in grpc::internal::ClientCallbackReaderWriterImpl<apache::rocketmq::v2::TelemetryCommand, apache::rocketmq::v2::TelemetryCommand>::Read(apache::rocketmq::v2::TelemetryCommand*) ()
from TestRocketMQModule-runtime/libs/librocketmq.so
#47 0x00007fb2c8a68382 in grpc::ClientBidiReactor<apache::rocketmq::v2::TelemetryCommand, apache::rocketmq::v2::TelemetryCommand>::StartRead(apache::rocketmq::v2::TelemetryCommand*) () from TestRocketMQModule-runtime/libs/librocketmq.so
#48 0x00007fb2c8a643c2 in rocketmq::TelemetryBidiReactor::OnReadDone(bool) ()
from TestRocketMQModule-runtime/libs/librocketmq.so
#49 0x00007fb2c8bcdb64 in _ZZN4grpc8internal30ClientCallbackReaderWriterImplIN6apache8rocketmq2v216TelemetryCommandES5_EC4ENS0_4CallEPNS_13ClientContextEPNS_17ClientBidiReactorIS5_S5_EEENKUlbE1_clEb ()
from TestRocketMQModule-runtime/libs/librocketmq.so
#50 0x00007fb2c8bd0bd0 in _ZNSt17_Function_handlerIFvbEZN4grpc8internal30ClientCallbackReaderWriterImplIN6apache8rocketmq2v216TelemetryCommandES7_EC4ENS2_4CallEPNS1_13ClientContextEPNS1_17ClientBidiReactorIS7_S7_EEEUlbE1_E9_M_invokeERKSt9_Any_dataOb
() from TestRocketMQModule-runtime/libs/librocketmq.so
#51 0x00007fb2c8b58ab6 in std::function<void (bool)>::operator()(bool) const ()
---Type to continue, or q to quit---
from TestRocketMQModule-runtime/libs/librocketmq.so
#52 0x00007fb2c8b567c9 in void grpc::internal::CatchingCallback<std::function<void (bool)>&, bool&>(std::function<void (bool)>&, bool&) () from TestRocketMQModule-runtime/libs/librocketmq.so
#53 0x00007fb2c8b5449a in grpc::internal::CallbackWithSuccessTag::Run(bool) ()
from TestRocketMQModule-runtime/libs/librocketmq.so
#54 0x00007fb2c8b543e0 in grpc::internal::CallbackWithSuccessTag::StaticRun(grpc_completion_queue_functor*, int) ()
from TestRocketMQModule-runtime/libs/librocketmq.so
#55 0x00007fb2c8b37457 in grpc::(anonymous namespace)::CallbackAlternativeCQ::Ref()::{lambda(void*)#1}::operator()(void*) const () from TestRocketMQModule-runtime/libs/librocketmq.so
#56 0x00007fb2c8b37485 in grpc::(anonymous namespace)::CallbackAlternativeCQ::Ref()::{lambda(void*)#1}::_FUN(void*) ()
from TestRocketMQModule-runtime/libs/librocketmq.so
#57 0x00007fb2c92b57ed in ZZN9grpc_core12_GLOBAL__N_120ThreadInternalsPosixC4EPKcPFvPvES4_PbRKNS_6Thread7OptionsEENKUlS4_E_clES4 () from TestRocketMQModule-runtime/libs/librocketmq.so
#58 0x00007fb2c92b581e in ZZN9grpc_core12_GLOBAL__N_120ThreadInternalsPosixC4EPKcPFvPvES4_PbRKNS_6Thread7OptionsEENUlS4_E_4_FUNES4 () from TestRocketMQModule-runtime/libs/librocketmq.so
#59 0x00007fb2d530bea5 in start_thread () from TestRocketMQModule-runtime/libs/libpthread.so.0
#60 0x00007fb2d5034b0d in clone () from TestRocketMQModule-runtime/libs/libc.so.6

Steps to Reproduce

5.x版本cpp客户端，运行producer和push_consumer，持续运行

What Did You Expect to See?

1

What Did You See Instead?

1

Additional Context

1

[lizhanhui]

Fixed