Repository: netty/netty (Java)

Issue:
FlowControlHandler is passing read events when auto-reading is turned off
### Expected behavior
Not passing read events onto the pipeline (calling ctx.read() which results in reading from the client socket) when the queue is not empty and auto-reading is turned off for the channel. I would suggest that the expected behavior should look something like the following piece of code:
```
if (dequeue(ctx, 1) == 0) {
	shouldConsume = true;
	ctx.read();
} else if (ctx.channel().config().isAutoRead()) {
	ctx.read();
}
```

### Actual behavior
Read events are passed onto the pipeline in every case regardless of the auto-reading configuration of the channel.

### Steps to reproduce
1. Let's suppose we have a channel with a FlowControlHandler in its pipeline.
2. Turn auto-reading off for the channel.
3. Manually call ctx.read()
4. For every ctx.read() call, the FlowControlHandler's read method will pass the read event onto the pipeline which will result in reading from the client socket.
5. If the messages, which we read are not processed fast enough according to our custom logic, all of this can lead to a significant amount of data being stored in memory which leads to out-of-memory errors.

### Minimal yet complete reproducer code (or URL to code)
n/a

### Netty version
4.1.85.Final and above

### JVM version (e.g. `java -version`)
any

### OS version (e.g. `uname -a`)
any



Hints:
@ivanangelov can you do a PR with the suggested changes and a unit test ? We love contributions 

Patch:
diff --git a/handler/src/main/java/io/netty/handler/flow/FlowControlHandler.java b/handler/src/main/java/io/netty/handler/flow/FlowControlHandler.java
--- a/handler/src/main/java/io/netty/handler/flow/FlowControlHandler.java
+++ b/handler/src/main/java/io/netty/handler/flow/FlowControlHandler.java
@@ -141,8 +141,10 @@ public void read(ChannelHandlerContext ctx) throws Exception {
             // messages from upstream and once one arrives it need to be
             // relayed to downstream to keep the flow going.
             shouldConsume = true;
+            ctx.read();
+        } else if (config.isAutoRead()) {
+            ctx.read();
         }
-        ctx.read();
     }
 
     @Override


Test patch:
diff --git a/handler/src/test/java/io/netty/handler/flow/FlowControlHandlerTest.java b/handler/src/test/java/io/netty/handler/flow/FlowControlHandlerTest.java
--- a/handler/src/test/java/io/netty/handler/flow/FlowControlHandlerTest.java
+++ b/handler/src/test/java/io/netty/handler/flow/FlowControlHandlerTest.java
@@ -408,6 +408,78 @@ public Boolean call() {
         }
     }
 
+    /**
+     * The {@link FlowControlHandler} will not pass read events onto the
+     * pipeline when the user is calling {@code read()} on their own if the
+     * queue is not empty and auto-reading is turned off for the channel.
+     */
+    @Test
+    public void testFlowAutoReadOffAndQueueNonEmpty() throws Exception {
+        final Exchanger<Channel> peerRef = new Exchanger<Channel>();
+        final CountDownLatch msgRcvLatch1 = new CountDownLatch(1);
+        final CountDownLatch msgRcvLatch2 = new CountDownLatch(2);
+        final CountDownLatch msgRcvLatch3 = new CountDownLatch(3);
+
+        ChannelInboundHandlerAdapter handler = new ChannelDuplexHandler() {
+            @Override
+            public void channelActive(ChannelHandlerContext ctx) throws Exception {
+                ctx.fireChannelActive();
+                peerRef.exchange(ctx.channel(), 1L, SECONDS);
+            }
+
+            @Override
+            public void channelRead(ChannelHandlerContext ctx, Object msg) {
+                msgRcvLatch1.countDown();
+                msgRcvLatch2.countDown();
+                msgRcvLatch3.countDown();
+            }
+        };
+
+        final FlowControlHandler flow = new FlowControlHandler();
+        Channel server = newServer(false, flow, handler);
+        Channel client = newClient(server.localAddress());
+        try {
+            // The client connection on the server side
+            Channel peer = peerRef.exchange(null, 1L, SECONDS);
+
+            // Write the first message
+            client.writeAndFlush(newOneMessage())
+                .syncUninterruptibly();
+
+            // channelRead(1)
+            peer.read();
+            assertTrue(msgRcvLatch1.await(1L, SECONDS));
+            assertFalse(peer.eventLoop().submit(new Callable<Boolean>() {
+                @Override
+                public Boolean call() {
+                    return flow.isQueueEmpty();
+                }
+            }).get());
+
+            // Write the second message
+            client.writeAndFlush(newOneMessage())
+                .syncUninterruptibly();
+
+            // channelRead(2)
+            peer.read();
+            assertTrue(msgRcvLatch2.await(1L, SECONDS));
+
+            // channelRead(3)
+            peer.read();
+            assertTrue(msgRcvLatch3.await(1L, SECONDS));
+
+            assertTrue(peer.eventLoop().submit(new Callable<Boolean>() {
+                @Override
+                public Boolean call() {
+                    return flow.isQueueEmpty();
+                }
+            }).get());
+        } finally {
+            client.close();
+            server.close();
+        }
+    }
+
     @Test
     public void testReentranceNotCausesNPE() throws Throwable {
         final Exchanger<Channel> peerRef = new Exchanger<Channel>();