Need help resolving Error Msg: "WARNING: Failed to release a message: SimpleLeakAwareByteBuf(SlicedAbstractByteBuf(freed))"

[L_DisplayName]

Greetings All,

  I am receiving the error "WARNING: Failed to release a message: SimpleLeakAwareByteBuf(SlicedAbstractByteBuf(freed) io.netty.util.IllegalReferenceCountException: refCnt: 0, decrement: 1" after executing a custom netty software application that transfers a single file from a client to a server through an intermediate (proxy) server. Can any one tell me how I can resolve this error?  The complete error message is specified below. The code is based off of netty's HexProxyDump example.  The code rendering the error is specified below also.  Note the code transfers a single 1GB file from a client to a server through a proxy server as follows: Client --> Proxy Server --> Server

Error Message (rendered on the proxy server that forwards all data coming in on it's inbound channel to it's outbound channel)

Sep 15, 2016 9:31:38 AM io.netty.util.ReferenceCountUtil safeRelease

WARNING: Failed to release a message: SimpleLeakAwareByteBuf(SlicedAbstractByteBuf(freed))

io.netty.util.IllegalReferenceCountException: refCnt: 0, decrement: 1

        at io.netty.buffer.AbstractReferenceCountedByteBuf.release(AbstractReferenceCountedByteBuf.java:101)

        at io.netty.buffer.AbstractDerivedByteBuf.release(AbstractDerivedByteBuf.java:50)

        at io.netty.buffer.WrappedByteBuf.release(WrappedByteBuf.java:827)

        at io.netty.buffer.SimpleLeakAwareByteBuf.release(SimpleLeakAwareByteBuf.java:34)

        at io.netty.util.ReferenceCountUtil.release(ReferenceCountUtil.java:59)

        at io.netty.util.ReferenceCountUtil.safeRelease(ReferenceCountUtil.java:84)

        at io.netty.channel.ChannelOutboundBuffer.remove0(ChannelOutboundBuffer.java:296)

        at io.netty.channel.ChannelOutboundBuffer.failFlushed(ChannelOutboundBuffer.java:621)

        at io.netty.channel.AbstractChannel$AbstractUnsafe.close(AbstractChannel.java:593)

        at io.netty.channel.AbstractChannel$AbstractUnsafe.flush0(AbstractChannel.java:776)

        at io.netty.channel.nio.AbstractNioChannel$AbstractNioUnsafe.forceFlush(AbstractNioChannel.java:317)

        at io.netty.channel.nio.NioEventLoop.processSelectedKey(NioEventLoop.java:519)

        at io.netty.channel.nio.NioEventLoop.processSelectedKeysOptimized(NioEventLoop.java:468)

        at io.netty.channel.nio.NioEventLoop.processSelectedKeys(NioEventLoop.java:382)

        at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:354)

        at io.netty.util.concurrent.SingleThreadEventExecutor$2.run(SingleThreadEventExecutor.java:112)

        at io.netty.util.concurrent.DefaultThreadFactory$DefaultRunnableDecorator.run(DefaultThreadFactory.java:137)

        at java.lang.Thread.run(Thread.java:745)

Code Running on the proxy Server: 

1. FileReceiverAndForwarder.java (BootStraps the server)

2. FileReceiverAndForwarderInitializer.java (Initializes the channel pipeline configuraiton)

3. FileReceiverHandler.java (This is where the error is being rendered from, within the ChannelRead0 method) This is where ByteBuf Msgs are forwarded from the inbound channel to the outbound channel)

4. FileForwarderHandler.java 

FileReceiverHandler.java code (code that generates the error within the ChannelRead0 method)

public class FileReceiverHandler extends SimpleChannelInboundHandler<ByteBuf> {


    private final String remoteHost;
    private final int remotePort;
    private final boolean amIaForwarder;
    private Logger logger;
    private final String theFilePath;
    private volatile Channel outboundChannel;

    //for the file
    private long currentOffset;
    private File emptyFile;
    private RandomAccessFile f;
    private FileChannel fc;
    private boolean readLengthOfFile = false;
    private boolean wasFileCreated;
 
    public FileReceiverHandler(String remoteHost, int remotePort, boolean amIaForwarder) throws Exception {
        this.remoteHost = remoteHost;
        this.remotePort = remotePort;
        this.amIaForwarder = amIaForwarder;
logger = Logger.getLogger(FileReceiverHandler.class.getName());
        theFilePath = "/root/1GB_File_COPY.dat";
        currentOffset = 0;
        fc = null;
        emptyFile = null;
        f = null;
        fc = null;
       wasFileCreated = false;


    }


    @Override
    public void channelActive(ChannelHandlerContext ctx) {
        logger.info("RECEIVER HANDLER ACTIVE METHOD: Current Machine( " + ctx.channel().localAddress().toString() + " ) is connected to the Remote Machine( " + ctx.channel().remoteAddress().toString() + " )");
        final Channel inboundChannel = ctx.channel();


        if (amIaForwarder) {
            // Start the connection attempt.
            Bootstrap b = new Bootstrap();
            b.group(inboundChannel.eventLoop())
                    .channel(ctx.channel().getClass())
.option(ChannelOption.AUTO_READ, false);
            ChannelFuture f = b.connect(remoteHost, remotePort);
            outboundChannel = f.channel();
            f.addListener(new ChannelFutureListener() {
                @Override
                public void operationComplete(ChannelFuture future) {
                    if (future.isSuccess()) {
                        // connection complete start to read first data
                        inboundChannel.read();
                    } else {
                        // Close the connection if the connection attempt has failed.
                        inboundChannel.close();
                  }
                }
            });
        }
        else {
            try {
                emptyFile = new File(theFilePath);
                f = new RandomAccessFile(emptyFile, "rw");
                fc = f.getChannel();
                wasFileCreated = emptyFile.createNewFile();
            }catch(Exception e){
               System.err.printf("Error Msg" + e.getMessage());
            }
        }
    }


    @Override
    public void channelRead0(final ChannelHandlerContext ctx, ByteBuf msg) throws Exception {
        if (amIaForwarder) {
            if (outboundChannel.isActive()) {
                outboundChannel.writeAndFlush(msg).addListener(new ChannelFutureListener() {
                    @Override
                    public void operationComplete(ChannelFuture future) {
                        if (future.isSuccess()) {
                            // was able to flush out data, start to read the next chunk
                            ctx.channel().read();
                        } else {
                            future.channel().close();
                        }
                    }
                });
            }
        } else {
            logger.info("Client:channel Reading " + msg.readableBytes() + "Bytes and Writing to file: " + theFilePath + " at offset " + currentOffset);
                fc.write(msg.nioBuffer(), currentOffset);
                currentOffset += msg.readableBytes(); //readable bytes is the number of bytes readable = length of Bytes In ByteBuf
            }


        }


    @Override
    public void channelInactive(ChannelHandlerContext ctx) {
        if (outboundChannel != null) {
            closeOnFlush(outboundChannel);
        }
    }


    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
        cause.printStackTrace();
        closeOnFlush(ctx.channel());
    }


    /**
     * Closes the specified channel after all queued write requests are flushed.
     */
    static void closeOnFlush(Channel ch) {
        if (ch.isActive()) {
            ch.writeAndFlush(Unpooled.EMPTY_BUFFER).addListener(ChannelFutureListener.CLOSE);
        }
    }
}

FileReceiverAndForwarder.java

public final class FileReceiverAndForwarder {


    static final int LOCAL_PORT = Integer.parseInt(System.getProperty("localPort", "4959"));
    static final String REMOTE_HOST = System.getProperty("remoteHost", "179.10.35.33");
    static final int REMOTE_PORT = Integer.parseInt(System.getProperty("remotePort", "4959"));
    private final static Logger logger = Logger.getLogger(FileReceiverAndForwarder.class.getName());


    public static void main(String[] args) throws Exception {


        System.out.printf("Enter local port(int) and (true or false) if this server is a proxy server");


        Scanner scanner = new Scanner(System.in);
        final int localPort = scanner.nextInt();
        final String forwarder = scanner.next();
        final boolean amIaForwarder = (forwarder.equalsIgnoreCase("true")?true:false);
final boolean amIaForwarder = (forwarder.equalsIgnoreCase("true")?true:false);


        // Configure the bootstrap.
        EventLoopGroup bossGroup = new NioEventLoopGroup(1);
        EventLoopGroup workerGroup = new NioEventLoopGroup();
        try {
            ServerBootstrap b = new ServerBootstrap();
            b.group(bossGroup, workerGroup)
             .channel(NioServerSocketChannel.class)
             .handler(new LoggingHandler(LogLevel.INFO))
             .childHandler(new FileReceiverAndForwarderInitializer(REMOTE_HOST,REMOTE_PORT ,amIaForwarder))
             .childOption(ChannelOption.AUTO_READ, false)
             .bind(LOCAL_PORT).sync().channel().closeFuture().sync();
             logger.info("Started the Server on port " + localPort);
        } finally {
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }
    }

FileReceiverAndForwarderInitializer.java

public class FileReceiverAndForwarderInitializer extends ChannelInitializer<SocketChannel> {


    private final String remoteHost;
    private final int remotePort;
    private final boolean amIaForwarder;


    public FileReceiverAndForwarderInitializer(String remoteHost, int remotePort, boolean amIaForwarder) {
        this.remoteHost = remoteHost;
        this.remotePort = remotePort;
        this.amIaForwarder = amIaForwarder;
    }


    @Override
    public void initChannel(SocketChannel ch) throws Exception {
        ch.pipeline().addLast(
                new LengthFieldBasedFrameDecoder(1024*1024*128, 0, 8, 0, 8),
                new FileReceiverHandler(remoteHost, remotePort, amIaForwarder));
   }
}