ChannelInitializer.initChannel() executed more than once when used with context executor

[atcurtis]

Expected behavior

Expected ChannelInitializer.initChannel() to be executed one.

Actual behavior

A subsequent execution causes exception to be thrown because it can't find itself or remove itself again.

2018-12-01 01:22:02,527 [WARN] [pool-2-thread-1] ChannelInitializer ? Failed to initialize a channel. Closing: [id: 0xce54469f, L:local:test - R:local:E:d49bc512]
java.lang.AssertionError: expected object to not be null
	at org.testng.Assert.fail(Assert.java:93) ~[testng-6.11.jar:?]
	at org.testng.Assert.assertNotNull(Assert.java:422) ~[testng-6.11.jar:?]
	at org.testng.Assert.assertNotNull(Assert.java:407) ~[testng-6.11.jar:?]
	at test.TestChannelInitializer$2.initChannel(TestChannelInitializer.java:93) ~[espresso-router-impl/:?]
	at io.netty.channel.ChannelInitializer.initChannel(ChannelInitializer.java:113) [netty-all-4.1.22.Final.jar:4.1.22.Final]
	at io.netty.channel.ChannelInitializer.channelRegistered(ChannelInitializer.java:76) [netty-all-4.1.22.Final.jar:4.1.22.Final]
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRegistered(AbstractChannelHandlerContext.java:149) [netty-all-4.1.22.Final.jar:4.1.22.Final]
	at io.netty.channel.AbstractChannelHandlerContext.access$000(AbstractChannelHandlerContext.java:38) [netty-all-4.1.22.Final.jar:4.1.22.Final]
	at io.netty.channel.AbstractChannelHandlerContext$1.run(AbstractChannelHandlerContext.java:140) [netty-all-4.1.22.Final.jar:4.1.22.Final]
	at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511) [?:1.8.0_121]
	at java.util.concurrent.FutureTask.run(FutureTask.java:266) [?:1.8.0_121]
	at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$201(ScheduledThreadPoolExecutor.java:180) [?:1.8.0_121]
	at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:293) [?:1.8.0_121]
	at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142) [?:1.8.0_121]
	at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617) [?:1.8.0_121]
	at java.lang.Thread.run(Thread.java:745) [?:1.8.0_121]
2018-12-01 01:22:02,533 [WARN] [pool-2-thread-1] LoggingHandler - [id: 0xce54469f, L:local:test ! R:local:E:d49bc512] READ: Hello World
2018-12-01 01:22:02,533 [WARN] [pool-2-thread-1] LoggingHandler - [id: 0xce54469f, L:local:test ! R:local:E:d49bc512] READ COMPLETE
2018-12-01 01:22:02,533 [WARN] [pool-2-thread-1] LoggingHandler - [id: 0xce54469f, L:local:test ! R:local:E:d49bc512] ACTIVE
2018-12-01 01:22:02,533 [WARN] [pool-2-thread-1] LoggingHandler - [id: 0xce54469f, L:local:test ! R:local:E:d49bc512] REGISTERED
2018-12-01 01:22:02,534 [WARN] [pool-2-thread-1] LoggingHandler - [id: 0xce54469f, L:local:test ! R:local:E:d49bc512] INACTIVE
2018-12-01 01:22:02,534 [WARN] [pool-2-thread-1] LoggingHandler - [id: 0xce54469f, L:local:test ! R:local:E:d49bc512] UNREGISTERED

java.lang.AssertionError: expected [1] but found [2]
Expected :1
Actual   :2


	at org.testng.Assert.fail(Assert.java:93)
	at org.testng.Assert.failNotEquals(Assert.java:512)
	at org.testng.Assert.assertEqualsImpl(Assert.java:134)
	at org.testng.Assert.assertEquals(Assert.java:115)
	at org.testng.Assert.assertEquals(Assert.java:388)
	at org.testng.Assert.assertEquals(Assert.java:398)
	at test.TestChannelInitializer.test(TestChannelInitializer.java:137)

Steps to reproduce

See repro code below.

Minimal yet complete reproducer code (or URL to code)

package test;

import io.netty.bootstrap.Bootstrap;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.DefaultEventLoopGroup;
import io.netty.channel.local.LocalAddress;
import io.netty.channel.local.LocalChannel;
import io.netty.channel.local.LocalServerChannel;
import io.netty.handler.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;
import io.netty.util.concurrent.AbstractEventExecutor;
import io.netty.util.concurrent.EventExecutor;
import io.netty.util.concurrent.Future;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import org.testng.Assert;
import org.testng.annotations.Test;


public class TestChannelInitializer {

  @Test
  public void test() throws InterruptedException {

    AtomicInteger invokeCount = new AtomicInteger();
    AtomicInteger completeCount = new AtomicInteger();

    LocalAddress addr = new LocalAddress("test");

    ChannelHandler logger = new LoggingHandler(LogLevel.WARN);

    ScheduledExecutorService execService = Executors.newSingleThreadScheduledExecutor();

    EventExecutor exec = new AbstractEventExecutor() {
      @Override
      public void shutdown() {
        throw new IllegalStateException();
      }

      @Override
      public boolean inEventLoop(Thread thread) {
        return false;
      }

      @Override
      public boolean isShuttingDown() {
        return false;
      }

      @Override
      public Future<?> shutdownGracefully(long quietPeriod, long timeout, TimeUnit unit) {
        throw new IllegalStateException();
      }

      @Override
      public Future<?> terminationFuture() {
        throw new IllegalStateException();
      }

      @Override
      public boolean isShutdown() {
        return execService.isShutdown();
      }

      @Override
      public boolean isTerminated() {
        return execService.isTerminated();
      }

      @Override
      public boolean awaitTermination(long timeout, TimeUnit unit) throws InterruptedException {
        return execService.awaitTermination(timeout, unit);
      }

      @Override
      public void execute(Runnable command) {
        execService.schedule(command, 1, TimeUnit.NANOSECONDS);
      }
    };

    ChannelInitializer<Channel> otherInitializer = new ChannelInitializer<Channel>() {
      @Override
      protected void initChannel(Channel ch) throws Exception {
        invokeCount.incrementAndGet();
        ChannelHandlerContext ctx = ch.pipeline().context(this);

        Assert.assertNotNull(ctx); // FAILS HERE

        ch.pipeline().addAfter(ctx.executor(), ctx.name(), null, logger);
        completeCount.decrementAndGet();
      }
    };

    DefaultEventLoopGroup group = new DefaultEventLoopGroup(1);

    ServerBootstrap serverBootstrap = new ServerBootstrap()
        .channel(LocalServerChannel.class)
        .group(group)
        .localAddress(addr)
        .childHandler(new ChannelInitializer<LocalChannel>() {
          @Override
          protected void initChannel(LocalChannel ch) throws Exception {
            ch.pipeline().addLast(exec, otherInitializer);
          }
        });

    Channel server = serverBootstrap.bind().sync().channel();

    Bootstrap clientBootstrap = new Bootstrap()
        .channel(LocalChannel.class)
        .group(group)
        .remoteAddress(addr)
        .handler(new ChannelInitializer<LocalChannel>() {
          @Override
          protected void initChannel(LocalChannel ch) throws Exception {
          }
        });

    Channel client = clientBootstrap.connect().sync().channel();

    client.writeAndFlush("Hello World").sync();

    Thread.sleep(1000);

    client.close().sync();
    server.close().sync();

    execService.shutdown();
    execService.awaitTermination(10, TimeUnit.SECONDS);

    Assert.assertEquals(invokeCount.get(), 1); // This seems to be 2 instead of 1
    Assert.assertEquals(invokeCount.get(), completeCount.get());
  }
}

Netty version

4.1.22.Final

JVM version (e.g. java -version)

java version "1.8.0_121"
Java(TM) SE Runtime Environment (build 1.8.0_121-b13)
Java HotSpot(TM) 64-Bit Server VM (build 25.121-b13, mixed mode)

OS version (e.g. uname -a)

RHEL7
3.10.0-514.55.4.el7.x86_64 #1 SMP Fri Aug 10 17:03:02 UTC 2018 x86_64 x86_64 x86_64 GNU/Linux

[normanmaurer]

@atcurtis can you please try netty 4.1.32.Final and let me know if the same problem happens there as well? I think I remember that we fixed something related to this at some point.

Please let me know if after the upgrade the problem still persists.

[atcurtis]

I have tried with 4.1.32.Final and it fails exactly the same except with slightly different line numbers:

	at io.netty.channel.ChannelInitializer.initChannel(ChannelInitializer.java:115)
	at io.netty.channel.ChannelInitializer.channelRegistered(ChannelInitializer.java:76)
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRegistered(AbstractChannelHandlerContext.java:149)
	at io.netty.channel.AbstractChannelHandlerContext.access$000(AbstractChannelHandlerContext.java:38)
	at io.netty.channel.AbstractChannelHandlerContext$1.run(AbstractChannelHandlerContext.java:140)

[normanmaurer]

I will have a look

…

Am 03.12.2018 um 21:08 schrieb Antony T Curtis ***@***.***>: I have tried with 4.1.32.Final and it fails exactly the same except with slightly different line numbers: at io.netty.channel.ChannelInitializer.initChannel(ChannelInitializer.java:115) at io.netty.channel.ChannelInitializer.channelRegistered(ChannelInitializer.java:76) at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRegistered(AbstractChannelHandlerContext.java:149) at io.netty.channel.AbstractChannelHandlerContext.access$000(AbstractChannelHandlerContext.java:38) at io.netty.channel.AbstractChannelHandlerContext$1.run(AbstractChannelHandlerContext.java:140) — You are receiving this because you commented. Reply to this email directly, view it on GitHub, or mute the thread.

[normanmaurer]

@atcurtis check #8620