diff --git a/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/ConnectionHealthTest.java b/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/ConnectionHealthTest.java index 5bf57bc9..d2472653 100644 --- a/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/ConnectionHealthTest.java +++ b/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(); diff --git a/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/IoThreadErrorSurfacedOnRowApiTest.java b/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/IoThreadErrorSurfacedOnRowApiTest.java index f9d401f6..d14c3a7c 100644 --- a/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/IoThreadErrorSurfacedOnRowApiTest.java +++ b/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 diff --git a/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/ServerErrorAckTerminalTest.java b/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/ServerErrorAckTerminalTest.java index fdc9c90d..04fd0048 100644 --- a/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/ServerErrorAckTerminalTest.java +++ b/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; diff --git a/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/SymbolDictRecycleFsnContinuityTest.java b/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/SymbolDictRecycleFsnContinuityTest.java index 5c77f340..81c6c68b 100644 --- a/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/SymbolDictRecycleFsnContinuityTest.java +++ b/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;