test(qwp): tolerate terminal errors latched before flush() returns

core/src/test/java/io/questdb/client/test/cutlass/qwp/client/ConnectionHealthTest.java

@@ -32,6 +32,7 @@
 import io.questdb.client.cutlass.http.client.HttpClientException;
 import io.questdb.client.cutlass.http.client.WebSocketClientFactory;
 import io.questdb.client.cutlass.http.client.WebSocketUpgradeException;
+import io.questdb.client.cutlass.line.LineSenderException;
 import io.questdb.client.cutlass.qwp.client.QwpColumnBatch;
 import io.questdb.client.cutlass.qwp.client.QwpColumnBatchHandler;
 import io.questdb.client.cutlass.qwp.client.QwpConnectionHealthTracker;
@@ -461,7 +462,11 @@ public void testSenderHealthIsFailedWithAnOutageAfterATerminalRejection() throws
                 Assert.assertEquals(ConnectionHealth.NONE, h.getOutageSinceEpochMillis());
 
                 sender.table("t").longColumn("v", 1).atNow();
-                sender.flush();
+                try {
+                    sender.flush();
+                } catch (LineSenderException ignored) {
+                    // The terminal NACK may arrive before flush() checks for errors.
+                }
                 // the I/O thread latches FAILED before it hands the terminal error to the handler
                 await(() -> errors.terminalCount() >= 1, 10_000, "the terminal rejection");
                 h = sender.health();

core/src/test/java/io/questdb/client/test/cutlass/qwp/client/IoThreadErrorSurfacedOnRowApiTest.java

@@ -77,7 +77,12 @@ public void testRowApiMethodSurfacesIoThreadTerminalError() throws Exception {
                 // handler routes the rejection through recordFatal, marking
                 // the loop terminal.
                 sender.table("foo").longColumn("v", 1L).atNow();
-                sender.flush();
+                try {
+                    sender.flush();
+                } catch (LineSenderException ignored) {
+                    // The error ACK may arrive before flush() checks for errors; without this catch
+                    // the outer catch would swallow it and skip every assertion below.
+                }
 
                 // Wait for the I/O thread to record the error. After this,
                 // cursorSendLoop.terminalError is populated and the loop has

core/src/test/java/io/questdb/client/test/cutlass/qwp/client/ServerErrorAckTerminalTest.java

@@ -178,7 +178,11 @@ public void testRetriableNackReplaysThenPoisonEscalates() throws Exception {
                     .build();
             try {
                 sender.table("foo").longColumn("v", 1L).atNow();
-                sender.flush();
+                try {
+                    sender.flush();
+                } catch (LineSenderException ignored) {
+                    // The poison escalation may latch before flush() checks for errors.
+                }
 
                 // Each NACK counts one poison strike and triggers a
                 // reconnect+replay; after DEFAULT_MAX_HEAD_FRAME_REJECTIONS strikes
@@ -267,7 +271,11 @@ public void testMaxFrameRejectionsConfigurableFromConnectString() throws Excepti
             Sender sender = Sender.fromConfig(cfg);
             try {
                 sender.table("foo").longColumn("v", 1L).atNow();
-                sender.flush();
+                try {
+                    sender.flush();
+                } catch (LineSenderException ignored) {
+                    // The poison escalation may latch before flush() checks for errors.
+                }
 
                 waitFor(() -> handler.totalBinaryReceived.get() >= 2, 10_000);
                 QwpWebSocketSender wss = (QwpWebSocketSender) sender;

core/src/test/java/io/questdb/client/test/cutlass/qwp/client/SymbolDictRecycleFsnContinuityTest.java

@@ -247,7 +247,12 @@ public void testSenderErrorSpansCarryExternalFsns() throws Exception {
                 sender.setErrorHandler(e -> asyncError.compareAndSet(null, e));
                 try {
                     sender.table("foo").longColumn("v", 1L).atNow();
-                    sender.flush();
+                    try {
+                        sender.flush();
+                    } catch (LineSenderServerException ignored) {
+                        // The terminal NACK may arrive before flush() checks for errors.
+                        // The error is sticky, so the next batch below must still throw it.
+                    }
 
                     waitFor(() -> handler.totalBinaryReceived.get() >= 1, 5_000);
                     waitFor(() -> sender.getLastTerminalError() != null, 5_000);
@@ -349,7 +354,12 @@ public void testLatchedErrorStillThrowsForOldEpochTarget() throws Exception {
                 QwpWebSocketSender sender2 = createRolledSender(server, fsn1);
                 try {
                     sender2.table("foo").longColumn("v", 2L).atNow();
-                    sender2.flush();
+                    try {
+                        sender2.flush();
+                    } catch (LineSenderException ignored) {
+                        // The terminal NACK may arrive before flush() checks for errors.
+                        // The error is sticky, so awaitAckedFsn() below must still throw.
+                    }
                     waitFor(() -> sender2.getLastTerminalError() != null, 5_000);
 
                     LineSenderException thrown = null;