Merge branch 'main' into fix-regression-getStreamTimeout

Author: scottf

src/test/java/io/nats/client/ConnectTests.java

@@ -18,7 +18,6 @@
 import io.nats.client.api.ServerInfo;
 import io.nats.client.impl.ListenerForTesting;
 import io.nats.client.impl.SimulateSocketDataPortException;
-import io.nats.client.utils.TestBase;
 import org.junit.jupiter.api.Test;
 
 import java.io.IOException;
@@ -30,7 +29,6 @@
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
-import java.util.concurrent.atomic.AtomicLong;
 
 import static io.nats.client.utils.TestBase.*;
 import static org.junit.jupiter.api.Assertions.*;
@@ -643,42 +641,4 @@ void testConnectWithHappyEyeballsShortCircuitCoverage() throws Exception {
         }
     }
 
-    @Test
-    void testConnectPendingCountCoverage() throws Exception {
-        TestBase.runInJsServer(nc -> {
-            AtomicLong outgoingPendingMessageCount = new AtomicLong();
-            AtomicLong outgoingPendingBytes = new AtomicLong();
-
-            AtomicBoolean tKeepGoing = new AtomicBoolean(true);
-            Thread t = new Thread(() -> {
-                while (tKeepGoing.get()) {
-                    outgoingPendingMessageCount.set(Math.max(outgoingPendingMessageCount.get(), nc.outgoingPendingMessageCount()));
-                    outgoingPendingBytes.set(Math.max(outgoingPendingBytes.get(), nc.outgoingPendingBytes()));
-                    try {
-                        Thread.sleep(10);
-                    }
-                    catch (InterruptedException e) {
-                        throw new RuntimeException(e);
-                    }
-                }
-            });
-            t.start();
-
-            String subject = subject();
-            // Keep the total queued bytes below the reconnect buffer size (default 8MB).
-            // This test only needs the outgoing-pending counters to report a backlog;
-            // flooding past the reconnect buffer would (correctly) throw if the connection
-            // briefly reconnects under load on a slow/contended machine, which is not what
-            // this test targets.
-            byte[] data = new byte[2 * 1024];
-            for (int x = 0; x < 3000; x++) {
-                nc.publish(subject, data);
-            }
-            tKeepGoing.set(false);
-            t.join();
-
-            assertTrue(outgoingPendingMessageCount.get() > 0);
-            assertTrue(outgoingPendingBytes.get() > outgoingPendingMessageCount.get() * 1000);
-        });
-    }
 }

src/test/java/io/nats/client/impl/KeyValueTests.java

@@ -2035,14 +2035,17 @@ else if (mcount == 4) {
             StreamInfo si = jsm.getStreamInfo(rawStream);
             assertEquals(1, si.getStreamState().getMsgCount());
 
-            long safety = 0;
+            // Bound the wait by time, not by a poll count. How much wall time a fixed number
+            // of unthrottled getStreamInfo round trips buys depends entirely on the machine.
+            long timeoutAt = System.currentTimeMillis() + 10_000;
             long gotZero = -1;
-            while (++safety < 10000 && errorLatch.getCount() > 0) {
+            while (System.currentTimeMillis() < timeoutAt && errorLatch.getCount() > 0) {
                 si = jsm.getStreamInfo(rawStream);
                 if (si.getStreamState().getMsgCount() == 0) {
                     gotZero = System.currentTimeMillis();
                     break;
                 }
+                sleep(10);
             }
             assertEquals(1, errorLatch.getCount(), error.get());
             assertEquals(2, messages.get());
@@ -2058,14 +2061,15 @@ else if (mcount == 4) {
             si = jsm.getStreamInfo(rawStream);
             assertEquals(1, si.getStreamState().getMsgCount());
 
-            safety = 0;
+            timeoutAt = System.currentTimeMillis() + 10_000;
             gotZero = -1;
-            while (++safety < 10000 && errorLatch.getCount() > 0) {
+            while (System.currentTimeMillis() < timeoutAt && errorLatch.getCount() > 0) {
                 si = jsm.getStreamInfo(rawStream);
                 if (si.getStreamState().getMsgCount() == 0) {
                     gotZero = System.currentTimeMillis();
                     break;
                 }
+                sleep(10);
             }
             assertEquals(1, errorLatch.getCount(), error.get());
             assertEquals(4, messages.get());
@@ -2141,19 +2145,22 @@ else if (mcount == 4) {
             StreamInfo si = jsm.getStreamInfo(rawStream);
             assertEquals(1, si.getStreamState().getMsgCount());
 
-            kv.delete(key);
             long mark = System.currentTimeMillis();
+            kv.delete(key);
             si = jsm.getStreamInfo(rawStream);
             assertEquals(1, si.getStreamState().getMsgCount());
 
-            long safety = 0;
+            // Bound the wait by time, not by a poll count. How much wall time a fixed number
+            // of unthrottled getStreamInfo round trips buys depends entirely on the machine.
+            long timeoutAt = System.currentTimeMillis() + 10_000;
             long gotZero = -1;
-            while (++safety < 10000 && errorLatch.getCount() > 0) {
+            while (System.currentTimeMillis() < timeoutAt && errorLatch.getCount() > 0) {
                 si = jsm.getStreamInfo(rawStream);
                 if (si.getStreamState().getMsgCount() == 0) {
                     gotZero = System.currentTimeMillis();
                     break;
                 }
+                sleep(10);
             }
             assertEquals(1, errorLatch.getCount(), error.get());
             assertEquals(2, messages.get());
@@ -2165,14 +2172,15 @@ else if (mcount == 4) {
             mark = System.currentTimeMillis();
             kv.purge(key);
 
-            safety = 0;
+            timeoutAt = System.currentTimeMillis() + 10_000;
             gotZero = -1;
-            while (++safety < 10000 && errorLatch.getCount() > 0) {
+            while (System.currentTimeMillis() < timeoutAt && errorLatch.getCount() > 0) {
                 si = jsm.getStreamInfo(rawStream);
                 if (si.getStreamState().getMsgCount() == 0) {
                     gotZero = System.currentTimeMillis();
                     break;
                 }
+                sleep(10);
             }
             assertEquals(1, errorLatch.getCount(), error.get());
             assertEquals(4, messages.get());

src/test/java/io/nats/client/impl/NatsConnectionImplTests.java

@@ -18,10 +18,7 @@
 import org.junit.jupiter.api.Test;
 
 import java.io.IOException;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
-import java.util.concurrent.ScheduledExecutorService;
-import java.util.concurrent.ThreadFactory;
+import java.util.concurrent.*;
 import java.util.concurrent.atomic.AtomicLong;
 
 import static io.nats.client.utils.TestBase.*;
@@ -323,4 +320,27 @@ public void testReaderWriterThreadFactories() throws Exception {
             assertFalse(writerFactory.created.isEmpty());
         }
     }
+
+    @Test
+    public void testOutgoingPendingCountCoverage() throws Exception {
+        runInServer(nc -> {
+            NatsConnection conn = (NatsConnection)nc;
+
+            // Stop the writer so nothing drains the outgoing queue, giving a deterministic backlog
+            // to exercise the pending-count getters against. Reading them while the writer is live
+            // is an unwinnable race (a fast machine drains to 0; a slow machine backs up past the
+            // reconnect buffer and the publish throws), and they can't be read during reconnect at
+            // all because that path holds closeSocketLock for the whole reconnect.
+            conn.getWriter().stop().get(LONG_TIMEOUT_MS, TimeUnit.MILLISECONDS);
+
+            String subject = subject();
+            byte[] data = new byte[2 * 1024]; // > 1000 bytes so pending bytes > count * 1000
+            for (int x = 0; x < 20; x++) {
+                conn.publish(subject, data);
+            }
+
+            assertTrue(conn.outgoingPendingMessageCount() > 0);
+            assertTrue(conn.outgoingPendingBytes() > conn.outgoingPendingMessageCount() * 1000);
+        });
+    }
 }