Unmute channel when flush last http stream chunk (#113222) (#115862)

Co-authored-by: Mikhail Berezovskiy <[email protected]>

Author: Tim-Brooks

modules/transport-netty4/src/main/java/org/elasticsearch/http/netty4/Netty4HttpRequestBodyStream.java

@@ -87,10 +87,6 @@ public void handleNettyContent(HttpContent httpContent) {
                 send();
             }
         }
-        if (hasLast) {
-            channel.config().setAutoRead(true);
-            channel.closeFuture().removeListener(closeListener);
-        }
     }
 
     // adds chunk to current buffer, will allocate composite buffer when need to hold more than 1 chunk
@@ -133,6 +129,10 @@ private void send() {
             tracer.onNext(bytesRef, hasLast);
         }
         handler.onNext(bytesRef, hasLast);
+        if (hasLast) {
+            channel.config().setAutoRead(true);
+            channel.closeFuture().removeListener(closeListener);
+        }
     }
 
     @Override

modules/transport-netty4/src/test/java/org/elasticsearch/http/netty4/Netty4HttpRequestBodyStreamTests.java

@@ -82,6 +82,15 @@ public void testFlushAllReceivedChunks() {
         assertEquals(chunkSize * totalChunks, totalBytes.get());
     }
 
+    // ensures that channel.setAutoRead(true) only when we flush last chunk
+    public void testSetAutoReadOnLastFlush() {
+        channel.writeInbound(randomLastContent(10));
+        assertFalse("should not auto-read on last content reception", channel.config().isAutoRead());
+        stream.next();
+        channel.runPendingTasks();
+        assertTrue("should set auto-read once last content is flushed", channel.config().isAutoRead());
+    }
+
     // ensures that we read from channel when no current chunks available
     // and pass next chunk downstream without holding
     public void testReadFromChannel() {