Fix intermittent out-of-order chunk helidon-io#7407

microprofile/server/src/test/java/io/helidon/microprofile/server/AutoFlushTest.java

@@ -21,6 +21,9 @@
 import java.io.FileOutputStream;
 import java.io.IOException;
 import java.io.InputStream;
+import java.security.MessageDigest;
+import java.security.NoSuchAlgorithmException;
+import java.util.Arrays;
 
 import io.helidon.common.http.Http;
 import io.helidon.microprofile.tests.junit5.AddBean;
@@ -32,7 +35,7 @@
 import jakarta.ws.rs.core.Response;
 import jakarta.ws.rs.core.StreamingOutput;
 import org.junit.jupiter.api.BeforeAll;
-import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.RepeatedTest;
 
 import static org.hamcrest.CoreMatchers.is;
 import static org.hamcrest.MatcherAssert.assertThat;
@@ -42,7 +45,9 @@
 class AutoFlushTest {
 
     private static File file;
-    private static final int FILE_SIZE = 6000000;
+    private static byte[] sha256;
+    private static final int FILE_SIZE = 40 * 1024 * 1024;
+    private static final int BUFFER_SIZE = 8 * 1024;
 
     @Inject
     private WebTarget target;
@@ -61,25 +66,48 @@ public Response getDefaultMessage() {
     }
 
     @BeforeAll
-    static void createTempFile() throws IOException {
-        file = File.createTempFile("zero", ".bin");
+    static void createTempFile() throws IOException, NoSuchAlgorithmException {
+        file = File.createTempFile("file", ".bin", new File("/tmp"));
         file.deleteOnExit();
-        byte[] zero = new byte[10 * 1000];
+        MessageDigest messageDigest = MessageDigest.getInstance("SHA-256");
         try (FileOutputStream fos = new FileOutputStream(file)) {
-            for (int i = 0; i < FILE_SIZE / zero.length; i++) {
-                fos.write(zero);
+            for (int i = 0; i < FILE_SIZE / BUFFER_SIZE; i++) {
+                byte[] bytes = dataBuffer(BUFFER_SIZE, (byte) (i % 10 + '0'));
+                fos.write(bytes);
+                messageDigest.update(bytes);
             }
         }
+        sha256 = messageDigest.digest();
     }
 
-    @Test
-    void testAutoFlush() {
+    static byte[] dataBuffer(int size, byte b) {
+        byte[] bytes = new byte[size];
+        Arrays.fill(bytes, b);
+        bytes[size - 1] = '\n';     // to open file in editor
+        return bytes;
+    }
+
+    @RepeatedTest(10)
+    void testAutoFlush() throws NoSuchAlgorithmException, IOException {
         try (Response resp = target.path("auto-flush")
                 .request()
                 .get()) {
             assertThat(resp.getStatus(), is(200));
             assertThat(resp.getHeaderString(Http.Header.CONTENT_LENGTH), is(String.valueOf(FILE_SIZE)));
-            assertThat(resp.readEntity(byte[].class).length, is(FILE_SIZE));
+
+            byte[] file = resp.readEntity(byte[].class);
+            File downloaded = File.createTempFile("downloaded", ".bin", new File("/tmp"));
+            try (FileOutputStream fos = new FileOutputStream(downloaded)) {
+                fos.write(file);
+            }
+            assertThat(file.length, is(FILE_SIZE));
+
+            MessageDigest messageDigest = MessageDigest.getInstance("SHA-256");
+            for (int i = 0; i < FILE_SIZE / BUFFER_SIZE; i++) {
+                messageDigest.update(file, i * BUFFER_SIZE, BUFFER_SIZE);
+            }
+            byte[] newSha256 = messageDigest.digest();
+            assertThat(sha256, is(newSha256));
         }
     }
 }

webserver/webserver/src/main/java/io/helidon/webserver/NettyChannel.java

@@ -75,14 +75,11 @@ void read() {
      * Request to flush all pending messages via this ChannelOutboundInvoker from Netty's event loop thread.
      */
     void flush() {
-        writeFuture = writeFuture.thenApply(f -> {
-            if (channel.eventLoop().inEventLoop()) {
-                channel.flush();
-            } else {
-                channel.eventLoop().execute(channel::flush);
-            }
-            return f;
-        });
+        if (channel.eventLoop().inEventLoop()) {
+            channel.flush();
+        } else {
+            channel.eventLoop().execute(channel::flush);
+        }
     }
 
     /**

webserver/webserver/src/test/java/io/helidon/webserver/WaterMarkedBackpressureIT.java

@@ -18,6 +18,9 @@
 
 import java.nio.ByteBuffer;
 import java.time.Duration;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.CompletionException;
 import java.util.concurrent.ExecutorService;
@@ -39,6 +42,7 @@
 import static io.helidon.webserver.BackpressureStrategy.AUTO_FLUSH;
 import static io.helidon.webserver.BackpressureStrategy.LINEAR;
 import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.contains;
 import static org.hamcrest.Matchers.greaterThan;
 import static org.hamcrest.Matchers.is;
 
@@ -131,15 +135,15 @@ void linear() {
         }
     }
 
-    @Test
+    @Test(invocationCount = 10)
     void autoFlush() {
         AtomicLong receivedSize = new AtomicLong(0);
 
         WebServer webServer = null;
         try {
             webServer = WebServer.builder()
                     .host("localhost")
-                    .backpressureBufferSize(500)
+                    .backpressureBufferSize(200)
                     .backpressureStrategy(AUTO_FLUSH)
                     .routing(r -> r.get("/", (req, res) -> {
                         res.send(Multi.range(0, 150)
@@ -157,6 +161,8 @@ void autoFlush() {
                     .start()
                     .await(TIMEOUT);
 
+            List<String> receivedData = new ArrayList<>(100);
+
             WebClient.builder()
                     .baseUri("http://localhost:" + webServer.port())
                     .build()
@@ -169,6 +175,7 @@ void autoFlush() {
                         receivedSize.addAndGet(bytes.length);
                         String data = new String(bytes);
                         chunk.release();
+                        receivedData.add(data);
                         return !data.equals("00100");
                     })
                     .ignoreElements()
@@ -178,6 +185,8 @@ void autoFlush() {
                     })
                     .await(TIMEOUT);
 
+            assertThat(receivedData, contains(Multi.range(0, 101)
+                    .map(l -> String.format("%05d", l)).collectList().await().toArray(String[]::new)));
 
             // Stochastic test as watermarking depends on Netty's flush callbacks
             assertThat(receivedSize.get(), greaterThan(499L));