lizhimins opened a new pull request, #1337: URL: https://github.com/apache/rocketmq-clients/pull/1337
### Which Issue(s) This PR Fixes Fixes #1335 ### Brief Description The synchronous producer APIs could block their calling thread indefinitely, and `Configuration::withRequestTimeout()` did not bound them. **`SendMessage` had no gRPC deadline.** `send()` was the only RPC in the `ClientManager` interface without a `timeout` parameter: | RPC | `timeout` parameter (before) | | --- | --- | | `ack(...)` | yes | | `changeInvisibleDuration(...)` | yes | | `endTransaction(...)` | yes | | `recallMessage(...)` | yes | | `send(...)` | **no** | So `ClientManagerImpl::send()` never called `set_deadline()`, `RpcClientImpl::asyncSend()` handed the context to the stub without one, and a stalled broker left the completion callback pending forever. This is a plumbing omission rather than a deliberate design choice, which is why the fix simply gives `send()` the same `timeout` parameter every sibling RPC already has. **All four synchronous wait sites used `absl::CondVar::Wait()` with no deadline, and three could also lose the wakeup:** | Site | Completion guard (before) | Deadline (before) | | --- | --- | --- | | `ProducerImpl::send()` sync | correct | none | | `endTransaction0()` | **none at all** | none | | `recallMessage()` | **no flag even existed** | none | | `getPublishInfo()` | **racy — read outside the mutex** | none | `getPublishInfo()` tested `complete` without holding the mutex, so if the callback ran between the check and `Wait()` the signal was dropped and the thread blocked forever with nobody left to wake it — a permanent hang, not a slow call. `endTransaction0()` and `recallMessage()` checked no completion flag at all, so an already-completed callback stranded the caller unconditionally. Changes: - Add a `timeout` parameter to `ClientManager::send()` and set the gRPC deadline in `ClientManagerImpl::send()`. - Pass `requestTimeout()` at the send call site, so `withRequestTimeout()` now bounds normal `SendMessage` RPCs. - Convert all four synchronous waits to `WaitWithDeadline()` guarded by a completion flag that is only read under the mutex, reporting `RequestTimeout` (or a null publish info) when the deadline expires. Note on scope: the issue as originally diagnosed covered `send()` and `getPublishInfo()`. While fixing those I found `endTransaction0()` and `recallMessage()` share the same defect in a worse form, so all four are fixed here rather than leaving two known hangs behind. The interface change is source-compatible for users, since `ClientManager` is internal (`cpp/source/client/include/`) and not part of the public `cpp/include/rocketmq/` surface. The mock and the four affected test call sites are updated accordingly. ### How Did You Test This Change? Added `ProducerImplTest.sendTimesOutWhenRouteStallsTest`, which stalls the route query (the mock never invokes its callback) and asserts that `send()` returns an error within a bounded time, with a hard 10s cap so a regression fails instead of hanging the suite. This test was verified to be a real guard: reverting `getPublishInfo()` to the previous unbounded `Wait()` makes the test process hang permanently — it cannot even be reclaimed by its own 10s cap, because the stuck thread can never be joined. That is exactly the reported symptom. Full C++ suite on this branch, based directly on `master`: ``` $ bazel test //source/... Executed 36 out of 37 tests: 37 tests pass. ``` This branch contains only this fix and sits directly on `master`, so it can be reviewed and merged independently of my consumer-side fix for #1334. -- 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]
