akpatnam25 opened a new pull request, #3630:
URL: https://github.com/apache/celeborn/pull/3630
<!--
Thanks for sending a pull request! Here are some tips for you:
- Make sure the PR title start w/ a JIRA ticket, e.g. '[CELEBORN-XXXX]
Your PR title ...'.
- Be sure to keep the PR description updated to reflect all changes.
- Please write your PR title to summarize what this PR proposes.
- If possible, provide a concise example to reproduce the issue for a
faster review.
-->
### What changes were proposed in this pull request?
While running jobs with TLS enabled, we encountered memory leaks which cause
worker OOMs.
```
26/02/13 21:02:52,779 ERROR [push-server-9-9] ResourceLeakDetector: LEAK:
ByteBuf.release() was not called before it's garbage-collected. See
https://netty.io/wiki/reference-counted-objects.html for more information.
Recent access records:
Created at:
io.netty.buffer.AbstractByteBufAllocator.compositeDirectBuffer(AbstractByteBufAllocator.java:224)
io.netty.buffer.AbstractByteBufAllocator.compositeBuffer(AbstractByteBufAllocator.java:202)
org.apache.celeborn.common.network.util.TransportFrameDecoder.decodeNext(TransportFrameDecoder.java:143)
org.apache.celeborn.common.network.util.TransportFrameDecoder.channelRead(TransportFrameDecoder.java:66)
io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:444)
io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:420)
io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:412)
io.netty.handler.ssl.SslHandler.unwrap(SslHandler.java:1475)
io.netty.handler.ssl.SslHandler.decodeJdkCompatible(SslHandler.java:1338)
io.netty.handler.ssl.SslHandler.decode(SslHandler.java:1387)
io.netty.handler.codec.ByteToMessageDecoder.decodeRemovalReentryProtection(ByteToMessageDecoder.java:529)
io.netty.handler.codec.ByteToMessageDecoder.callDecode(ByteToMessageDecoder.java:468)
io.netty.handler.codec.ByteToMessageDecoder.channelRead(ByteToMessageDecoder.java:290)
io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:444)
io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:420)
io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:412)
io.netty.channel.DefaultChannelPipeline$HeadContext.channelRead(DefaultChannelPipeline.java:1410)
io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:440)
io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:420)
io.netty.channel.DefaultChannelPipeline.fireChannelRead(DefaultChannelPipeline.java:919)
io.netty.channel.nio.AbstractNioByteChannel$NioByteUnsafe.read(AbstractNioByteChannel.java:166)
io.netty.channel.nio.NioEventLoop.processSelectedKey(NioEventLoop.java:788)
io.netty.channel.nio.NioEventLoop.processSelectedKeysOptimized(NioEventLoop.java:724)
io.netty.channel.nio.NioEventLoop.processSelectedKeys(NioEventLoop.java:650)
io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:562)
io.netty.util.concurrent.SingleThreadEventExecutor$4.run(SingleThreadEventExecutor.java:997)
io.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74)
io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30)
java.base/java.lang.Thread.run(Thread.java:840)
```
When a Celeborn worker receives a PushData or PushMergedData message, it
replicates that frame to a secondary worker for fault tolerance. On an
SSL-enabled cluster this replication goes through SslMessageEncoder.encode().
Here is the flow of what happens inside SslMessageEncoder.encode():
- The encoder asks the message body for an SSL-friendly copy by calling
convertToNettyForSsl(). For shuffle data, the body is a NettyManagedBuffer —
data already loaded in off-heap memory. This call runs
buf.duplicate().retain(), which creates a second reference to the same memory
and increments the reference count from 1 to 2.
- The encoder places this second reference inside a composite buffer and
hands it to Netty for writing.
- Netty writes the composite to the network, then releases it — decrementing
the count from 2 to 1.
- Nothing releases the original NettyManagedBuffer's hold on the data, so
the count stays at 1 forever.
- This results in every replicated PushData frame leaking a chunk of
off-heap memory, eventually causing OOM and worker crash.
### Why are the changes needed?
fix memory leak
### Does this PR resolve a correctness bug?
<!-- Yes/No. (Note: If yes, committer will add `correctness` label to
current pull request). -->
### Does this PR introduce _any_ user-facing change?
no
### How was this patch tested?
already internally in production and tested.
Also added unit tests
--
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]
