fix: websocket reconnect null point error (#296)

Closes: TD-38690

Author: sheyanjie-qq

CHANGELOG.md

@@ -7,12 +7,33 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [unreleased]
 
-**Full Changelog**: [3.7.6...](https://github.com/taosdata/taos-connector-jdbc/compare/3.7.6...)
+**Full Changelog**: [3.7.8...](https://github.com/taosdata/taos-connector-jdbc/compare/3.7.8...)
+
+### Features
+
+
+- Optimize the load balancing logic to achieve rebalancing after node failure recovery. (#291) ([9c473bc](https://github.com/taosdata/taos-connector-jdbc/commit/9c473bccc699ddba33316efa313da2230143ac34))
+
+
+## [3.7.8] - 2025-11-20
+
+**Full Changelog**: [3.7.7...3.7.8](https://github.com/taosdata/taos-connector-jdbc/compare/3.7.7...3.7.8)
+
+### Bug Fixes
+
+
+- Fix getTables need identifier quote string bug. (#292) ([ce5651d](https://github.com/taosdata/taos-connector-jdbc/commit/ce5651d31c87174c06ba567b9a1596c54a6c288b))
+
+
+## [3.7.7] - 2025-11-13
+
+**Full Changelog**: [3.7.6...3.7.7](https://github.com/taosdata/taos-connector-jdbc/compare/3.7.6...3.7.7)
 
 ### Bug Fixes
 
 
 - Load properties issue in windows (#289) ([b6841d7](https://github.com/taosdata/taos-connector-jdbc/commit/b6841d726f9520556ddf069a5e0aea95af3be876))
+- WebSocket statement timeout setting resets transport's timeout (#290) ([c9792c0](https://github.com/taosdata/taos-connector-jdbc/commit/c9792c0eea77e9978ff29bf64c55b823d0c0755f))
 
 
 ## [3.7.6] - 2025-10-17

deploy-pom.xml

@@ -5,7 +5,7 @@
 
     <groupId>com.taosdata.jdbc</groupId>
     <artifactId>taos-jdbcdriver</artifactId>
-    <version>3.7.8</version>
+    <version>3.7.9</version>
 
     <packaging>jar</packaging>
 
@@ -48,6 +48,7 @@
         <netty-all.version>4.1.127.Final</netty-all.version>
         <junit.version>4.13.2</junit.version>
         <mockito.version>4.11.0</mockito.version>
+        <byte-buddy.version>1.12.22</byte-buddy.version>
         <slf4j.version>1.7.36</slf4j.version>
         <logback.version>1.3.16</logback.version>
     </properties>
@@ -71,6 +72,12 @@
             <version>${mockito.version}</version>
             <scope>test</scope>
         </dependency>
+        <dependency>
+            <groupId>net.bytebuddy</groupId>
+            <artifactId>byte-buddy</artifactId>
+            <version>${byte-buddy.version}</version>
+            <scope>test</scope>
+        </dependency>
         <dependency>
             <groupId>ch.qos.logback</groupId>
             <artifactId>logback-classic</artifactId>

pom.xml

@@ -3,7 +3,7 @@
     <modelVersion>4.0.0</modelVersion>
     <groupId>com.taosdata.jdbc</groupId>
     <artifactId>taos-jdbcdriver</artifactId>
-    <version>3.7.8</version>
+    <version>3.7.9</version>
 
     <packaging>jar</packaging>
     <name>JDBCDriver</name>

src/main/java/com/taosdata/jdbc/ws/Transport.java

@@ -117,13 +117,16 @@ public Response send(Request request, boolean reSend, long timeout) throws SQLEx
         try {
             connectionManager.getCurrentClient().send(reqString);
         } catch (WebsocketNotConnectedException e) {
-            connectionManager.handleConnectionException(false, this);
             try {
+                connectionManager.handleConnectionException(this);
                 if (!reSend) {
                     inFlightRequest.remove(request.getAction(), request.id());
-                    throw new SQLException("reconnect, need to resend " + request.getAction() + " msg");
+                    throw TSDBError.createSQLException(TSDBErrorNumbers.ERROR_CONNECTION_CLOSED);
                 }
                 connectionManager.getCurrentClient().send(reqString);
+            } catch (SQLException ex) {
+                inFlightRequest.remove(request.getAction(), request.id());
+                throw ex;
             } catch (Exception ex) {
                 inFlightRequest.remove(request.getAction(), request.id());
                 throw TSDBError.createSQLException(TSDBErrorNumbers.ERROR_RESTFul_Client_IOException, e.getMessage());
@@ -135,7 +138,11 @@ public Response send(Request request, boolean reSend, long timeout) throws SQLEx
         try {
             response = responseFuture.get();
             handleTaosdError(response);
-        } catch (InterruptedException | ExecutionException e) {
+        } catch (InterruptedException e) {
+            Thread.currentThread().interrupt();
+            inFlightRequest.remove(request.getAction(), request.id());
+            throw TSDBError.createSQLException(TSDBErrorNumbers.ERROR_QUERY_TIMEOUT, e.getMessage());
+        } catch (ExecutionException e) {
             inFlightRequest.remove(request.getAction(), request.id());
             throw TSDBError.createSQLException(TSDBErrorNumbers.ERROR_QUERY_TIMEOUT, e.getMessage());
         }
@@ -193,8 +200,8 @@ public Response send(String action, long reqId, long resultId, long type, byte[]
             Utils.retainByteBuf(buffer);
             connectionManager.getCurrentClient().send(buffer);
         } catch (WebsocketNotConnectedException e) {
-            connectionManager.handleConnectionException(false, this);
             try {
+                connectionManager.handleConnectionException(this);
                 Utils.retainByteBuf(buffer);
                 connectionManager.getCurrentClient().send(buffer);
             } catch (Exception ex) {
@@ -210,7 +217,11 @@ public Response send(String action, long reqId, long resultId, long type, byte[]
         try {
             response = responseFuture.get();
             handleTaosdError(response);
-        } catch (InterruptedException | ExecutionException e) {
+        } catch (InterruptedException e) {
+            Thread.currentThread().interrupt();
+            inFlightRequest.remove(action, reqId);
+            throw TSDBError.createSQLException(TSDBErrorNumbers.ERROR_QUERY_TIMEOUT, e.getMessage());
+        } catch (ExecutionException e) {
             inFlightRequest.remove(action, reqId);
             throw TSDBError.createSQLException(TSDBErrorNumbers.ERROR_QUERY_TIMEOUT, e.getMessage());
         }
@@ -243,7 +254,7 @@ public void sendFetchBlockAsync(long reqId, long resultId) throws SQLException {
             Utils.retainByteBuf(buffer);
             connectionManager.getCurrentClient().send(buffer);
         } catch (WebsocketNotConnectedException e) {
-            connectionManager.handleConnectionException(false, this);
+            connectionManager.handleConnectionException(this);
             throw TSDBError.createSQLException(TSDBErrorNumbers.ERROR_STATEMENT_CLOSED,
                     "Websocket reconnected, but the result set is closed");
         } finally {
@@ -276,13 +287,20 @@ public Response send(String action, long reqId, ByteBuf buffer, boolean resend,
             Utils.retainByteBuf(buffer);
             connectionManager.getCurrentClient().send(buffer);
         } catch (WebsocketNotConnectedException e) {
-            connectionManager.handleConnectionException(false, this);
+            try {
+                connectionManager.handleConnectionException(this);
+            } catch (SQLException ex) {
+                inFlightRequest.remove(action, reqId);
+                throw ex;
+            }
+
+            if (!resend) {
+                inFlightRequest.remove(action, reqId);
+                throw TSDBError.createSQLException(TSDBErrorNumbers.ERROR_CONNECTION_CLOSED);
+            }
+
             try {
                 Utils.retainByteBuf(buffer);
-                if (!resend) {
-                    inFlightRequest.remove(action, reqId);
-                    throw new SQLException("reconnect, need to resend " + action + " msg");
-                }
                 connectionManager.getCurrentClient().send(buffer);
             } catch (Exception ex) {
                 inFlightRequest.remove(action, reqId);
@@ -299,9 +317,13 @@ public Response send(String action, long reqId, ByteBuf buffer, boolean resend,
         try {
             response = responseFuture.get();
             handleTaosdError(response);
-        } catch (InterruptedException | ExecutionException e) {
+        } catch (InterruptedException e) {
+            Thread.currentThread().interrupt();
             inFlightRequest.remove(action, reqId);
             throw TSDBError.createSQLException(TSDBErrorNumbers.ERROR_QUERY_TIMEOUT, e.getMessage());
+        } catch (ExecutionException e) {
+           inFlightRequest.remove(action, reqId);
+            throw TSDBError.createSQLException(TSDBErrorNumbers.ERROR_QUERY_TIMEOUT, e.getMessage());
         }
         return response;
     }
@@ -353,9 +375,13 @@ public Response sendWithoutRetry(Request request, long timeout) throws SQLExcept
                 completableFuture, timeout, TimeUnit.MILLISECONDS, reqString);
         try {
             response = responseFuture.get();
-        } catch (InterruptedException | ExecutionException e) {
-            inFlightRequest.remove(request.getAction(), request.id());
+        } catch (InterruptedException e) {
+            Thread.currentThread().interrupt();
+            throw TSDBError.createSQLException(TSDBErrorNumbers.ERROR_QUERY_TIMEOUT, e.getMessage());
+        } catch (ExecutionException e) {
             throw TSDBError.createSQLException(TSDBErrorNumbers.ERROR_QUERY_TIMEOUT, e.getMessage());
+        } finally {
+            inFlightRequest.remove(request.getAction(), request.id());
         }
         return response;
     }
@@ -374,7 +400,7 @@ public void sendWithoutResponse(Request request) throws SQLException {
         try {
             connectionManager.getCurrentClient().send(request.toString());
         } catch (WebsocketNotConnectedException e) {
-            connectionManager.handleConnectionException(false, this);
+            connectionManager.handleConnectionException(this);
             try {
                 connectionManager.getCurrentClient().send(request.toString());
             } catch (Exception ex) {

src/main/java/com/taosdata/jdbc/ws/WSClient.java

@@ -10,25 +10,19 @@
 import io.netty.bootstrap.Bootstrap;
 import io.netty.buffer.ByteBuf;
 import io.netty.buffer.PooledByteBufAllocator;
-import io.netty.channel.*;
-import io.netty.channel.socket.SocketChannel;
+import io.netty.channel.Channel;
+import io.netty.channel.ChannelFuture;
+import io.netty.channel.ChannelOption;
+import io.netty.channel.ConnectTimeoutException;
 import io.netty.channel.socket.nio.NioSocketChannel;
-import io.netty.handler.codec.http.DefaultHttpHeaders;
-import io.netty.handler.codec.http.HttpClientCodec;
-import io.netty.handler.codec.http.HttpObjectAggregator;
-import io.netty.handler.codec.http.websocketx.*;
-import io.netty.handler.codec.http.websocketx.extensions.WebSocketClientExtensionHandler;
-import io.netty.handler.codec.http.websocketx.extensions.WebSocketClientExtensionHandshaker;
-import io.netty.handler.codec.http.websocketx.extensions.compression.DeflateFrameClientExtensionHandshaker;
-import io.netty.handler.codec.http.websocketx.extensions.compression.PerMessageDeflateClientExtensionHandshaker;
-import io.netty.handler.ssl.SslContext;
-import io.netty.handler.ssl.SslContextBuilder;
-import io.netty.handler.ssl.util.InsecureTrustManagerFactory;
+import io.netty.handler.codec.http.websocketx.BinaryWebSocketFrame;
+import io.netty.handler.codec.http.websocketx.CloseWebSocketFrame;
+import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
 import io.netty.handler.timeout.TimeoutException;
+import io.netty.util.concurrent.Promise;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import javax.net.ssl.SSLException;
 import java.net.URI;
 import java.net.URISyntaxException;
 import java.sql.SQLException;
@@ -80,66 +74,7 @@ private  Bootstrap createBootstrap() {
                 .channel(NioSocketChannel.class)
                 .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, connectionParam.getConnectTimeout())
                 .option(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT)
-                .handler(new ChannelInitializer<SocketChannel>() {
-                    @Override
-                    protected void initChannel(SocketChannel ch) throws SSLException {
-                        ChannelPipeline p = ch.pipeline();
-
-                        if (connectionParam.isUseSsl()) {
-                            if (connectionParam.isDisableSslCertValidation()){
-                                SslContext sslCtx = SslContextBuilder.forClient()
-                                        .trustManager(InsecureTrustManagerFactory.INSTANCE)
-                                        .build();
-                                p.addLast(sslCtx.newHandler(ch.alloc(), host, port));
-                            } else {
-                                SslContext sslCtx = SslContextBuilder.forClient().build();
-                                p.addLast(sslCtx.newHandler(ch.alloc(), host, port));
-                            }
-                        }
-
-                        // for debug
-                        //p.addLast(new LoggingHandler(LogLevel.DEBUG));
-
-                        p.addLast(new HttpClientCodec());
-                        p.addLast(new HttpObjectAggregator(8192));
-
-                        // use custom websocket client handshaker to avoid mask encode
-                        WebSocketClientHandshaker handshaker = new CustomWebSocketClientHandshaker(serverUri,
-                                WebSocketVersion.V13,
-                                null,
-                                true,
-                                new DefaultHttpHeaders(),
-                                100 * 1024 * 1024,
-                                true,
-                                false,
-                                -1L);
-                        p.addLast(new WebSocketHandshakeHandler(handshaker));
-
-                        if (connectionParam.isEnableCompression()) {
-                            WebSocketClientExtensionHandshaker deflateHandshaker = new PerMessageDeflateClientExtensionHandshaker(
-                                    6, false,
-                                    15,       // clientMaxWindowSize (2^15 = 32KB)
-                                    true,     // clientNoContextTakeover
-                                    true    // serverNoContextTakeover
-
-                            );
-
-                            WebSocketClientExtensionHandler extensionHandler = new WebSocketClientExtensionHandler(deflateHandshaker,  new DeflateFrameClientExtensionHandshaker(false),
-                                    new DeflateFrameClientExtensionHandshaker(true));
-                            p.addLast(extensionHandler);
-                        }
-
-                        p.addLast(new WebSocketFrameAggregator(100 * 1024 * 1024)); // max 100MB
-
-                        // Connect with V13 (RFC 6455 aka HyBi-17). You can change it to V08 or V00.
-                        // If you change it to V00, ping is not supported and remember to change
-                        // HttpResponseDecoder to WebSocketHttpResponseDecoder in the pipeline.
-                        final WebSocketClientHandler handler =
-                                new WebSocketClientHandler(connectionParam.getTextMessageHandler(),
-                                        connectionParam.getBinaryMessageHandler());
-                        p.addLast(handler);
-                    }
-                });
+                .handler(new WebSocketChannelInitializer(connectionParam, host, port, serverUri));
         return b;
     }
 
@@ -228,8 +163,32 @@ private Channel getChannel() throws SQLException {
             }
 
             tmpChn = connectFuture.channel();
+            Promise<WebSocketHandshakeHandler> wsHandlerPromise = tmpChn.eventLoop().newPromise();
+
+            final Channel chn = tmpChn;
+            tmpChn.eventLoop().execute(() -> {
+                try {
+                    WebSocketHandshakeHandler wsHandler = chn.pipeline().get(WebSocketHandshakeHandler.class);
+                    if (wsHandler == null) {
+                        wsHandlerPromise.setFailure(new IllegalStateException("WebSocketHandshakeHandler not initialized，initChannel execute failed"));
+                    } else {
+                        wsHandlerPromise.setSuccess(wsHandler);
+                    }
+                } catch (Exception e) {
+                    wsHandlerPromise.setFailure(e);
+                }
+            });
 
-            WebSocketHandshakeHandler wsHandler = tmpChn.pipeline().get(WebSocketHandshakeHandler.class);
+            // wait for the Promise to complete with timeout
+            wsHandlerPromise.awaitUninterruptibly(connectionParam.getConnectTimeout());
+            if (!wsHandlerPromise.isSuccess()) {
+                tmpChn.close().syncUninterruptibly();
+                log.error("get wsHandler failed", wsHandlerPromise.cause());
+                throw TSDBError.createSQLException(TSDBErrorNumbers.ERROR_UNKNOWN, "get wsHandler failed: " + wsHandlerPromise.cause().getMessage());
+            }
+
+            // get result from the Promise
+            WebSocketHandshakeHandler wsHandler = wsHandlerPromise.getNow();
             ChannelFuture handshakeFuture = wsHandler.handshakeFuture();
 
             if (!handshakeFuture.awaitUninterruptibly(connectionParam.getConnectTimeout())) {
@@ -339,23 +298,23 @@ public CompletableFuture<Void> closeAsync() {
 
         return resultFuture;
     }
-   public boolean reconnectBlockingWithoutRetry() {
-       return connectBlocking();
-    }
 
     public boolean connectBlocking(){
-        if (channel != null && channel.isActive()){
-            return true;
-        }
-        if (channel != null){
-            channel.close().syncUninterruptibly();
-        }
+        synchronized (this) {
+            if (channel != null && channel.isActive()) {
+                return true;
+            }
+            if (channel != null) {
+                channel.close().syncUninterruptibly();
+            }
 
-        try {
-            channel = getChannel();
-            return true;
-        } catch (SQLException e){
-            return false;
+            try {
+                channel = getChannel();
+                return true;
+            } catch (SQLException e) {
+                log.error("WebSocket connect failed: ", e);
+                return false;
+            }
         }
     }