Kafka Connector OOM when there're lots of messages in Couchbase

I have a 12 node couchbase cluster with couchbase 3.0.3 installed.
I am using Kafka connector 1.1.0;
There're 4 billion items in couchbase.
I used the Kafka connector to streaming the data out.

The memory foot print continue goes up, and at last it reports OOM, the error looks something like this:
2016-02-22 02:31:17,782 WARN AbstractGenericHandler:171 - [lvsdbx1406.lvs.ebay.com/10.29.4.30:11210][DCPEndpoint]: Caught unknown exception: java.lang.OutOfMemoryError: Direct buffer memory
com.couchbase.client.deps.io.netty.handler.codec.DecoderException: java.lang.OutOfMemoryError: Direct buffer memory
at com.couchbase.client.deps.io.netty.handler.codec.ByteToMessageDecoder.channelRead(ByteToMessageDecoder.java:248)
at com.couchbase.client.deps.io.netty.channel.CombinedChannelDuplexHandler.channelRead(CombinedChannelDuplexHandler.java:147)
at com.couchbase.client.deps.io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:308)
at com.couchbase.client.deps.io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:294)
at com.couchbase.client.deps.io.netty.channel.DefaultChannelPipeline.fireChannelRead(DefaultChannelPipeline.java:846)
at com.couchbase.client.deps.io.netty.channel.nio.AbstractNioByteChannel$NioByteUnsafe.read(AbstractNioByteChannel.java:131)
at com.couchbase.client.deps.io.netty.channel.nio.NioEventLoop.processSelectedKey(NioEventLoop.java:511)
at com.couchbase.client.deps.io.netty.channel.nio.NioEventLoop.processSelectedKeysOptimized(NioEventLoop.java:468)
at com.couchbase.client.deps.io.netty.channel.nio.NioEventLoop.processSelectedKeys(NioEventLoop.java:382)
at com.couchbase.client.deps.io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:354)
at com.couchbase.client.deps.io.netty.util.concurrent.SingleThreadEventExecutor$2.run(SingleThreadEventExecutor.java:112)
at com.couchbase.client.deps.io.netty.util.concurrent.DefaultThreadFactory$DefaultRunnableDecorator.run(DefaultThreadFactory.java:137)
at java.lang.Thread.run(Thread.java:745)
Caused by: java.lang.OutOfMemoryError: Direct buffer memory
at java.nio.Bits.reserveMemory(Bits.java:658)
at java.nio.DirectByteBuffer.<init>(DirectByteBuffer.java:123)
at java.nio.ByteBuffer.allocateDirect(ByteBuffer.java:306)
at com.couchbase.client.deps.io.netty.buffer.PoolArena$DirectArena.newChunk(PoolArena.java:645)
at com.couchbase.client.deps.io.netty.buffer.PoolArena.allocateNormal(PoolArena.java:228)
at com.couchbase.client.deps.io.netty.buffer.PoolArena.allocate(PoolArena.java:212)
at com.couchbase.client.deps.io.netty.buffer.PoolArena.reallocate(PoolArena.java:358)
at com.couchbase.client.deps.io.netty.buffer.PooledByteBuf.capacity(PooledByteBuf.java:118)
at com.couchbase.client.deps.io.netty.buffer.AbstractByteBuf.ensureWritable(AbstractByteBuf.java:250)
at com.couchbase.client.deps.io.netty.buffer.AbstractByteBuf.writeBytes(AbstractByteBuf.java:848)
at com.couchbase.client.deps.io.netty.buffer.AbstractByteBuf.writeBytes(AbstractByteBuf.java:840)
at com.couchbase.client.deps.io.netty.buffer.AbstractByteBuf.writeBytes(AbstractByteBuf.java:830)
at com.couchbase.client.deps.io.netty.handler.codec.ByteToMessageDecoder$1.cumulate(ByteToMessageDecoder.java:92)
at com.couchbase.client.deps.io.netty.handler.codec.ByteToMessageDecoder.channelRead(ByteToMessageDecoder.java:242)
... 12 more

I tried to do some heap dump, and 94% memory are used by the byte[];