Fixing Flaky ReactorNetty4StreamingStressIT test case (another attempt)

Signed-off-by: Andriy Redko <drreta@gmail.com>

Author: reta

client/rest/src/main/java/org/opensearch/client/RestClient.java

@@ -312,6 +312,14 @@ public boolean isRunning() {
         return client.getStatus() == IOReactorStatus.ACTIVE;
     }
 
+    /**
+     * check client shutdown (closed) status
+     * @return client shutdown (closed) status
+     */
+    public boolean isClosed() {
+        return client.getStatus() == IOReactorStatus.SHUT_DOWN;
+    }
+
     /**
      * Sends a streaming request to the OpenSearch cluster that the client points to and returns streaming response. <strong>This is an experimental API</strong>.
      * @param request streaming request

plugins/transport-reactor-netty4/src/javaRestTest/java/org/opensearch/rest/ReactorNetty4StreamingStressIT.java

@@ -22,6 +22,7 @@
 import java.nio.ByteBuffer;
 import java.nio.charset.StandardCharsets;
 import java.time.Duration;
+import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.stream.Stream;
 
@@ -44,7 +45,6 @@ public void tearDown() throws Exception {
         super.tearDown();
     }
 
-    @AwaitsFix(bugUrl = "https://github.com/opensearch-project/OpenSearch/issues/15840")
     public void testCloseClientStreamingRequest() throws Exception {
         final VirtualTimeScheduler scheduler = VirtualTimeScheduler.create(true);
 
@@ -72,12 +72,25 @@ public void testCloseClientStreamingRequest() throws Exception {
             .then(() -> {
                 try {
                     client().close();
+
+                    final long await = TimeUnit.SECONDS.toNanos(5);
+                    final long started = System.nanoTime();
+                    while (client().isClosed() == false) {
+                        Thread.onSpinWait();
+                        if ((System.nanoTime() - started) > await) {
+                            throw new InterruptedIOException("The client is still shutting down");
+                        }
+                    }
                 } catch (final IOException ex) {
                     throw new UncheckedIOException(ex);
                 }
             })
             .then(() -> scheduler.advanceTimeBy(delay))
-            .expectErrorMatches(t -> t instanceof InterruptedIOException || t instanceof ConnectionClosedException)
+            .then(() -> scheduler.advanceTimeBy(delay))
+            .then(() -> scheduler.advanceTimeBy(delay))
+            .expectErrorMatches(
+                t -> t instanceof InterruptedIOException || t instanceof ConnectionClosedException || t instanceof UncheckedIOException
+            )
             .verify(Duration.ofSeconds(10));
     }
 }