This is an automated email from the ASF dual-hosted git repository.
lollipopjin pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/rocketmq-clients.git
The following commit(s) were added to refs/heads/master by this push:
new da8271b5 [ISSUE #1334] [C++] Fix PushConsumer teardown crash and add
explicit shutdown() (#1336)
da8271b5 is described below
commit da8271b572b41b627bcbbdfb5b3310c0c5e4736b
Author: lizhimins <[email protected]>
AuthorDate: Mon Aug 17 10:39:21 2026 +0800
[ISSUE #1334] [C++] Fix PushConsumer teardown crash and add explicit
shutdown() (#1336)
---
cpp/include/rocketmq/PushConsumer.h | 17 ++++++++++++
cpp/source/base/ThreadPoolImpl.cpp | 18 +++++++++++--
cpp/source/base/tests/ThreadPoolTest.cpp | 31 ++++++++++++++++++++++
cpp/source/rocketmq/ConsumeTask.cpp | 6 +++++
cpp/source/rocketmq/PushConsumer.cpp | 10 +++++++
cpp/source/rocketmq/PushConsumerImpl.cpp | 5 +++-
cpp/source/rocketmq/tests/ConsumeTaskTest.cpp | 27 +++++++++++++++++++
cpp/source/rocketmq/tests/PushConsumerImplTest.cpp | 10 +++++++
8 files changed, 121 insertions(+), 3 deletions(-)
diff --git a/cpp/include/rocketmq/PushConsumer.h
b/cpp/include/rocketmq/PushConsumer.h
index d51c7e8c..803bf536 100644
--- a/cpp/include/rocketmq/PushConsumer.h
+++ b/cpp/include/rocketmq/PushConsumer.h
@@ -41,6 +41,23 @@ public:
void unsubscribe(const std::string& topic) noexcept;
+ /**
+ * Gracefully shut the consumer down.
+ *
+ * Call this from the application (owner) thread before releasing the
+ * PushConsumer. It cancels periodic tasks, then drains and joins the consume
+ * worker threads synchronously on the calling thread. Because the join
blocks
+ * the owner until all in-flight consume tasks finish, a consume worker can
+ * never become the last owner of the underlying implementation, so the final
+ * destruction always happens on the owner thread rather than on a worker
+ * thread concurrently with process/static teardown.
+ *
+ * Idempotent: safe to call multiple times and safe to call again from the
+ * destructor. If never called explicitly, teardown still runs from the
+ * destructor, but then it is not guaranteed to run on the owner thread.
+ */
+ void shutdown() noexcept;
+
private:
friend class PushConsumerBuilder;
diff --git a/cpp/source/base/ThreadPoolImpl.cpp
b/cpp/source/base/ThreadPoolImpl.cpp
index f19ee1d9..4a9477a1 100644
--- a/cpp/source/base/ThreadPoolImpl.cpp
+++ b/cpp/source/base/ThreadPoolImpl.cpp
@@ -79,10 +79,24 @@ void ThreadPoolImpl::shutdown() {
if (state_.compare_exchange_strong(expected, State::STOPPING,
std::memory_order_relaxed)) {
work_guard_->reset();
context_.stop();
+ auto current_id = std::this_thread::get_id();
for (auto& thread : threads_) {
- if (thread.joinable()) {
- thread.join();
+ if (!thread.joinable()) {
+ continue;
}
+ if (thread.get_id() == current_id) {
+ // shutdown() was invoked from within one of our own worker threads,
+ // e.g. the last shared_ptr to the owning PushConsumerImpl was released
+ // on a consume worker, so ~PushConsumerImpl -> shutdown() runs here.
+ // Joining the current thread raises std::system_error(EDEADLK); on the
+ // noexcept teardown path that would std::terminate. Detach self so the
+ // worker unwinds naturally once io_context::run() returns.
Deterministic
+ // teardown must be driven from a non-worker thread via
PushConsumer::shutdown().
+ SPDLOG_WARN("ThreadPool::shutdown() invoked from a worker thread;
detaching self to avoid self-join");
+ thread.detach();
+ continue;
+ }
+ thread.join();
}
state_.store(State::STOPPED, std::memory_order_relaxed);
}
diff --git a/cpp/source/base/tests/ThreadPoolTest.cpp
b/cpp/source/base/tests/ThreadPoolTest.cpp
index 9cbde6cf..3a71bbcf 100644
--- a/cpp/source/base/tests/ThreadPoolTest.cpp
+++ b/cpp/source/base/tests/ThreadPoolTest.cpp
@@ -20,6 +20,7 @@
#include "rocketmq/RocketMQ.h"
#include "gtest/gtest.h"
#include <atomic>
+#include <chrono>
#include <functional>
#include <thread>
@@ -73,4 +74,34 @@ TEST_F(ThreadPoolTest, testBasics) {
}
}
+// Regression: shutting the pool down from within one of its own worker threads
+// must not attempt to join the calling thread (self-join deadlocks and raises
+// std::system_error EDEADLK). This mirrors the PushConsumer teardown race
where
+// ~PushConsumerImpl runs on a consume worker and drives
ThreadPoolImpl::shutdown().
+TEST_F(ThreadPoolTest, shutdownFromWorkerThreadDoesNotThrowTest) {
+ std::atomic<bool> threw{false};
+ std::atomic<bool> finished{false};
+
+ pool_->submit([&]() {
+ try {
+ pool_->shutdown();
+ } catch (...) {
+ threw.store(true);
+ }
+ finished.store(true);
+ });
+
+ auto deadline = std::chrono::steady_clock::now() + std::chrono::seconds(5);
+ while (!finished.load() && std::chrono::steady_clock::now() < deadline) {
+ std::this_thread::sleep_for(std::chrono::milliseconds(10));
+ }
+
+ EXPECT_TRUE(finished.load());
+ EXPECT_FALSE(threw.load());
+
+ // Give the detached worker a moment to unwind out of io_context::run()
+ // before the fixture destroys the pool (test-only synchronization).
+ std::this_thread::sleep_for(std::chrono::milliseconds(200));
+}
+
ROCKETMQ_NAMESPACE_END
diff --git a/cpp/source/rocketmq/ConsumeTask.cpp
b/cpp/source/rocketmq/ConsumeTask.cpp
index 5981d076..8675efea 100644
--- a/cpp/source/rocketmq/ConsumeTask.cpp
+++ b/cpp/source/rocketmq/ConsumeTask.cpp
@@ -136,6 +136,12 @@ void ConsumeTask::process() {
}
std::shared_ptr<PushConsumerImpl> consumer = svc->consumer().lock();
+ if (!consumer) {
+ // The owning PushConsumerImpl has been destructed. Bail out before
touching
+ // consumer stats/config so we never dereference a null consumer.
+ SPDLOG_DEBUG("PushConsumer has been destructed; skip processing");
+ return;
+ }
auto self = shared_from_this();
diff --git a/cpp/source/rocketmq/PushConsumer.cpp
b/cpp/source/rocketmq/PushConsumer.cpp
index 726ee673..ee9e941d 100644
--- a/cpp/source/rocketmq/PushConsumer.cpp
+++ b/cpp/source/rocketmq/PushConsumer.cpp
@@ -41,6 +41,16 @@ void PushConsumer::unsubscribe(const std::string& topic)
noexcept {
}
}
+void PushConsumer::shutdown() noexcept {
+ try {
+ if (impl_) {
+ impl_->shutdown();
+ }
+ } catch (const std::exception& e) {
+ SPDLOG_ERROR("Exception in shutdown: {}", e.what());
+ }
+}
+
PushConsumerBuilder PushConsumer::newBuilder() {
return {};
}
diff --git a/cpp/source/rocketmq/PushConsumerImpl.cpp
b/cpp/source/rocketmq/PushConsumerImpl.cpp
index 124db9f8..ecab6ea4 100644
--- a/cpp/source/rocketmq/PushConsumerImpl.cpp
+++ b/cpp/source/rocketmq/PushConsumerImpl.cpp
@@ -42,7 +42,10 @@ PushConsumerImpl::PushConsumerImpl(absl::string_view
group_name) : ClientImpl(gr
}
PushConsumerImpl::~PushConsumerImpl() {
- SPDLOG_DEBUG("DefaultMQPushConsumerImpl is destructed");
+ // Do NOT log here. The destructor may run during process/static teardown
when
+ // the static spdlog default logger has already been destroyed; logging then
+ // dereferences a dangling logger (observed as EXC_BAD_ACCESS at 0x18 on
macOS).
+ // Deterministic teardown should be driven earlier via
PushConsumer::shutdown().
shutdown();
}
diff --git a/cpp/source/rocketmq/tests/ConsumeTaskTest.cpp
b/cpp/source/rocketmq/tests/ConsumeTaskTest.cpp
index a039b042..6086ce54 100644
--- a/cpp/source/rocketmq/tests/ConsumeTaskTest.cpp
+++ b/cpp/source/rocketmq/tests/ConsumeTaskTest.cpp
@@ -162,6 +162,33 @@ TEST_F(ConsumeTaskTest, processWithEmptyMessagesTest) {
task->process();
}
+// Regression: the service is still alive but the owning PushConsumerImpl has
+// already been destructed, so consumer().lock() yields null. process() must
not
+// dereference the null consumer (metrics/stats access), it must bail out
before
+// invoking the listener.
+TEST_F(ConsumeTaskTest, processWithExpiredConsumerTest) {
+ auto msg = buildMessage("topic", "body");
+ auto task = std::make_shared<ConsumeTask>(weak_service_, weak_pq_, msg);
+
+ bool listener_invoked = false;
+ MessageListener listener = [&listener_invoked](const Message&) {
+ listener_invoked = true;
+ return ConsumeResult::SUCCESS;
+ };
+
+ // Consumer weak_ptr is expired (owning PushConsumerImpl already gone).
+ EXPECT_CALL(*service_,
consumer()).WillRepeatedly(testing::Return(std::weak_ptr<PushConsumerImpl>()));
+ ON_CALL(*service_, listener()).WillByDefault(testing::ReturnRef(listener));
+
+ // With no consumer, process() must not run the listener or ack/nack.
+ EXPECT_CALL(*service_, preHandle(testing::_)).Times(0);
+ EXPECT_CALL(*service_, ack(testing::_, testing::_)).Times(0);
+ EXPECT_CALL(*service_, nack(testing::_, testing::_)).Times(0);
+
+ task->process(); // must not dereference a null consumer
+ EXPECT_FALSE(listener_invoked);
+}
+
// --- process() state-machine: Consume → Ack on SUCCESS ---
TEST_F(ConsumeTaskTest, processConsumeSuccessCallsAckTest) {
diff --git a/cpp/source/rocketmq/tests/PushConsumerImplTest.cpp
b/cpp/source/rocketmq/tests/PushConsumerImplTest.cpp
index 5a3831bc..810ed87d 100644
--- a/cpp/source/rocketmq/tests/PushConsumerImplTest.cpp
+++ b/cpp/source/rocketmq/tests/PushConsumerImplTest.cpp
@@ -334,4 +334,14 @@ TEST(PushConsumerImplTest, maxCachedMessageMemoryTest) {
EXPECT_EQ(MixAll::DEFAULT_CACHED_MESSAGE_MEMORY,
consumer->maxCachedMessageMemory());
}
+// shutdown() must be idempotent and safe to call repeatedly, including the
+// implicit call from the destructor. This underpins the public
+// PushConsumer::shutdown() contract (explicit shutdown followed by
destruction).
+TEST(PushConsumerImplTest, shutdownIsIdempotentTest) {
+ auto consumer = createConsumer();
+ consumer->shutdown();
+ consumer->shutdown();
+ // Destruction here triggers shutdown() once more; must not crash.
+}
+
ROCKETMQ_NAMESPACE_END
