[Bug] Send-failure cause is discarded: requestFail() never records f.cause()
- Dominant language
- Java
- Stars
- 22.6k
- Forks
- 12k
- Avg merge
- 3d 1h
- Merged PRs (30d)
- 27
Description
### Before Creating the Bug Report
- [x] I found a bug, not just asking a question, which should be created in GitHub Discussions.
- [x] I have searched the GitHub Issues and GitHub Discussions 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
OS: Linux 5.14.0 (x86_64), containerized
Client embedded in an Apache Flink 2.2.1 TaskManager (producer only)
Broker: 5.1.4, single node
### RocketMQ version
branch: develop
version: reproduced on 5.3.1; the same code is present unchanged in 5.5.1 and on develop
Git commit id: 88846a0a5b550116d183f62c1a468e75ca3f61d4
### JDK Version
Compiler: OpenJDK 17.0.x
Runtime: OpenJDK 21.0.x (Flink TaskManager)
OS: Linux 5.14.0
### Describe the Bug
When `writeAndFlush` fails, `NettyRemotingAbstract` drops the netty-side `f.cause()` in two
independent places, so the reason a send failed is unrecoverable both programmatically and from the
logs.
`remoting/src/main/java/org/apache/rocketmq/remoting/netty/NettyRemotingAbstract.java` (develop):
1. `invoke0`, the `writeAndFlush` listener — `f.cause()` is available and used nowhere:
```java
channel.writeAndFlush(request).addListener((ChannelFutureListener) f -> {
if (f.isSuccess()) {
responseFuture.setSendRequestOK(true);
return;
}
requestFail(opaque); // cause not passed
log.warn("send a request command to channel <{}>, channelId={}, failed.",
RemotingHelper.parseChannelRemoteAddr(channel), channel.id()); // throwable not passed
});
```
2. `requestFail(final int opaque)` — never calls `setCause`:
```java
private void requestFail(final int opaque) {
ResponseFuture responseFuture = responseTable.remove(opaque);
if (responseFuture != null) {
responseFuture.setSendRequestOK(false);
responseFuture.putResponse(null); // no setCause(...) anywhere
...
}
}
```
The consumer side is already written to carry a cause — `ResponseFuture#executeInvokeCallback`
builds the exception *from* `getCause()`:
```java
if (!isSendRequestOK()) {
invokeCallback.operationFail(
new RemotingSendRequestException(channel.remoteAddress().toString(), getCause()));
}
```
Because `requestFail` never populates the field, `getCause()` is always `null` here. The plumbing
exists; only the assignment is missing. Both the synchronous path (`invokeSyncImpl`, which rethrows
`e.getCause()` from the `CompletableFuture`) and the async path (`invokeAsyncImpl`) inherit the
null cause.
Two further sites drop a throwable that is in hand:
3. `invokeOnewayImpl` listener: `log.warn("send a request command to channel <" + channel.remoteAddress() + "> failed.")` — no `f.cause()`.
4. `invokeOnewayImpl` catch block: `log.warn("write send a request command to channel <" + channel.remoteAddress() + "> failed.")` — `e` is caught and wrapped into the thrown exception, but never logged.
`failFast(Channel)` also routes through `requestFail`. It is invoked from
`NettyRemotingClient`'s `close()` handler, so "the channel was closed" is a known, accurate cause
that could be reported instead of nothing.
### Steps to Reproduce
Unit-level reproduction, no broker required. Make `writeAndFlush` return an already-failed promise
and assert on what the caller receives:
```java
@Test
public void testWriteFailurePropagatesNettyCauseToCaller() throws InterruptedException {
final Throwable writeFailure =
new OutOfMemoryError("Cannot reserve 16777216 bytes of direct buffer memory");
Channel channel = new MockChannel() {
@Override
public ChannelFuture writeAndFlush(Object msg) {
DefaultChannelPromise promise =
new DefaultChannelPromise(this, ImmediateEventExecutor.INSTANCE);
promise.setFailure(writeFailure);
return promise;
}
@Override
public LocalAddress remoteAddress() {
return new LocalAddress("write-failure-test");
}
};
final Semaphore semaphore = new Semaphore(0);
final AtomicReference observed = new AtomicReference<>();
remotingAbstract.invokeAsyncImpl(channel, RemotingCommand.createRequestCommand(1, null), 3000,
new InvokeCallback() {
@Override public void operationComplete(ResponseFuture responseFuture) { }
@Override public void operationSucceed(RemotingCommand response) { }
@Override public void operationFail(Throwable throwable) {
observed.set(throwable);
semaphore.release();
}
});
assertThat(semaphore.tryAcquire(1, 10, TimeUnit.SECONDS)).isTrue();
// walk getCause() transitively
assertThat(causeChainOf(observed.get())).contains(writeFailure);
}
```
On current `develop` this fails: `writeFailure` appears nowhere in the cause chain.
Production reproduction: run two Flink jobs, each with its own producer, in one TaskManager JVM
whose `-XX:MaxDirectMemorySize` leaves less than one netty arena chunk (16 MiB) free. The second
job's producer fails every send.
### What Did You Expect to See?
The netty write failure reaches the caller and the log:
- `requestFail` records `f.cause()` on the `ResponseFuture`, so
`RemotingSendRequestException.getCause()` is the real failure.
- The `log.warn` calls in `invoke0` and `invokeOnewayImpl` pass the throwable.
- `failFast` reports channel closure as the cause rather than nothing.
### What Did You See Instead?
`RemotingSendRequestException` with `getCause() == null`, and a one-line
`send a request command to channel <...> failed.` with no stack trace.
In our case the discarded cause was:
```
java.lang.OutOfMemoryError: Cannot reserve 16777216 bytes of direct buffer memory
(allocated: 595741295, limit: 606664136)
at java.base/java.nio.Bits.reserveMemory(Bits.java:178)
at java.base/java.nio.DirectByteBuffer.(DirectByteBuffer.java:111)
at io.netty.buffer.PoolArena$DirectArena.allocateDirect(PoolArena.java:632)
at io.netty.buffer.PooledByteBufAllocator.newDirectBuffer(PooledByteBufAllocator.java:395)
...
```
Direct memory was 98.2% exhausted, so netty could not allocate a 16 MiB arena chunk. Every send
failed, each failure closed the channel, and the client entered a reconnect storm (14045 CONNECT /
13869 ACTIVE / 28125 CLOSE events). Because the cause was discarded at both the exception and the
log, the only observable symptom was "sends fail and the client reconnects forever" — pointing at
the network or the broker, not at memory. We reached the real cause only after independently
obtaining the netty `exceptionCaught` log, which prints `f.cause()` itself.
The fix is small and self-contained: thread the cause through `requestFail`, and pass the throwable
to the four `log.warn` calls. `ResponseFuture` already stores and consumes it.
### Additional Context
Happy to open a PR. Our patch against 5.3.1 (identical shape applies to develop):
- `requestFail(final int opaque)` -> `requestFail(final int opaque, final Throwable cause)`, calling
`responseFuture.setCause(cause)` before `putResponse(null)` — order matters, `putResponse`
releases the latch the synchronous caller is blocked on.
- `invoke0` listener: `requestFail(opaque, f.cause())` and `log.warn(..., f.cause())`.
- `failFast`: `requestFail(opaque, new ClosedChannelException())`, accurate because the only caller
is `NettyRemotingClient`'s `close()` handler.
- `invokeOnewayImpl`: pass `f.cause()` in the listener, and `e` in the catch block; switch both to
parameterized logging.
Contributor guide
Research direction
Start in remoting/src/main/java/org/apache/rocketmq/remoting/netty/NettyRemotingAbstract.java, focusing on invoke0, requestFail, failFast, and invokeOnewayImpl. Add or run the unit-level failed-write test described in the issue and inspect ResponseFuture handling. Done means the Netty throwable reaches the caller and relevant warning logs, while channel closure reports its cause.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- java
- Domain
- backend, networking
- Issue type
- Bug
- Difficulty
- 3/5
- Estimated time
- 1-2 days
- Activity status
- Active
- Clarity
- Clearly specified
- Newbie friendliness
- 78/100