Repository: netty/netty
Issue: [bug] when SO_LINGGER is used in the TCP half-closed scenario, the client in the fin_wait2 state cannot receive the data sent from the server which is in the close_wait state

![SO_LINGERHalfClosureBug](https://user-images.githubusercontent.com/5794695/148641718-987a2774-6d10-4400-8d32-c14feaa0417b.png)

as shown in the picture above。

**when `SO_LINGGER` is used in the client side**, and client call `NioSocketChannel#shutdownOutput()` to **half close** the tcp connection。

then server side `OP_READ` is active，server begin to `shutdowninput` and `triger ChannelInputShutdownEvent`，next send some  data to client in the process handler of the ChannelInputShutdownEvent event。

### Actual behavior

client in the `FIN_WAIT2` state **can not read and process these data** which is sended by server in the `CLOSE_WAIT` state.

### Expected behavior

client in the `FIN_WAIT2` state can read and process these data which is sended by server in the `CLOSE_WAIT` state.

### Steps to reproduce

```
 ServerBootstrap server = new ServerBootstrap();
 server.childOption(ChannelOption.ALLOW_HALF_CLOSURE, true)

 Bootstrap client = new Bootstrap();
 client.option(ChannelOption.SO_LINGER, 1)
```
### Minimal yet complete reproducer code (or URL to code)

[see unit test method testHalfClosureReceiveDataOnFinalWait2StateWhenSoLingerSet](https://github.com/huibinliupush/netty/blob/afd2de8d81146301465ebc62fa1b5675931dd950/testsuite/src/main/java/io/netty/testsuite/transport/socket/SocketHalfClosedTest.java)

### Netty version
4.1.72.Final




Issue comments:

Pull request review comments:
Can we just use no codec for this test ?
Please follow the same structure for tests as the rest of the file. This ensures the test will be run for NIO,OIO,Epoll and Kqueue. I suspect the same fix will need to go there as well. 
Like said above there is no need to use a codec here. 
Use an CountdownLatch this way you can also remove the `Thread.sleep(...)` below.
See above 
Call `ReferenceCountUtil.release(msg);`
Call `ReferenceCountUtil.release(msg);`
Like said above remove the extra codecs
```suggestion
            // When a side enables SO_LINGER and calls showdownOutput(...) to start TCP half-closure, we can not call doDeregister here
```
```suggestion
                // The shutdown function does not block regardless of the SO_LINGER setting on the socket,so we don't need to use
```
```suggestion
                // GlobalEventExecutor to execute the shutdown
```
```suggestion
```
```suggestion
```
```suggestion
```
```suggestion
```
```suggestion
```
```suggestion
```
```suggestion
```
Please don't use wildcard imports
@huibinliupush I am sure you will need to have the same change in the other transports as well (Epoll, KQueue).
> Please don't use wildcard imports

really sorry for this
> @huibinliupush I am sure you will need to have the same change in the other transports as well (Epoll, KQueue).

@normanmaurer OK, i'll check it right now

@normanmaurer 

i have checked `EpollSocketChannel` and `KQueueSocketChannel` again，they all call 
`((AbstractUnsafe) unsafe()).shutdownOutput(promise)` method to shutdownOutput In their parent class

Am I missing something?
ah ok... Sorry I thought they also offload :) 


Patch that resolved the issue:
diff --git a/transport/src/main/java/io/netty/channel/AbstractChannel.java b/transport/src/main/java/io/netty/channel/AbstractChannel.java
index 5aa2ab92db0..887a3e3c66e 100644
--- a/transport/src/main/java/io/netty/channel/AbstractChannel.java
+++ b/transport/src/main/java/io/netty/channel/AbstractChannel.java
@@ -650,38 +650,20 @@ private void shutdownOutput(final ChannelPromise promise, Throwable cause) {
             final Throwable shutdownCause = cause == null ?
                     new ChannelOutputShutdownException("Channel output shutdown") :
                     new ChannelOutputShutdownException("Channel output shutdown", cause);
-            Executor closeExecutor = prepareToClose();
-            if (closeExecutor != null) {
-                closeExecutor.execute(new Runnable() {
-                    @Override
-                    public void run() {
-                        try {
-                            // Execute the shutdown.
-                            doShutdownOutput();
-                            promise.setSuccess();
-                        } catch (Throwable err) {
-                            promise.setFailure(err);
-                        } finally {
-                            // Dispatch to the EventLoop
-                            eventLoop().execute(new Runnable() {
-                                @Override
-                                public void run() {
-                                    closeOutboundBufferForShutdown(pipeline, outboundBuffer, shutdownCause);
-                                }
-                            });
-                        }
-                    }
-                });
-            } else {
-                try {
-                    // Execute the shutdown.
-                    doShutdownOutput();
-                    promise.setSuccess();
-                } catch (Throwable err) {
-                    promise.setFailure(err);
-                } finally {
-                    closeOutboundBufferForShutdown(pipeline, outboundBuffer, shutdownCause);
-                }
+
+            // When a side enables SO_LINGER and calls showdownOutput(...) to start TCP half-closure
+            // we can not call doDeregister here because we should ensure this side in fin_wait2 state
+            // can still receive and process the data which is send by another side in the close_wait state。
+            // See https://github.com/netty/netty/issues/11981
+            try {
+                // The shutdown function does not block regardless of the SO_LINGER setting on the socket
+                // so we don't need to use GlobalEventExecutor to execute the shutdown
+                doShutdownOutput();
+                promise.setSuccess();
+            } catch (Throwable err) {
+                promise.setFailure(err);
+            } finally {
+                closeOutboundBufferForShutdown(pipeline, outboundBuffer, shutdownCause);
             }
         }
 


Test patch:
diff --git a/testsuite/src/main/java/io/netty/testsuite/transport/socket/SocketHalfClosedTest.java b/testsuite/src/main/java/io/netty/testsuite/transport/socket/SocketHalfClosedTest.java
index f2d6466dc6f..8e90217372a 100644
--- a/testsuite/src/main/java/io/netty/testsuite/transport/socket/SocketHalfClosedTest.java
+++ b/testsuite/src/main/java/io/netty/testsuite/transport/socket/SocketHalfClosedTest.java
@@ -19,6 +19,7 @@
 import io.netty.bootstrap.ServerBootstrap;
 import io.netty.buffer.ByteBuf;
 import io.netty.buffer.ByteBufAllocator;
+import io.netty.buffer.Unpooled;
 import io.netty.channel.Channel;
 import io.netty.channel.ChannelConfig;
 import io.netty.channel.ChannelFuture;
@@ -33,6 +34,8 @@
 import io.netty.channel.socket.ChannelInputShutdownReadComplete;
 import io.netty.channel.socket.ChannelOutputShutdownEvent;
 import io.netty.channel.socket.DuplexChannel;
+import io.netty.channel.socket.SocketChannel;
+import io.netty.util.ReferenceCountUtil;
 import io.netty.util.UncheckedBooleanSupplier;
 import io.netty.util.internal.PlatformDependent;
 import org.junit.jupiter.api.Test;
@@ -51,6 +54,81 @@
 import static org.junit.jupiter.api.Assumptions.assumeFalse;
 
 public class SocketHalfClosedTest extends AbstractSocketTest {
+
+    @Test
+    @Timeout(value = 5000, unit = MILLISECONDS)
+    public void testHalfClosureReceiveDataOnFinalWait2StateWhenSoLingerSet(TestInfo testInfo) throws Throwable {
+        run(testInfo, new Runner<ServerBootstrap, Bootstrap>() {
+            @Override
+            public void run(ServerBootstrap serverBootstrap, Bootstrap bootstrap) throws Throwable {
+                testHalfClosureReceiveDataOnFinalWait2StateWhenSoLingerSet(serverBootstrap, bootstrap);
+            }
+        });
+    }
+
+    private void testHalfClosureReceiveDataOnFinalWait2StateWhenSoLingerSet(ServerBootstrap sb, Bootstrap cb)
+            throws Throwable {
+        Channel serverChannel = null;
+        Channel clientChannel = null;
+
+        final CountDownLatch waitHalfClosureDone = new CountDownLatch(1);
+        try {
+            sb.childOption(ChannelOption.SO_LINGER, 1)
+              .childHandler(new ChannelInitializer<Channel>() {
+
+                  @Override
+                  protected void initChannel(Channel ch) throws Exception {
+                      ch.pipeline().addLast(new ChannelInboundHandlerAdapter() {
+
+                            @Override
+                            public void channelActive(final ChannelHandlerContext ctx) {
+                                SocketChannel channel = (SocketChannel) ctx.channel();
+                                channel.shutdownOutput();
+                            }
+
+                            @Override
+                            public void channelRead(ChannelHandlerContext ctx, Object msg) {
+                                ReferenceCountUtil.release(msg);
+                                waitHalfClosureDone.countDown();
+                            }
+                        });
+                  }
+              });
+
+            cb.option(ChannelOption.ALLOW_HALF_CLOSURE, true)
+              .handler(new ChannelInitializer<Channel>() {
+                  @Override
+                  protected void initChannel(Channel ch) throws Exception {
+                      ch.pipeline().addLast(new ChannelInboundHandlerAdapter() {
+
+                            @Override
+                            public void userEventTriggered(ChannelHandlerContext ctx, Object evt) {
+                                if (ChannelInputShutdownEvent.INSTANCE == evt) {
+                                    ctx.writeAndFlush(ctx.alloc().buffer().writeZero(16));
+                                }
+
+                                if (ChannelInputShutdownReadComplete.INSTANCE == evt) {
+                                    ctx.close();
+                                }
+                            }
+                        });
+                  }
+              });
+
+            serverChannel = sb.bind().sync().channel();
+            clientChannel = cb.connect(serverChannel.localAddress()).sync().channel();
+            waitHalfClosureDone.await();
+        } finally {
+            if (clientChannel != null) {
+                clientChannel.close().sync();
+            }
+
+            if (serverChannel != null) {
+                serverChannel.close().sync();
+            }
+        }
+    }
+
     @Test
     @Timeout(value = 10000, unit = MILLISECONDS)
     public void testHalfClosureOnlyOneEventWhenAutoRead(TestInfo testInfo) throws Throwable {