Nio|Epoll|KqueueEventLoop task execution might throw UnsupportedOpera…

…tionException on shutdown. (netty#8476)

Motivation:

There is a racy UnsupportedOperationException instead because the task removal is delegated to MpscChunkedArrayQueue that does not support removal. This happens with SingleThreadEventExecutor that overrides the newTaskQueue to return an MPSC queue instead of the LinkedBlockingQueue returned by the base class such as NioEventLoop, EpollEventLoop and KQueueEventLoop.

Modifications:

- Catch the UnsupportedOperationException
- Add unit test.

Result:

Fix netty#8475

common/src/main/java/io/netty/util/concurrent/SingleThreadEventExecutor.java

@@ -778,8 +778,20 @@ public void execute(Runnable task) {
         addTask(task);
         if (!inEventLoop) {
             startThread();
-            if (isShutdown() && removeTask(task)) {
-                reject();
+            if (isShutdown()) {
+                boolean reject = false;
+                try {
+                    if (removeTask(task)) {
+                        reject = true;
+                    }
+                } catch (UnsupportedOperationException e) {
+                    // The task queue does not support removal so the best thing we can do is to just move on and
+                    // hope we will be able to pick-up the task before its completely terminated.
+                    // In worst case we will log on termination.
+                }
+                if (reject) {
+                    reject();
+                }
             }
         }
 

transport/src/test/java/io/netty/channel/nio/NioEventLoopTest.java

@@ -22,14 +22,17 @@
 import io.netty.channel.socket.ServerSocketChannel;
 import io.netty.channel.socket.nio.NioServerSocketChannel;
 import io.netty.util.concurrent.Future;
+import org.hamcrest.core.IsInstanceOf;
 import org.junit.Test;
 
 import java.net.InetSocketAddress;
 import java.nio.channels.SelectionKey;
 import java.nio.channels.Selector;
 import java.nio.channels.SocketChannel;
 import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.RejectedExecutionException;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
 
 import static org.junit.Assert.*;
 
@@ -171,4 +174,41 @@ public void channelUnregistered(SocketChannel ch, Throwable cause) {
             group.shutdownGracefully();
         }
     }
+
+    @SuppressWarnings("deprecation")
+    @Test
+    public void testTaskRemovalOnShutdownThrowsNoUnsupportedOperationException() throws Exception {
+        final AtomicReference<Throwable> error = new AtomicReference<Throwable>();
+        final Runnable task = new Runnable() {
+            @Override
+            public void run() {
+                // NOOP
+            }
+        };
+        // Just run often enough to trigger it normally.
+        for (int i = 0; i < 1000; i++) {
+            NioEventLoopGroup group = new NioEventLoopGroup(1);
+            final NioEventLoop loop = (NioEventLoop) group.next();
+
+            Thread t = new Thread(new Runnable() {
+                @Override
+                public void run() {
+                    try {
+                        for (;;) {
+                            loop.execute(task);
+                        }
+                    } catch (Throwable cause) {
+                        error.set(cause);
+                    }
+                }
+            });
+            t.start();
+            group.shutdownNow();
+            t.join();
+            group.terminationFuture().syncUninterruptibly();
+            assertThat(error.get(), IsInstanceOf.instanceOf(RejectedExecutionException.class));
+            error.set(null);
+        }
+    }
+
 }