lizhimins opened a new issue, #1334: URL: https://github.com/apache/rocketmq-clients/issues/1334
### Before Creating the Bug Report - [X] I found a bug, not just asking a question, which should be created in [GitHub Discussions](https://github.com/apache/rocketmq-clients/discussions). - [X] I have searched the [GitHub Issues](https://github.com/apache/rocketmq-clients/issues) of this repository and believe that this is not a duplicate. - [X] I have confirmed that this bug belongs to the current repository, not other repositories of RocketMQ. ### Runtime platform environment macOS arm64, Bazel opt build ### RocketMQ description Apache rocketmq-clients C++ (verified against master `3a8ec615`) ### RocketMQ Client version cpp-5.1.0 and current master ### Run or compilation description Bazel 6.6.0, opt build ### Bug description The public `PushConsumer` API has no `shutdown()` / `close()`; teardown relies entirely on releasing the last `shared_ptr`. When an application releases its last reference while a message is still being consumed, the process crashes with `EXC_BAD_ACCESS` (`KERN_INVALID_ADDRESS` at `0x18`) on the SDK consume worker thread. Observed stack: ``` spdlog::logger::log(...) rocketmq::PushConsumerImpl::~PushConsumerImpl() rocketmq::ConsumeTask::process() rocketmq::ThreadPoolImpl worker ``` **Root cause.** `ConsumeTask::process()` promotes the service's `weak_ptr` to an *owning* `shared_ptr` on the consume worker thread: ```cpp // cpp/source/rocketmq/ConsumeTask.cpp:138 std::shared_ptr<PushConsumerImpl> consumer = svc->consumer().lock(); ``` `ConsumeMessageServiceImpl` holds the consumer only weakly (`ConsumeMessageService.h:65`), and `ClientManagerImpl` keeps clients as `std::vector<std::weak_ptr<Client>>`, so the application's reference really is the only strong one. Once it is dropped mid-consumption, the local `consumer` above becomes the last owner and `~PushConsumerImpl()` runs on the worker thread when `process()` returns. Two independent failures follow from that: 1. `~PushConsumerImpl()` logs before doing anything else, so it may touch the static spdlog default logger after it has been destroyed during process teardown — this is the observed `0x18` fault: ```cpp // cpp/source/rocketmq/PushConsumerImpl.cpp:45 PushConsumerImpl::~PushConsumerImpl() { SPDLOG_DEBUG("DefaultMQPushConsumerImpl is destructed"); shutdown(); } ``` 2. If that log survives, `shutdown()` reaches `ThreadPoolImpl::shutdown()`, which joins **every** thread unconditionally, including the calling worker. Self-join raises `std::system_error(EDEADLK)`; because `PushConsumerImpl::shutdown()` is `noexcept`, that terminates the process. Reproduced in a unit test as `terminate called without an active exception`. Separately, `ConsumeTask::process()` dereferences `consumer` without a null check, so an already-expired consumer is a third crash path. The crash is **not FIFO-specific**: `ConsumeTask::process()`, `~PushConsumerImpl()` and `ThreadPoolImpl::shutdown()` are shared by all message types; `fifo_` only selects a `NextStep` branch. FIFO merely widens the window because consumption is serialized. **Why the official example does not hit this.** `ExamplePushConsumer` sleeps 30 minutes and returns, so destruction happens (a) on the main thread, (b) while no consume task is in flight, and (c) before static destructors run. All three hold only by accident and none is enforced or documented by the API. For comparison, the other SDKs do expose explicit shutdown — Java's `PushConsumer extends Closeable` with `close()`, Go's `GracefulStop()`. C++ is the only one missing it, which is why this keeps resurfacing. Prior reports were auto-closed as stale rather than fixed: #421, #521. ### Steps to reproduce 1. Start a `PushConsumer` with a listener returning `ConsumeResult::SUCCESS`. 2. From the application thread, destroy the `PushConsumer` right after the listener signals completion (i.e. while `ConsumeTask::process()` is still on the worker stack). 3. The process faults on the consume worker thread. ### What did you expect to see? A supported way to shut the consumer down deterministically from the owner thread, draining in-flight consume tasks and ACK callbacks, with the final destruction happening on the owner thread. ### What did you see instead? `SIGSEGV` (exit 139) on the SDK consume worker thread after the message was successfully consumed and acked. ### Additional context I have a fix and will open a PR referencing this issue. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
