Fixed multiple race conditions in shared buffer tests

Author: ok2c

httpcore5/src/test/java/org/apache/hc/core5/http/nio/support/classic/TestSharedOutputBuffer.java

@@ -37,14 +37,14 @@
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 import java.util.concurrent.Future;
-import java.util.concurrent.locks.Condition;
 import java.util.concurrent.locks.ReentrantLock;
 
 import org.apache.hc.core5.http.Header;
 import org.apache.hc.core5.http.WritableByteChannelMock;
 import org.apache.hc.core5.http.nio.DataStreamChannel;
 import org.apache.hc.core5.util.Timeout;
 import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.RepeatedTest;
 import org.junit.jupiter.api.Test;
 import org.mockito.Mockito;
 
@@ -57,12 +57,10 @@ static class DataStreamChannelMock implements DataStreamChannel {
         private final WritableByteChannelMock channel;
 
         private final ReentrantLock lock;
-        private final Condition condition;
 
         DataStreamChannelMock(final WritableByteChannelMock channel) {
             this.channel = channel;
             this.lock = new ReentrantLock();
-            this.condition = lock.newCondition();
         }
 
         @Override
@@ -77,20 +75,13 @@ public int write(final ByteBuffer src) throws IOException {
 
         @Override
         public void requestOutput() {
-            lock.lock();
-            try {
-                condition.signalAll();
-            } finally {
-                lock.unlock();
-            }
         }
 
         @Override
         public void endStream(final List<? extends Header> trailers) throws IOException {
             lock.lock();
             try {
                 channel.close();
-                condition.signalAll();
             } finally {
                 lock.unlock();
             }
@@ -101,15 +92,6 @@ public void endStream() throws IOException {
             endStream(null);
         }
 
-        public void awaitOutputRequest() throws InterruptedException {
-            lock.lock();
-            try {
-                condition.await();
-            } finally {
-                lock.unlock();
-            }
-        }
-
     }
 
     @Test
@@ -164,7 +146,7 @@ void testFlush() throws Exception {
         Assertions.assertEquals(30, outputBuffer.capacity());
     }
 
-    @Test
+    @RepeatedTest(20)
     void testMultithreadingWriteStream() throws Exception {
 
         final Charset charset = StandardCharsets.US_ASCII;
@@ -174,36 +156,37 @@ void testMultithreadingWriteStream() throws Exception {
         final DataStreamChannelMock dataStreamChannel = new DataStreamChannelMock(channel);
 
         final ExecutorService executorService = Executors.newFixedThreadPool(2);
-        final Future<Boolean> task1 = executorService.submit(() -> {
-            final byte[] tmp = "1234567890".getBytes(charset);
-            outputBuffer.write(tmp, 0, tmp.length);
-            outputBuffer.write(tmp, 0, tmp.length);
-            outputBuffer.write('1');
-            outputBuffer.write('2');
-            outputBuffer.write(tmp, 0, tmp.length);
-            outputBuffer.write(tmp, 0, tmp.length);
-            outputBuffer.write(tmp, 0, tmp.length);
-            outputBuffer.writeCompleted();
-            outputBuffer.writeCompleted();
-            return Boolean.TRUE;
-        });
-        final Future<Boolean> task2 = executorService.submit(() -> {
-            for (;;) {
-                outputBuffer.flush(dataStreamChannel);
-                if (outputBuffer.isEndStream()) {
-                    break;
-                }
-                if (!outputBuffer.hasData()) {
-                    dataStreamChannel.awaitOutputRequest();
+        try {
+            final Future<Boolean> task1 = executorService.submit(() -> {
+                final byte[] tmp = "1234567890".getBytes(charset);
+                outputBuffer.write(tmp, 0, tmp.length);
+                outputBuffer.write(tmp, 0, tmp.length);
+                outputBuffer.write('1');
+                outputBuffer.write('2');
+                outputBuffer.write(tmp, 0, tmp.length);
+                outputBuffer.write(tmp, 0, tmp.length);
+                outputBuffer.write(tmp, 0, tmp.length);
+                outputBuffer.writeCompleted();
+                outputBuffer.writeCompleted();
+                return Boolean.TRUE;
+            });
+            final Future<Boolean> task2 = executorService.submit(() -> {
+                for (;;) {
+                    outputBuffer.flush(dataStreamChannel);
+                    if (outputBuffer.isEndStream()) {
+                        break;
+                    }
                 }
-            }
-            return Boolean.TRUE;
-        });
+                return Boolean.TRUE;
+            });
 
-        Assertions.assertEquals(Boolean.TRUE, task1.get(TIMEOUT.getDuration(), TIMEOUT.getTimeUnit()));
-        Assertions.assertEquals(Boolean.TRUE, task2.get(TIMEOUT.getDuration(), TIMEOUT.getTimeUnit()));
+            Assertions.assertEquals(Boolean.TRUE, task1.get(TIMEOUT.getDuration(), TIMEOUT.getTimeUnit()));
+            Assertions.assertEquals(Boolean.TRUE, task2.get(TIMEOUT.getDuration(), TIMEOUT.getTimeUnit()));
 
-        Assertions.assertEquals("1234567890123456789012123456789012345678901234567890", new String(channel.toByteArray(), charset));
+            Assertions.assertEquals("1234567890123456789012123456789012345678901234567890", new String(channel.toByteArray(), charset));
+        } finally {
+            executorService.shutdownNow();
+        }
     }
 
     @Test