ISSUE-2729: 【BUG】BookKeeper Netty chennel OOM

[sijie]

Original Issue: apache#2729

---

BUG REPORT

Describe the bug
When running hundreds of bookie in BookKeeper cluster, and auditor triggered to check ledgers, it throw OOM

18:12:03.200 [bookkeeper-io-44-2] ERROR org.apache.bookkeeper.common.allocator.impl.ByteBufAllocatorImpl - Unable to allocate memory
java.lang.OutOfMemoryError: Direct buffer memory
        at java.nio.Bits.reserveMemory(Bits.java:187) ~[?:?]
        at java.nio.DirectByteBuffer.<init>(DirectByteBuffer.java:123) ~[?:?]
        at java.nio.ByteBuffer.allocateDirect(ByteBuffer.java:310) ~[?:?]
        at io.netty.buffer.PoolArena$DirectArena.allocateDirect(PoolArena.java:755) ~[io.netty-netty-buffer-4.1.51.Final.jar:4.1.51.Final]
        at io.netty.buffer.PoolArena$DirectArena.newChunk(PoolArena.java:731) ~[io.netty-netty-buffer-4.1.51.Final.jar:4.1.51.Final]
        at io.netty.buffer.PoolArena.allocateNormal(PoolArena.java:247) ~[io.netty-netty-buffer-4.1.51.Final.jar:4.1.51.Final]
        at io.netty.buffer.PoolArena.allocate(PoolArena.java:215) ~[io.netty-netty-buffer-4.1.51.Final.jar:4.1.51.Final]
        at io.netty.buffer.PoolArena.allocate(PoolArena.java:147) ~[io.netty-netty-buffer-4.1.51.Final.jar:4.1.51.Final]
        at io.netty.buffer.PooledByteBufAllocator.newDirectBuffer(PooledByteBufAllocator.java:356) ~[io.netty-netty-buffer-4.1.51.Final.jar:4.1.51.Final]
        at io.netty.buffer.AbstractByteBufAllocator.directBuffer(AbstractByteBufAllocator.java:187) ~[io.netty-netty-buffer-4.1.51.Final.jar:4.1.51.Final]
        at org.apache.bookkeeper.common.allocator.impl.ByteBufAllocatorImpl.newDirectBuffer(ByteBufAllocatorImpl.java:164) [org.apache.bookkeeper-bookkeeper-common-allocator-4.12.0.jar:4.12.0]
        at org.apache.bookkeeper.common.allocator.impl.ByteBufAllocatorImpl.newDirectBuffer(ByteBufAllocatorImpl.java:158) [org.apache.bookkeeper-bookkeeper-common-allocator-4.12.0.jar:4.12.0]
        at io.netty.buffer.AbstractByteBufAllocator.directBuffer(AbstractByteBufAllocator.java:187) [io.netty-netty-buffer-4.1.51.Final.jar:4.1.51.Final]
        at org.apache.bookkeeper.proto.BookieProtoEncoding.serializeProtobuf(BookieProtoEncoding.java:366) [org.apache.bookkeeper-bookkeeper-server-4.12.0.jar:4.12.0]
        at org.apache.bookkeeper.proto.BookieProtoEncoding.access$100(BookieProtoEncoding.java:54) [org.apache.bookkeeper-bookkeeper-server-4.12.0.jar:4.12.0]
        at org.apache.bookkeeper.proto.BookieProtoEncoding$RequestEnDecoderV3.encode(BookieProtoEncoding.java:328) [org.apache.bookkeeper-bookkeeper-server-4.12.0.jar:4.12.0]
        at org.apache.bookkeeper.proto.BookieProtoEncoding$RequestEncoder.write(BookieProtoEncoding.java:400) [org.apache.bookkeeper-bookkeeper-server-4.12.0.jar:4.12.0]
        at io.netty.channel.AbstractChannelHandlerContext.invokeWrite0(AbstractChannelHandlerContext.java:717) [io.netty-netty-transport-4.1.51.Final.jar:4.1.51.Final]
        at io.netty.channel.AbstractChannelHandlerContext.invokeWriteAndFlush(AbstractChannelHandlerContext.java:764) [io.netty-netty-transport-4.1.51.Final.jar:4.1.51.Final]
        at io.netty.channel.AbstractChannelHandlerContext.write(AbstractChannelHandlerContext.java:790) [io.netty-netty-transport-4.1.51.Final.jar:4.1.51.Final]
        at io.netty.channel.AbstractChannelHandlerContext.writeAndFlush(AbstractChannelHandlerContext.java:758) [io.netty-netty-transport-4.1.51.Final.jar:4.1.51.Final]
        at io.netty.channel.AbstractChannelHandlerContext.writeAndFlush(AbstractChannelHandlerContext.java:808) [io.netty-netty-transport-4.1.51.Final.jar:4.1.51.Final]
        at org.apache.bookkeeper.proto.AuthHandler$ClientSideHandler$AuthHandshakeCompleteCallback.operationComplete(AuthHandler.java:437) [org.apache.bookkeeper-bookkeeper-server-4.12.0.jar:4.12.0]
        at org.apache.bookkeeper.proto.AuthHandler$ClientSideHandler$AuthHandshakeCompleteCallback.operationComplete(AuthHandler.java:423) [org.apache.bookkeeper-bookkeeper-server-4.12.0.jar:4.12.0]
        at org.apache.bookkeeper.auth.AuthProviderFactoryFactory$NullClientAuthProviderFactory.newProvider(AuthProviderFactoryFactory.java:102) [org.apache.bookkeeper-bookkeeper-server-4.12.0.jar:4.12.0]
        at org.apache.bookkeeper.proto.AuthHandler$ClientSideHandler.channelActive(AuthHandler.java:258) [org.apache.bookkeeper-bookkeeper-server-4.12.0.jar:4.12.0]
        at io.netty.channel.AbstractChannelHandlerContext.invokeChannelActive(AbstractChannelHandlerContext.java:230) [io.netty-netty-transport-4.1.51.Final.jar:4.1.51.Final]
        at io.netty.channel.AbstractChannelHandlerContext.invokeChannelActive(AbstractChannelHandlerContext.java:216) [io.netty-netty-transport-4.1.51.Final.jar:4.1.51.Final]
        at io.netty.channel.AbstractChannelHandlerContext.fireChannelActive(AbstractChannelHandlerContext.java:209) [io.netty-netty-transport-4.1.51.Final.jar:4.1.51.Final]
        at io.netty.channel.DefaultChannelPipeline$HeadContext.channelActive(DefaultChannelPipeline.java:1398) [io.netty-netty-transport-4.1.51.Final.jar:4.1.51.Final]
        at io.netty.channel.AbstractChannelHandlerContext.invokeChannelActive(AbstractChannelHandlerContext.java:230) [io.netty-netty-transport-4.1.51.Final.jar:4.1.51.Final]
        at io.netty.channel.AbstractChannelHandlerContext.invokeChannelActive(AbstractChannelHandlerContext.java:216) [io.netty-netty-transport-4.1.51.Final.jar:4.1.51.Final]
        at io.netty.channel.DefaultChannelPipeline.fireChannelActive(DefaultChannelPipeline.java:895) [io.netty-netty-transport-4.1.51.Final.jar:4.1.51.Final]
        at io.netty.channel.epoll.AbstractEpollChannel$AbstractEpollUnsafe.fulfillConnectPromise(AbstractEpollChannel.java:620) [io.netty-netty-transport-native-epoll-4.1.51.Final.jar:4.1.51.Final]
        at io.netty.channel.epoll.AbstractEpollChannel$AbstractEpollUnsafe.finishConnect(AbstractEpollChannel.java:653) [io.netty-netty-transport-native-epoll-4.1.51.Final.jar:4.1.51.Final]
        at io.netty.channel.epoll.AbstractEpollChannel$AbstractEpollUnsafe.epollOutReady(AbstractEpollChannel.java:529) [io.netty-netty-transport-native-epoll-4.1.51.Final.jar:4.1.51.Final]
        at io.netty.channel.epoll.EpollEventLoop.processReady(EpollEventLoop.java:465) [io.netty-netty-transport-native-epoll-4.1.51.Final.jar:4.1.51.Final]
        at io.netty.channel.epoll.EpollEventLoop.run(EpollEventLoop.java:378) [io.netty-netty-transport-native-epoll-4.1.51.Final.jar:4.1.51.Final]
        at io.netty.util.concurrent.SingleThreadEventExecutor$4.run(SingleThreadEventExecutor.java:989) [io.netty-netty-common-4.1.51.Final.jar:4.1.51.Final]
        at io.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74) [io.netty-netty-common-4.1.51.Final.jar:4.1.51.Final]
        at io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30) [io.netty-netty-common-4.1.51.Final.jar:4.1.51.Final]
        at java.lang.Thread.run(Thread.java:844) [?:?]

The code introduced by apache#2441 , and it change serializeProtobuf message from heap to direct memory. The code as follow

private static ByteBuf serializeProtobuf(MessageLite msg, ByteBufAllocator allocator) {
        int size = msg.getSerializedSize();
        // Protobuf serialization is the last step of the netty pipeline. We used to allocate
        // a heap buffer while serializing and pass it down to netty library.
        // In AbstractChannel#filterOutboundMessage(), netty copies that data to a direct buffer if
        // it is currently in heap (otherwise skips it and uses it directly).
        // Allocating a direct buffer reducing unncessary CPU cycles for buffer copies in BK client
        // and also helps alleviate pressure off the GC, since there is less memory churn.
        // Bookies aren't usually CPU bound. This change improves READ_ENTRY code paths by a small factor as well.
        ByteBuf buf = allocator.directBuffer(size, size);

        try {
            msg.writeTo(CodedOutputStream.newInstance(buf.nioBuffer(buf.readerIndex(), size)));
        } catch (IOException e) {
            // This is in-memory serialization, should not fail
            throw new RuntimeException(e);
        }

        // Advance writer idx
        buf.writerIndex(buf.capacity());
        return buf;
   }

I doubt whether will casue direct memory leak due to not release.