non-reentrant reads

Author: mhl-b

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

@@ -58,17 +58,21 @@ public void addTracingHandler(ChunkHandler chunkHandler) {
         tracingHandlers.add(chunkHandler);
     }
 
+    private void scheduleRead() {
+        channel.eventLoop().execute(channel::read);
+    }
+
     @Override
     public void next() {
         assert handler != null : "handler must be set before requesting next chunk";
         requestContext = threadContext.newStoredContext();
-        channel.eventLoop().execute(channel::read);
+        scheduleRead();
     }
 
     public void handleNettyContent(HttpContent httpContent) {
         if (closing) {
             httpContent.release();
-            channel.eventLoop().execute(channel::read);
+            scheduleRead();
         } else {
             try (var ignored = threadContext.restoreExistingContext(requestContext)) {
                 var isLast = httpContent instanceof LastHttpContent;
@@ -78,7 +82,7 @@ public void handleNettyContent(HttpContent httpContent) {
                 }
                 handler.onNext(buf, isLast);
                 if (isLast) {
-                    channel.read();
+                    scheduleRead();
                     channel.closeFuture().removeListener(closeListener);
                 }
             }
@@ -104,6 +108,6 @@ private void doClose() {
                 handler.close();
             }
         }
-        channel.eventLoop().execute(channel::read);
+        scheduleRead();
     }
 }