Fix resource leaks in QwpWebSocketSender

Close the encoder's native memory when MicrobatchBuffer
allocation fails in the constructor. Previously only buffer0
was cleaned up, leaving the NativeBufferWriter leaked.

In ensureConnected(), close the WebSocket client if
WebSocketSendQueue construction fails (e.g. Thread.start()
throws OOM), preventing a dangling socket and I/O thread.

In flushSync(), fail the in-flight window entry when
sendBinary() throws, so close() does not hang on
awaitEmpty() waiting for an ACK that will never arrive.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

Author: mtopolnik

core/src/main/java/io/questdb/client/cutlass/qwp/client/QwpWebSocketSender.java

@@ -182,11 +182,14 @@ private QwpWebSocketSender(
         // Initialize double-buffering if async mode (window > 1)
         if (inFlightWindowSize > 1) {
             int microbatchBufferSize = Math.max(DEFAULT_MICROBATCH_BUFFER_SIZE, autoFlushBytes * 2);
-            this.buffer0 = new MicrobatchBuffer(microbatchBufferSize, autoFlushRows, autoFlushBytes, autoFlushIntervalNanos);
             try {
+                this.buffer0 = new MicrobatchBuffer(microbatchBufferSize, autoFlushRows, autoFlushBytes, autoFlushIntervalNanos);
                 this.buffer1 = new MicrobatchBuffer(microbatchBufferSize, autoFlushRows, autoFlushBytes, autoFlushIntervalNanos);
             } catch (Throwable t) {
-                buffer0.close();
+                if (buffer0 != null) {
+                    buffer0.close();
+                }
+                encoder.close();
                 throw t;
             }
             this.activeBuffer = buffer0;
@@ -980,9 +983,16 @@ private void ensureConnected() {
             // Initialize send queue for async mode (window > 1)
             // The send queue handles both sending AND receiving (single I/O thread)
             if (inFlightWindowSize > 1) {
-                sendQueue = new WebSocketSendQueue(client, inFlightWindow,
-                        WebSocketSendQueue.DEFAULT_ENQUEUE_TIMEOUT_MS,
-                        WebSocketSendQueue.DEFAULT_SHUTDOWN_TIMEOUT_MS);
+                try {
+                    sendQueue = new WebSocketSendQueue(client, inFlightWindow,
+                            WebSocketSendQueue.DEFAULT_ENQUEUE_TIMEOUT_MS,
+                            WebSocketSendQueue.DEFAULT_SHUTDOWN_TIMEOUT_MS);
+                } catch (Throwable t) {
+                    inFlightWindow = null;
+                    client.close();
+                    client = null;
+                    throw new LineSenderException("Failed to start I/O thread for " + host + ":" + port, t);
+                }
             }
             // Sync mode (window=1): no send queue - we send and read ACKs synchronously
 
@@ -1134,8 +1144,18 @@ private void flushSync() {
 
                 LOG.debug("Sending sync batch [seq={}, bytes={}, rows={}, maxSentSymbolId={}, useSchemaRef={}]", batchSequence, messageSize, tableBuffer.getRowCount(), currentBatchMaxSymbolId, useSchemaRef);
 
-                // Send over WebSocket
-                client.sendBinary(buffer.getBufferPtr(), messageSize);
+                // Send over WebSocket and fail the in-flight entry if send throws,
+                // so close() does not hang waiting for an ACK that will never arrive.
+                try {
+                    client.sendBinary(buffer.getBufferPtr(), messageSize);
+                } catch (LineSenderException e) {
+                    failExpectedIfNeeded(batchSequence, e);
+                    throw e;
+                } catch (Throwable t) {
+                    LineSenderException error = new LineSenderException("Failed to send batch " + batchSequence, t);
+                    failExpectedIfNeeded(batchSequence, error);
+                    throw error;
+                }
 
                 // Wait for ACK synchronously
                 waitForAck(batchSequence);