tchivs opened a new issue, #11010:
URL: https://github.com/apache/rocketmq/issues/11010
### 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<Throwable> 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.<init>(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.
--
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]
