RockteMQ-AI commented on issue #10747:
URL: https://github.com/apache/rocketmq/issues/10747#issuecomment-5155546516
**Automated Fix Proposal (v1)**
Code base: `a06836dd564e` (develop)
## Summary
**Root Cause:** `invokeAsyncImpl()` chains `whenComplete → thenAccept →
exceptionally`, firing `operationComplete` before
`operationSucceed`/`operationFail` — violating the `InvokeCallback` contract.
**Fix Strategy:** Replace the three-stage pipeline with a single terminal
`whenComplete`:
- Success: `operationSucceed` → `operationComplete`
- Failure: `operationFail` → `operationComplete`
- Callback exceptions isolated from invocation outcome routing
**Files to Modify:** 1 file, 1 method (~17 lines)
-
`remoting/src/main/java/org/apache/rocketmq/remoting/netty/NettyRemotingAbstract.java`
**Test Plan:** 7 deterministic unit tests (no network/timers)
- Callback ordering on success/failure paths
- Exception isolation (callback throws do not cross-contaminate)
- #9119/#9120 raw-exception unwrap preservation
**Backward Compatibility:** No breaking changes — behavior corrected to
match documented contract.
**Risk:** Low — localized change, clear reference implementation
(`ResponseFuture.executeInvokeCallback`).
<details>
<summary>Full Specification</summary>
# Fix Specification — Issue #10747
## `InvokeCallback` completion-order violation in
`NettyRemotingAbstract.invokeAsyncImpl`
---
## 1. Root Cause Analysis
| Field | Value |
|---|---|
| **File** |
`remoting/src/main/java/org/apache/rocketmq/remoting/netty/NettyRemotingAbstract.java`
|
| **Method** | `invokeAsyncImpl(Channel, RemotingCommand, long,
InvokeCallback)` |
| **Lines** | 677–693 |
### Current code (broken)
```java
invokeImpl(channel, request, timeoutMillis)
.whenComplete((v, t) -> { // ① operationComplete
fires FIRST
if (t == null) {
invokeCallback.operationComplete(v);
} else {
ResponseFuture responseFuture = new ResponseFuture(...);
responseFuture.setCause(t);
invokeCallback.operationComplete(responseFuture);
}
})
.thenAccept(responseFuture -> // ② operationSucceed fires
SECOND
invokeCallback.operationSucceed(responseFuture.getResponseCommand()))
.exceptionally(t -> { // ③ operationFail fires
THIRD
invokeCallback.operationFail(ExceptionUtils.getRealException(t));
return null;
});
```
### Contract (from `InvokeCallback` Javadoc, introduced in #7321/#7322)
> `operationComplete` is expected to be invoked **after**
`operationSucceed(RemotingCommand)` or `operationFail(Throwable)`.
The correct reference implementation is
`ResponseFuture.executeInvokeCallback()` (lines 62–80), which calls
`operationSucceed`/`operationFail` first and `operationComplete` last.
### Two distinct bugs
| # | Bug | Mechanism |
|---|---|---|
| **B1** | **Ordering reversal** — `operationComplete` fires before
`operationSucceed` or `operationFail`. | `whenComplete` is the first stage in
the chain; its action runs before the dependent `thenAccept`/`exceptionally`
stages. |
| **B2** | **Callback-exception cross-contamination** — if
`operationComplete` (or `operationSucceed`) throws, the dependent stage becomes
exceptional, `operationSucceed` is skipped on success, and the user-thrown
exception is routed to `operationFail` as though the remoting invocation itself
had failed. | `thenAccept` and `exceptionally` are *dependent* stages — an
exception in a predecessor propagates downstream and suppresses the normal
path. |
---
## 2. Fix Strategy
Replace the three-stage pipeline with a **single terminal `whenComplete`**
that inspects the outcome of `invokeImpl` and dispatches callbacks in the
correct order, isolated from each other.
### Pseudocode
```java
invokeImpl(channel, request, timeoutMillis)
.whenComplete((responseFuture, throwable) -> {
if (throwable == null) {
// Success path
invokeCallback.operationSucceed(responseFuture.getResponseCommand());
invokeCallback.operationComplete(responseFuture);
} else {
// Failure path — mirror ResponseFuture.executeInvokeCallback
ResponseFuture failedFuture = new ResponseFuture(
channel, request.getOpaque(), request, timeoutMillis, null,
null);
failedFuture.setCause(throwable);
invokeCallback.operationFail(ExceptionUtils.getRealException(throwable));
invokeCallback.operationComplete(failedFuture);
}
});
```
### Step-by-step changes
1. **Delete** the existing `.whenComplete(...)`, `.thenAccept(...)`, and
`.exceptionally(...)` calls on lines 680–693.
2. **Add** a single `.whenComplete((responseFuture, throwable) -> { ... })`
block.
3. **Success branch** (`throwable == null`):
- Call
`invokeCallback.operationSucceed(responseFuture.getResponseCommand())`.
- Then call `invokeCallback.operationComplete(responseFuture)`.
4. **Failure branch** (`throwable != null`):
- Build the synthetic `ResponseFuture` exactly as the current code does
(preserving the behavior introduced by #9119/#9120 — `setCause(t)` on the new
`ResponseFuture`).
- Call
`invokeCallback.operationFail(ExceptionUtils.getRealException(throwable))`.
- Then call `invokeCallback.operationComplete(failedFuture)`.
5. **Do not** wrap individual callback invocations in try/catch at this
layer. `ResponseFuture.executeInvokeCallback` does not either — exception
propagation to the caller of `invokeAsyncImpl` is the existing contract. If
hardening is desired, it should be done in a separate, explicitly-scoped change.
---
## 3. Files to Modify
| File | Change |
|---|---|
|
`remoting/src/main/java/org/apache/rocketmq/remoting/netty/NettyRemotingAbstract.java`
| Rewrite `invokeAsyncImpl` (lines 677–693) as described above. |
No other source files require changes. `InvokeCallback.java`,
`ResponseFuture.java`, and all callers of `invokeAsyncImpl` remain untouched.
---
## 4. Test Plan
All tests use already-completed `CompletableFuture<ResponseFuture>`
instances — no network, timers, or scheduling.
### 4.1 New test class
`remoting/src/test/java/org/apache/rocketmq/remoting/netty/NettyRemotingAbstractInvokeAsyncImplTest.java`
### 4.2 Test cases
| # | Test name | What it verifies |
|---|---|---|
| T1 | `successCallbackOrder` | On a normally-completed future,
`operationSucceed` is invoked **before** `operationComplete` (Mockito
`InOrder`). `operationFail` is never invoked. |
| T2 | `failureCallbackOrder` | On an exceptionally-completed future
(sentinel `RemotingException`), `operationFail` is invoked **before**
`operationComplete`. `operationSucceed` is never invoked. |
| T3 | `operationCompleteThrowsDoesNotSuppressSucceed` | `operationComplete`
throws a sentinel `RuntimeException`. Verify `operationSucceed` was still
invoked and `operationFail` was **not** invoked. |
| T4 | `operationCompleteThrowsDoesNotTriggerFail` | On the failure path,
`operationComplete` throws. Verify `operationFail` was still invoked exactly
once (with the original invocation exception, not the callback exception). |
| T5 | `operationSucceedThrowsDoesNotTriggerFail` | `operationSucceed`
throws a sentinel exception. Verify `operationFail` is **not** invoked. (This
test documents current behavior after the fix — callback exceptions propagate
but do not cross-trigger other callbacks.) |
| T6 | `realExceptionUnwrapPreserved` | Complete the future with a
`CompletionException` wrapping a `RemotingTimeoutException`. Verify
`operationFail` receives the unwrapped `RemotingTimeoutException` (preserves
#9119/#9120 behavior via `ExceptionUtils.getRealException`). |
| T7 | `syntheticResponseFutureOnFailure` | On failure, verify that
`operationComplete` receives a `ResponseFuture` whose `getCause()` matches the
original throwable. |
### 4.3 Test infrastructure
- Subclass `NettyRemotingAbstract` with a minimal concrete stub (the class
is abstract) that overrides `invokeImpl` to return a controllable
`CompletableFuture<ResponseFuture>`.
- Use Mockito `mock(InvokeCallback.class)` and `InOrder` for ordering
verification.
- No existing tests need modification — the current test suite does not
assert callback ordering.
---
## 5. Backward Compatibility
| Aspect | Assessment |
|---|---|
| **Public API** | No change. `invokeAsyncImpl` signature and
`InvokeCallback` interface are unchanged. |
| **Callback ordering** | This is a **bug fix** that brings behavior in line
with the documented contract. Any code that accidentally depended on the
reversed order (e.g., reading state set in `operationComplete` from within
`operationSucceed`) was already incorrect. |
| **Exception propagation** | Callback exceptions will no longer be silently
routed to `operationFail`. Callers that relied on `operationFail` firing for
*all* errors (including callback bugs) will see a behavioral change — but the
previous behavior was itself a bug that masked real invocation outcomes. |
| **#9119/#9120 raw-exception unwrap** | Preserved.
`ExceptionUtils.getRealException` is still called before `operationFail`. |
| **`ResponseFuture.executeInvokeCallback`** | Unchanged. It already had
correct ordering and serves as the reference implementation. |
| **Binary compatibility** | No change to method signatures, class
hierarchy, or public fields. |
**Verdict: No breaking changes.** The fix corrects behavior to match the
documented and expected contract.
---
## 6. Risk Assessment
| Risk | Likelihood | Impact | Mitigation |
|---|---|---|---|
| **Callback exception propagation** — After the fix, if `operationSucceed`
throws, the exception propagates out of `whenComplete` rather than being caught
by a downstream `exceptionally`. | Medium | Low–Medium | This matches
`ResponseFuture.executeInvokeCallback` behavior. Callers that wrap
`invokeAsyncImpl` should already handle exceptions from their own callbacks. If
needed, a follow-up change can add per-callback try/catch with logging. |
| **Subtle behavioral change for existing callers** — Code that worked "by
accident" with the reversed order might break. | Low | Low | The reversed order
was a contract violation. Any code depending on it was fragile. The test suite
should catch regressions. |
| **Incomplete `ResponseFuture` on failure path** — The synthetic
`ResponseFuture` built on failure has `null` for `invokeCallback` and
`rpcHook`. | None (no change) | None | This is identical to the current
behavior — the synthetic future is only passed to `operationComplete`, which
uses it for context (channel, request, cause), not for re-invoking callbacks. |
| **Thread-safety** — `whenComplete` runs on the completing thread (same as
before). | Low | None | No new concurrency concerns. The existing code already
ran all three callbacks from the same thread context. |
| **Missed callers of `invokeAsyncImpl`** | Low | Low | Grep confirms
`invokeAsyncImpl` is called from `invokeAsync` in the same class and from
`NettyRemotingClient`/`NettyRemotingServer`. All go through the same code path
— the fix applies uniformly. |
### Overall risk: **Low**
The change is localized to a single 17-line method, has a clear reference
implementation (`ResponseFuture.executeInvokeCallback`), and is fully testable
with deterministic unit tests.
</details>
**Next Steps:**
- Reply `/approve` to proceed with PR generation
- Reply `/revise <feedback>` to request changes
- Reply `/reject` to close this proposal
*This proposal expires in 72 hours.*
---
*Automated by RockteMQ-AI*
--
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]
