diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/ipc/Netty4RpcServer.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/ipc/Netty4RpcServer.java index 4d51935..60af4ec 100644 --- a/hbase-server/src/main/java/org/apache/hadoop/hbase/ipc/Netty4RpcServer.java +++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/ipc/Netty4RpcServer.java @@ -21,8 +21,8 @@ package org.apache.hadoop.hbase.ipc; import static org.apache.hadoop.fs.CommonConfigurationKeysPublic.HADOOP_SECURITY_AUTHORIZATION; import io.netty.bootstrap.ServerBootstrap; import io.netty.buffer.ByteBuf; +import io.netty.buffer.PooledByteBufAllocator; import io.netty.buffer.Unpooled; -import io.netty.buffer.UnpooledByteBufAllocator; import io.netty.channel.Channel; import io.netty.channel.ChannelFuture; import io.netty.channel.ChannelFutureListener; @@ -219,11 +219,12 @@ public class Netty4RpcServer implements RpcServerInterface, } else { bootstrap.channel(NioServerSocketChannel.class); } - //bootstrap.option(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT); + //bootstrap.option(ChannelOption.ALLOCATOR, new PooledByteBufAllocator(true)); bootstrap.childOption(ChannelOption.TCP_NODELAY, true); bootstrap.childOption(ChannelOption.SO_KEEPALIVE, true); bootstrap.childOption(ChannelOption.SO_LINGER, 0); - //bootstrap.childOption(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT); + bootstrap.childOption(ChannelOption.ALLOCATOR, new PooledByteBufAllocator( + true)); // bootstrap.childOption(ChannelOption.WRITE_BUFFER_HIGH_WATER_MARK, 32 * 1024); // bootstrap.childOption(ChannelOption.WRITE_BUFFER_LOW_WATER_MARK, 8 * 1024); bootstrap.childHandler(new Initializer()); @@ -840,11 +841,14 @@ public class Netty4RpcServer implements RpcServerInterface, LOG.warn("Exception while creating response " + e); } this.response = bc; - responseBB = UnpooledByteBufAllocator.DEFAULT.buffer(this.response.size()); - ByteBuffer[] buffers = this.response.getBuffers(); - for (ByteBuffer bb : buffers) { - responseBB.writeBytes(bb); - } + + // responseBB = + // UnpooledByteBufAllocator.DEFAULT.buffer(this.response.size()); + // ByteBuffer[] buffers = this.response.getBuffers(); + // for (ByteBuffer bb : buffers) { + // responseBB.writeBytes(bb); + // } + responseBB = Unpooled.wrappedBuffer(this.response.getBuffers()); } private BufferChain wrapWithSasl(BufferChain bc) throws IOException { @@ -1149,6 +1153,8 @@ public class Netty4RpcServer implements RpcServerInterface, try { if (msg instanceof ByteBuf) { ByteBuf data = (ByteBuf) msg; + // LOG.info("data: " + data.getClass() + ", " + data.readableBytes(), + // new Exception()); if (cumulation == null) { cumulation = data; try {