Repository: netty/netty
Issue: OutOfDirectMemoryError causes CPU 100% and fd is full

### Expected behavior
Release connection

### Actual behavior
Always occupy the connection

### Steps to reproduce
see [# 10424](https://github.com/netty/netty/issues/10424) OOM Exception , OOM  in netty 4.1.51.Final Fix，but OOM problem  caused this problem.  Should need to be fixed, because the next OOM will appear again .

### Minimal yet complete reproducer code (or URL to code)

Gave me some thoughts  in the code .

in read event:
```java
@Override
        public final void read() {
            final ChannelConfig config = config();
            if (shouldBreakReadReady(config)) {
                clearReadPending();
                return;
            }
            final ChannelPipeline pipeline = pipeline();
            final ByteBufAllocator allocator = config.getAllocator();
            final RecvByteBufAllocator.Handle allocHandle = recvBufAllocHandle();
            allocHandle.reset(config);

            ByteBuf byteBuf = null;
            boolean close = false;
            try {
                do {

                    // * If an oom exception occurs in its place,  oom will also be here  *
                    byteBuf = allocHandle.allocate(allocator);

                    allocHandle.lastBytesRead(doReadBytes(byteBuf));
                    if (allocHandle.lastBytesRead() <= 0) {
                        // nothing was read. release the buffer.
                        byteBuf.release();
                        byteBuf = null;
                        close = allocHandle.lastBytesRead() < 0;
                        if (close) {
                            // There is nothing left to read as we received an EOF.
                            readPending = false;
                        }
                        break;
                    }

                    allocHandle.incMessagesRead(1);
                    readPending = false;
                    pipeline.fireChannelRead(byteBuf);
                    byteBuf = null;
                } while (allocHandle.continueReading());

                allocHandle.readComplete();
                pipeline.fireChannelReadComplete();

                if (close) {
                    closeOnRead(pipeline);
                }
            } catch (Throwable t) {
                //  there catch oom exceptin and handler it 
                handleReadException(pipeline, byteBuf, t, close, allocHandle);
            } finally {
                .....
            }
        }
    }
```

handleReadException method:
```java

  private void handleReadException(ChannelPipeline pipeline, ByteBuf byteBuf, Throwable cause, boolean close,
                RecvByteBufAllocator.Handle allocHandle) {
            if (byteBuf != null) {
                if (byteBuf.isReadable()) {
                    readPending = false;
                    pipeline.fireChannelRead(byteBuf);
                } else {
                    byteBuf.release();
                }
            }
            allocHandle.readComplete();
            pipeline.fireChannelReadComplete();
            pipeline.fireExceptionCaught(cause);

           // If it is oom , clase value is false, no change, 
           //  and cause not instanceof IOException,  because instanceof OutOfMemoryError ,
           // So event is not fire , event readPending always , If http server, request block always too.

           // In high concurrency scenarios , this also explains why the socket is full and cpu full too
           //  If oom maybe set close = ture, or condition `cause instanceof OutOfMemoryError`
            if (close || cause instanceof IOException) {
                closeOnRead(pipeline);
            }
        }


```

```shell
[root@app ~]# ss -s
Total: 61569 (kernel 61731)
TCP:   62328 (estab 49757, closed 9067, orphaned 3, synrecv 0, timewait 235/0), ports 411

Transport Total     IP        IPv6
*	  61731     -         -        
RAW	  0         0         0        
UDP	  1         1         0        
TCP	  53261     6         53255    
INET	  53262     7         53255    
FRAG	  0         0         0    
```

### Netty version
4.1.50.Final


### JVM version (e.g. `java -version`)
java version "1.8.0_221"
Java(TM) SE Runtime Environment (build 1.8.0_221-b11)
Java HotSpot(TM) 64-Bit Server VM (build 25.221-b11, mixed mode)


### OS version (e.g. `uname -a`)
Windows
Linux


Issue comments:
I don't think netty was ever intended to properly recover from OutOfMemoryError...it would be a nice addition but hard to recover from errors in every random place.
> I don't think netty was ever intended to properly recover from OutOfMemoryError...it would be a nice addition but hard to recover from errors in every random place.

yeah,  recovery from Outofmemoryerror is difficult.

I've only found out that `AbstractNioByteChannel.java#read` OutOfMemoryError causes the pipeline read event to be blocked. If many requests are in the request, many scokets will be occupied(socket). 
```shell
[root@app ~]# ss -s
Total: 61569 (kernel 61731)
TCP:   62328 (estab 49757, closed 9067, orphaned 3, synrecv 0, timewait 235/0), ports 411

Transport Total     IP        IPv6
*	  61731     -         -        
RAW	  0         0         0        
UDP	  1         1         0        
TCP	  53261     6         53255    
INET	  53262     7         53255    
FRAG	  0         0         0    
```

In each channel read implementation, netty creates a new receive buffer to read all inbound data when it enters the read event, in this case if OutOfMemoryError is supposed to release the pipeline read event. Otherwise,  it will be blocked for a certain period of time, causing the server to monitor the alarm. 

If we find out where to use `RecvByteBufAllocator.java#Handle#allocate`,  and then catch in the catch to add out of the OutOfMemoryError judgment, and then call the release pipeline read event, will it solve the problem?

I should probably test the 'EpollStreamChannel' KQueueStreamChannel environment to see if it will be the same, but I haven't tried to do it.   

Pull request review comments:

Patch that resolved the issue:
diff --git a/transport-native-epoll/src/main/java/io/netty/channel/epoll/AbstractEpollStreamChannel.java b/transport-native-epoll/src/main/java/io/netty/channel/epoll/AbstractEpollStreamChannel.java
index 95e31b380f7..8a5b79d4938 100644
--- a/transport-native-epoll/src/main/java/io/netty/channel/epoll/AbstractEpollStreamChannel.java
+++ b/transport-native-epoll/src/main/java/io/netty/channel/epoll/AbstractEpollStreamChannel.java
@@ -726,7 +726,10 @@ private void handleReadException(ChannelPipeline pipeline, ByteBuf byteBuf, Thro
             allocHandle.readComplete();
             pipeline.fireChannelReadComplete();
             pipeline.fireExceptionCaught(cause);
-            if (close || cause instanceof IOException) {
+
+            // If oom will close the read event, release connection.
+            // See https://github.com/netty/netty/issues/10434
+            if (close || cause instanceof OutOfMemoryError || cause instanceof IOException) {
                 shutdownInput(false);
             }
         }
diff --git a/transport-native-kqueue/src/main/java/io/netty/channel/kqueue/AbstractKQueueStreamChannel.java b/transport-native-kqueue/src/main/java/io/netty/channel/kqueue/AbstractKQueueStreamChannel.java
index 4b7f4e06e5b..a412d6ac82b 100644
--- a/transport-native-kqueue/src/main/java/io/netty/channel/kqueue/AbstractKQueueStreamChannel.java
+++ b/transport-native-kqueue/src/main/java/io/netty/channel/kqueue/AbstractKQueueStreamChannel.java
@@ -587,7 +587,10 @@ private void handleReadException(ChannelPipeline pipeline, ByteBuf byteBuf, Thro
                 allocHandle.readComplete();
                 pipeline.fireChannelReadComplete();
                 pipeline.fireExceptionCaught(cause);
-                if (close || cause instanceof IOException) {
+
+                // If oom will close the read event, release connection.
+                // See https://github.com/netty/netty/issues/10434
+                if (close || cause instanceof OutOfMemoryError || cause instanceof IOException) {
                     shutdownInput(false);
                 }
             }
diff --git a/transport/src/main/java/io/netty/channel/nio/AbstractNioByteChannel.java b/transport/src/main/java/io/netty/channel/nio/AbstractNioByteChannel.java
index db87b100b0c..d0b6acd880b 100644
--- a/transport/src/main/java/io/netty/channel/nio/AbstractNioByteChannel.java
+++ b/transport/src/main/java/io/netty/channel/nio/AbstractNioByteChannel.java
@@ -123,7 +123,10 @@ private void handleReadException(ChannelPipeline pipeline, ByteBuf byteBuf, Thro
             allocHandle.readComplete();
             pipeline.fireChannelReadComplete();
             pipeline.fireExceptionCaught(cause);
-            if (close || cause instanceof IOException) {
+
+            // If oom will close the read event, release connection.
+            // See https://github.com/netty/netty/issues/10434
+            if (close || cause instanceof OutOfMemoryError || cause instanceof IOException) {
                 closeOnRead(pipeline);
             }
         }
diff --git a/transport/src/main/java/io/netty/channel/oio/AbstractOioByteChannel.java b/transport/src/main/java/io/netty/channel/oio/AbstractOioByteChannel.java
index 54ea0deadd8..87c24b7dce8 100644
--- a/transport/src/main/java/io/netty/channel/oio/AbstractOioByteChannel.java
+++ b/transport/src/main/java/io/netty/channel/oio/AbstractOioByteChannel.java
@@ -93,7 +93,10 @@ private void handleReadException(ChannelPipeline pipeline, ByteBuf byteBuf, Thro
         allocHandle.readComplete();
         pipeline.fireChannelReadComplete();
         pipeline.fireExceptionCaught(cause);
-        if (close || cause instanceof IOException) {
+
+        // If oom will close the read event, release connection.
+        // See https://github.com/netty/netty/issues/10434
+        if (close || cause instanceof OutOfMemoryError || cause instanceof IOException) {
             closeOnRead(pipeline);
         }
     }


Test patch: