Remove needless flush call from Netty4 write handlers (elastic#89647)

These inline flush calls will never make the channel writable since
they will just get queued on the executor and run a redundant flush
execution later.
We can simplify the code here and save some cycles by simply giving up
and forwarding the flush once the channel stops being writable and
then have the logic pick up again on the next writability changed
call.

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

@@ -225,23 +225,14 @@ private boolean doFlush(ChannelHandlerContext ctx) {
             failQueuedWrites();
             return false;
         }
-        boolean needsFlush = true;
         while (channel.isWritable()) {
             final WriteOperation currentWrite = queuedWrites.poll();
             if (currentWrite == null) {
                 break;
             }
             ctx.write(currentWrite.msg, currentWrite.promise);
-            needsFlush = true;
-            if (channel.isWritable() == false) {
-                // try flushing to make channel writable again, loop will only continue if channel becomes writable again
-                ctx.flush();
-                needsFlush = false;
-            }
-        }
-        if (needsFlush) {
-            ctx.flush();
         }
+        ctx.flush();
         if (channel.isActive() == false) {
             failQueuedWrites();
         }

modules/transport-netty4/src/main/java/org/elasticsearch/transport/netty4/Netty4WriteThrottlingHandler.java

@@ -116,7 +116,6 @@ private boolean doFlush(ChannelHandlerContext ctx) {
             failQueuedWrites();
             return false;
         }
-        boolean needsFlush = true;
         while (channel.isWritable()) {
             if (currentWrite == null) {
                 currentWrite = queuedWrites.poll();
@@ -137,22 +136,14 @@ private boolean doFlush(ChannelHandlerContext ctx) {
                 writeBuffer = write.buf;
             }
             final ChannelFuture writeFuture = ctx.write(writeBuffer);
-            needsFlush = true;
             if (sliced == false) {
                 currentWrite = null;
                 writeFuture.addListener(forwardResultListener(ctx, write.promise));
             } else {
                 writeFuture.addListener(forwardFailureListener(ctx, write.promise));
             }
-            if (channel.isWritable() == false) {
-                // try flushing to make channel writable again, loop will only continue if channel becomes writable again
-                ctx.flush();
-                needsFlush = false;
-            }
-        }
-        if (needsFlush) {
-            ctx.flush();
         }
+        ctx.flush();
         if (channel.isActive() == false) {
             failQueuedWrites();
         }