HandlerRemoveRace.java

import io.netty.bootstrap.Bootstrap;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;

import java.net.InetSocketAddress;
import java.util.concurrent.CountDownLatch;

public class HandlerRemoveRace {

    public static void main(String[] args) throws Exception {
        NioEventLoopGroup eventLoop = new NioEventLoopGroup(1);

        try {
            // Dummy server that just accepts connections, nothing more
            ServerBootstrap server = new ServerBootstrap();
            server.group(eventLoop)
                    .channel(NioServerSocketChannel.class)
                    .childHandler(new ChannelInboundHandlerAdapter() {
                        @Override
                        public void channelRead(ChannelHandlerContext ctx, Object msg) { }
                    });
            server.bind(9999).sync();

            // Signals to coordinate threads and force the race condition
            CountDownLatch startWriting = new CountDownLatch(1);
            CountDownLatch doneWriting = new CountDownLatch(1);

            Bootstrap client = new Bootstrap()
                    .group(eventLoop)
                    .channel(NioSocketChannel.class)
                    .handler(new RemoveAfterConnect(startWriting, doneWriting));
            Channel ch = client.connect(new InetSocketAddress("127.0.0.1", 9999)).channel();

            startWriting.await();
            // IO thread is in channelActive, write immediately before IO thread removes handler
            ChannelFuture writeFuture = ch.write("BOOM");
            doneWriting.countDown();
            writeFuture.sync(); // should throw IllegalStateException from the write
        } catch (Exception e) {
            e.printStackTrace(System.out);
        } finally {
            eventLoop.shutdownGracefully();
        }
    }

    static class RemoveAfterConnect extends ChannelDuplexHandler {
        private final CountDownLatch startWriting, doneWriting;

        RemoveAfterConnect(CountDownLatch startWriting, CountDownLatch doneWriting) {
            this.startWriting = startWriting;
            this.doneWriting = doneWriting;
        }

        @Override
        public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
            System.out.printf("write: %s, (removed=%b)\n", msg, ctx.isRemoved());
            if (ctx.isRemoved()) {
                promise.setFailure(new IllegalStateException("write called after handler removed!"));
            }
            else promise.setSuccess();
        }

        @Override
        public void channelActive(ChannelHandlerContext ctx) throws Exception {
            System.out.println("channel active, signal writers");
            startWriting.countDown(); // start writing
            doneWriting.await(); // wait for writes
            System.out.println("now removing this handler");
            ctx.pipeline().remove(this);
            // Should find that the write tasks end up invoking this handler
            // because they were instantiated with a ref to this ChannelHandlerContext
            // just before the handler was removed.
        }
    }
}