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 ccc0d4fc [ISSUE #1335] [C++] Bound the synchronous producer paths with
the request timeout (#1337)
ccc0d4fc is described below
commit ccc0d4fcc848eed66636757a8a70565fce6d4eca
Author: lizhimins <[email protected]>
AuthorDate: Mon Aug 17 10:38:57 2026 +0800
[ISSUE #1335] [C++] Bound the synchronous producer paths with the request
timeout (#1337)
---
cpp/source/client/ClientManagerImpl.cpp | 5 ++
cpp/source/client/include/ClientManager.h | 1 +
cpp/source/client/include/ClientManagerImpl.h | 1 +
.../client/mocks/include/ClientManagerMock.h | 5 +-
cpp/source/client/tests/ClientManagerTest.cpp | 5 +-
cpp/source/rocketmq/ProducerImpl.cpp | 55 ++++++++++++++++++----
.../rocketmq/tests/FifoProducerPartitionTest.cpp | 7 +--
cpp/source/rocketmq/tests/ProducerImplTest.cpp | 34 ++++++++++++-
8 files changed, 98 insertions(+), 15 deletions(-)
diff --git a/cpp/source/client/ClientManagerImpl.cpp
b/cpp/source/client/ClientManagerImpl.cpp
index 3690711d..8511ce22 100644
--- a/cpp/source/client/ClientManagerImpl.cpp
+++ b/cpp/source/client/ClientManagerImpl.cpp
@@ -288,6 +288,7 @@ void ClientManagerImpl::doHeartbeat() {
bool ClientManagerImpl::send(const std::string& target_host,
const Metadata& metadata,
SendMessageRequest& request,
+ std::chrono::milliseconds timeout,
SendResultCallback cb) {
assert(cb);
SPDLOG_DEBUG("Prepare to send message to {} asynchronously. Request: {}",
target_host, request.ShortDebugString());
@@ -296,6 +297,10 @@ bool ClientManagerImpl::send(const std::string&
target_host,
auto invocation_context = new InvocationContext<SendMessageResponse>();
invocation_context->task_name = fmt::format("Send message to {}",
target_host);
invocation_context->remote_address = target_host;
+ // Bound the SendMessage RPC. Without a deadline a stalled broker leaves the
+ // completion callback pending forever, which in turn blocks the synchronous
+ // send path indefinitely.
+ invocation_context->context.set_deadline(std::chrono::system_clock::now() +
timeout);
for (const auto& entry : metadata) {
invocation_context->context.AddMetadata(entry.first, entry.second);
}
diff --git a/cpp/source/client/include/ClientManager.h
b/cpp/source/client/include/ClientManager.h
index d6e20ace..1258e2d6 100644
--- a/cpp/source/client/include/ClientManager.h
+++ b/cpp/source/client/include/ClientManager.h
@@ -97,6 +97,7 @@ public:
virtual bool send(const std::string& target_host,
const Metadata& metadata,
SendMessageRequest& request,
+ std::chrono::milliseconds timeout,
SendResultCallback cb) = 0;
virtual std::error_code notifyClientTermination(const std::string&
target_host, const Metadata& metadata,
diff --git a/cpp/source/client/include/ClientManagerImpl.h
b/cpp/source/client/include/ClientManagerImpl.h
index 08b5afa8..c3da69b0 100644
--- a/cpp/source/client/include/ClientManagerImpl.h
+++ b/cpp/source/client/include/ClientManagerImpl.h
@@ -83,6 +83,7 @@ public:
bool send(const std::string& target_host,
const Metadata& metadata,
SendMessageRequest& request,
+ std::chrono::milliseconds timeout,
SendResultCallback cb) override LOCKS_EXCLUDED(rpc_clients_mtx_);
/**
diff --git a/cpp/source/client/mocks/include/ClientManagerMock.h
b/cpp/source/client/mocks/include/ClientManagerMock.h
index a144b754..398a3fb5 100644
--- a/cpp/source/client/mocks/include/ClientManagerMock.h
+++ b/cpp/source/client/mocks/include/ClientManagerMock.h
@@ -89,7 +89,10 @@ public:
ReceiveMessageCallback),
(override));
- MOCK_METHOD(bool, send, (const std::string&, const Metadata&,
SendMessageRequest&, SendResultCallback), (override));
+ MOCK_METHOD(bool, send,
+ (const std::string&, const Metadata&, SendMessageRequest&,
std::chrono::milliseconds,
+ SendResultCallback),
+ (override));
MOCK_METHOD(std::error_code, notifyClientTermination,
(const std::string&, const Metadata&, const
NotifyClientTerminationRequest&, std::chrono::milliseconds),
diff --git a/cpp/source/client/tests/ClientManagerTest.cpp
b/cpp/source/client/tests/ClientManagerTest.cpp
index 7fab9e8d..006b6cd7 100644
--- a/cpp/source/client/tests/ClientManagerTest.cpp
+++ b/cpp/source/client/tests/ClientManagerTest.cpp
@@ -16,6 +16,7 @@
*/
#include <apache/rocketmq/v2/definition.pb.h>
+#include <chrono>
#include <memory>
#include <system_error>
@@ -254,7 +255,7 @@ TEST_F(ClientManagerTest, sendSuccessTest) {
cv.SignalAll();
};
- client_manager_->send(target_host_, metadata_, request, callback);
+ client_manager_->send(target_host_, metadata_, request,
std::chrono::seconds(3), callback);
{
absl::MutexLock lk(&mtx);
@@ -297,7 +298,7 @@ TEST_F(ClientManagerTest, sendReturnsErrorOnBadRequestTest)
{
cv.SignalAll();
};
- client_manager_->send(target_host_, metadata_, request, callback);
+ client_manager_->send(target_host_, metadata_, request,
std::chrono::seconds(3), callback);
{
absl::MutexLock lk(&mtx);
diff --git a/cpp/source/rocketmq/ProducerImpl.cpp
b/cpp/source/rocketmq/ProducerImpl.cpp
index 22027ab8..47915b41 100644
--- a/cpp/source/rocketmq/ProducerImpl.cpp
+++ b/cpp/source/rocketmq/ProducerImpl.cpp
@@ -227,6 +227,7 @@ SendReceipt ProducerImpl::send(MessageConstPtr message,
std::error_code& ec) noe
auto cv = std::make_shared<absl::CondVar>();
bool completed = false;
SendReceipt send_receipt;
+ const std::string topic = message->topic();
// Define callback
auto callback =
@@ -248,8 +249,15 @@ SendReceipt ProducerImpl::send(MessageConstPtr message,
std::error_code& ec) noe
{
absl::MutexLock lk(mtx.get());
- if (!completed) {
- cv->Wait(mtx.get());
+ // Bound the synchronous wait. The callback may never run if the underlying
+ // RPC stalls, so an unbounded Wait() would hang the calling thread
forever.
+ auto deadline = absl::Now() + requestTimeout();
+ while (!completed) {
+ if (cv->WaitWithDeadline(mtx.get(), deadline)) {
+ SPDLOG_WARN("Timeout waiting for send result of topic[{}]", topic);
+ ec = ErrorCode::RequestTimeout;
+ break;
+ }
}
}
@@ -361,7 +369,7 @@ void ProducerImpl::sendImpl(std::shared_ptr<SendContext>
context) {
context->onSuccess(send_result);
};
- client_manager_->send(target, metadata, request, callback);
+ client_manager_->send(target, metadata, request,
absl::ToChronoMilliseconds(requestTimeout()), callback);
}
void ProducerImpl::send0(MessageConstPtr message, const SendCallback&
callback, std::vector<rmq::MessageQueue> list) {
@@ -462,7 +470,17 @@ bool ProducerImpl::endTransaction0(const MiniTransaction&
transaction, Transacti
{
absl::MutexLock lk(mtx.get());
- cv->Wait(mtx.get());
+ // Guard on `completed`: the callback may already have run (and signalled
with
+ // no waiter present) before we get here, in which case an unguarded Wait()
+ // would block forever. The deadline bounds a stalled RPC.
+ auto deadline = absl::Now() + requestTimeout();
+ while (!completed) {
+ if (cv->WaitWithDeadline(mtx.get(), deadline)) {
+ SPDLOG_WARN("Timeout waiting for {} transaction result of topic[{}]",
action, topic);
+ success = false;
+ break;
+ }
+ }
}
return success;
}
@@ -559,6 +577,7 @@ RecallReceipt ProducerImpl::recall(const std::string&
topic, std::string& recall
auto cv = std::make_shared<absl::CondVar>();
RecallReceipt recall_receipt;
+ bool completed = false;
auto callback =
[&, mtx, cv, topic](const std::error_code& code, const
RecallMessageResponse& response) {
@@ -569,6 +588,7 @@ RecallReceipt ProducerImpl::recall(const std::string&
topic, std::string& recall
{
absl::MutexLock lk(mtx.get());
+ completed = true;
cv->SignalAll();
}
};
@@ -578,7 +598,16 @@ RecallReceipt ProducerImpl::recall(const std::string&
topic, std::string& recall
{
absl::MutexLock lk(mtx.get());
- cv->Wait(mtx.get());
+ // Guard on `completed` so an already-completed callback cannot strand this
+ // thread in an unbounded Wait(), and bound the wait with a deadline.
+ auto deadline = absl::Now() + requestTimeout();
+ while (!completed) {
+ if (cv->WaitWithDeadline(mtx.get(), deadline)) {
+ SPDLOG_WARN("Timeout waiting for recall result of topic[{}]", topic);
+ ec = ErrorCode::RequestTimeout;
+ break;
+ }
+ }
}
return recall_receipt;
@@ -637,10 +666,20 @@ TopicPublishInfoPtr ProducerImpl::getPublishInfo(const
std::string& topic) {
};
getPublishInfoAsync(topic, cb);
- // Wait till acquiring topic publish info completes
- while (!complete) {
+ // Wait till acquiring topic publish info completes.
+ // `complete` must only be read while holding mtx: testing it outside the
lock
+ // races with the callback and can miss the signal entirely (the callback may
+ // complete between the check and Wait()), which used to hang this thread
+ // forever. The deadline additionally bounds a stalled route query.
+ {
absl::MutexLock lk(mtx.get());
- cv->Wait(mtx.get());
+ auto deadline = absl::Now() + requestTimeout();
+ while (!complete) {
+ if (cv->WaitWithDeadline(mtx.get(), deadline)) {
+ SPDLOG_WARN("Timeout acquiring publish info of topic[{}]", topic);
+ return nullptr;
+ }
+ }
}
// TODO: propagate error_code to caller
diff --git a/cpp/source/rocketmq/tests/FifoProducerPartitionTest.cpp
b/cpp/source/rocketmq/tests/FifoProducerPartitionTest.cpp
index ec324ad3..98fe3d64 100644
--- a/cpp/source/rocketmq/tests/FifoProducerPartitionTest.cpp
+++ b/cpp/source/rocketmq/tests/FifoProducerPartitionTest.cpp
@@ -15,6 +15,7 @@
* limitations under the License.
*/
#include <atomic>
+#include <chrono>
#include <memory>
#include <string>
#include <system_error>
@@ -139,7 +140,7 @@ protected:
void installSendSuccessHandler() {
ON_CALL(*client_manager_, send)
.WillByDefault(testing::Invoke(
- [this](const std::string&, const Metadata&, SendMessageRequest&,
SendResultCallback cb) {
+ [this](const std::string&, const Metadata&, SendMessageRequest&,
std::chrono::milliseconds, SendResultCallback cb) {
{
absl::MutexLock lk(&mtx_);
send_count_++;
@@ -224,7 +225,7 @@ TEST_F(FifoProducerPartitionTest,
multipleMessagesPreserveOrderTest) {
ON_CALL(*client_manager_, send)
.WillByDefault(testing::Invoke(
[this, &sent_bodies](const std::string&, const Metadata&,
SendMessageRequest& request,
- SendResultCallback cb) {
+ std::chrono::milliseconds, SendResultCallback
cb) {
{
absl::MutexLock lk(&mtx_);
if (request.messages_size() > 0) {
@@ -307,7 +308,7 @@ TEST_F(FifoProducerPartitionTest,
failedMessageRetriedViaOnCompleteTest) {
std::atomic<int> attempt{0};
ON_CALL(*client_manager_, send)
.WillByDefault(testing::Invoke(
- [this, &attempt](const std::string&, const Metadata&,
SendMessageRequest&, SendResultCallback cb) {
+ [this, &attempt](const std::string&, const Metadata&,
SendMessageRequest&, std::chrono::milliseconds, SendResultCallback cb) {
int current = attempt.fetch_add(1);
{
absl::MutexLock lk(&mtx_);
diff --git a/cpp/source/rocketmq/tests/ProducerImplTest.cpp
b/cpp/source/rocketmq/tests/ProducerImplTest.cpp
index 011eb3ce..aeedbb98 100644
--- a/cpp/source/rocketmq/tests/ProducerImplTest.cpp
+++ b/cpp/source/rocketmq/tests/ProducerImplTest.cpp
@@ -15,6 +15,8 @@
* limitations under the License.
*/
#include <atomic>
+#include <chrono>
+#include <future>
#include <memory>
#include <string>
#include <system_error>
@@ -135,7 +137,8 @@ protected:
void installSendSuccessHandler() {
ON_CALL(*client_manager_, send)
.WillByDefault(testing::Invoke(
- [this](const std::string&, const Metadata&, SendMessageRequest&,
SendResultCallback cb) {
+ [this](const std::string&, const Metadata&, SendMessageRequest&,
std::chrono::milliseconds,
+ SendResultCallback cb) {
{
absl::MutexLock lk(&mtx_);
send_count_++;
@@ -281,4 +284,33 @@ TEST_F(ProducerImplTest,
topicsOfInterestReturnsConfiguredTopicsTest) {
EXPECT_EQ("topic-c", result_topics[2]);
}
+// Regression: when the route query never completes (stalled name server), the
+// synchronous send path must not block forever. getPublishInfo() has to honour
+// the request timeout and return, so send() reports an error within a bounded
+// time instead of hanging the calling thread.
+TEST_F(ProducerImplTest, sendTimesOutWhenRouteStallsTest) {
+ // Route query that never invokes its callback (simulates a stalled broker).
+ ON_CALL(*client_manager_, resolveRoute)
+ .WillByDefault(testing::Invoke(
+ [](const std::string&, const Metadata&, const QueryRouteRequest&,
std::chrono::milliseconds,
+ const std::function<void(const std::error_code&, const
TopicRouteDataPtr&)>&) {
+ // Intentionally drop the callback.
+ }));
+
+ producer_->withRequestTimeout(std::chrono::milliseconds(300));
+
+ auto fut = std::async(std::launch::async, [this]() {
+ auto msg = makeMessage("stall-topic");
+ std::error_code ec;
+ producer_->send(std::move(msg), ec);
+ return ec;
+ });
+
+ // Hard cap well above the request timeout: if send() still hangs, fail the
+ // test rather than block the whole suite forever.
+ ASSERT_EQ(std::future_status::ready, fut.wait_for(std::chrono::seconds(10)))
+ << "send() did not return after route query stalled";
+ EXPECT_TRUE(static_cast<bool>(fut.get())) << "expected an error when route
query stalls";
+}
+
ROCKETMQ_NAMESPACE_END
