Merge branch 'main' into feat/TS-6222

Author: sheyanjie-qq

deploy-pom.xml

@@ -5,7 +5,7 @@
 
   <groupId>com.taosdata.jdbc</groupId>
   <artifactId>taos-jdbcdriver</artifactId>
-  <version>3.6.0</version>
+  <version>3.6.1</version>
   <packaging>jar</packaging>
 
   <name>JDBCDriver</name>

pom.xml

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

src/main/java/com/taosdata/jdbc/AbstractDriver.java

@@ -65,7 +65,7 @@ protected Connection getWSConnection(String url, ConnectionParam param, Properti
         if (log.isDebugEnabled()){
             log.debug("getWSConnection, url = {}", StringUtils.getBasicUrl(url));
             try {
-                ObjectMapper objectMapper = new ObjectMapper();
+                ObjectMapper objectMapper = JsonUtil.getObjectMapper();
                 log.debug("getWSConnection, ConnectionParam = {}", objectMapper.writeValueAsString(param));
             } catch (JsonProcessingException e) {
                 log.error("Error serializing ConnectionParam", e);

src/main/java/com/taosdata/jdbc/TSDBJNIConnector.java

@@ -5,6 +5,7 @@
 import com.fasterxml.jackson.databind.node.ObjectNode;
 import com.taosdata.jdbc.enums.SchemalessProtocolType;
 import com.taosdata.jdbc.enums.SchemalessTimestampType;
+import com.taosdata.jdbc.utils.JsonUtil;
 import com.taosdata.jdbc.utils.TaosInfo;
 
 import java.io.UnsupportedEncodingException;
@@ -86,7 +87,7 @@ public static void init(Properties props) throws SQLWarning {
         synchronized (LOCK) {
             if (!isInitialized) {
 
-                ObjectMapper objectMapper = new ObjectMapper();
+                ObjectMapper objectMapper = JsonUtil.getObjectMapper();
                 ObjectNode configJSON = objectMapper.createObjectNode();
 
                 for (String key : props.stringPropertyNames()) {

src/main/java/com/taosdata/jdbc/rs/RestfulStatement.java

@@ -8,6 +8,7 @@
 import com.taosdata.jdbc.TSDBError;
 import com.taosdata.jdbc.TSDBErrorNumbers;
 import com.taosdata.jdbc.utils.HttpClientPoolUtil;
+import com.taosdata.jdbc.utils.JsonUtil;
 import com.taosdata.jdbc.utils.SqlSyntaxValidator;
 
 import java.sql.Connection;
@@ -86,7 +87,7 @@ public boolean execute(String sql, Long reqId) throws SQLException {
         boolean result = true;
 
         String response = HttpClientPoolUtil.execute(getUrl(), sql, this.conn.getAuth(), reqId);
-        ObjectMapper objectMapper = new ObjectMapper();
+        ObjectMapper objectMapper = JsonUtil.getObjectMapper();
         JsonNode jsonObject = null;
         try {
             jsonObject = objectMapper.readTree(response);

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

@@ -177,6 +177,7 @@ public boolean next() throws SQLException {
 
             if (resp.isCompleted() || isClosed) {
                 blockData.setCompleted(true);
+                isCompleted = true;
                 return false;
             }
 

src/main/java/com/taosdata/jdbc/ws/entity/Request.java

@@ -2,6 +2,7 @@
 
 import com.fasterxml.jackson.core.JsonProcessingException;
 import com.fasterxml.jackson.databind.ObjectMapper;
+import com.taosdata.jdbc.utils.JsonUtil;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -41,7 +42,7 @@ public void setArgs(Payload args) {
 
     @Override
     public String toString() {
-        ObjectMapper objectMapper = new ObjectMapper();
+        ObjectMapper objectMapper = JsonUtil.getObjectMapper();
         try {
             return objectMapper.writeValueAsString(this);
         } catch (JsonProcessingException e) {

src/main/java/com/taosdata/jdbc/ws/tmq/WSConsumer.java

@@ -140,13 +140,19 @@ private ConsumerRecords<V> doPoll(Duration timeout, Deserializer<V> deserializer
             return ConsumerRecords.emptyRecord();
         }
 
+        messageId = pollResp.getMessageId();
+
         if (pollResp.getMessageType() == TmqMessageType.TMQ_RES_TABLE_META.getCode() || pollResp.getMessageType() == TmqMessageType.TMQ_RES_METADATA.getCode()) {
             Request fetchJsonMetaReq = factory.generateFetchJsonMeata(pollResp.getMessageId());
             FetchJsonMetaResp fetchJsonMetaResp = (FetchJsonMetaResp) transport.send(fetchJsonMetaReq);
             if (Code.SUCCESS.getCode() != fetchJsonMetaResp.getCode()) {
                 throw new SQLException("consumer fetch json meta error, code: (0x" + Integer.toHexString(fetchJsonMetaResp.getCode()) + "), message: " + fetchJsonMetaResp.getMessage());
             }
-            messageId = pollResp.getMessageId();
+
+            if (fetchJsonMetaResp.getData() == null || fetchJsonMetaResp.getData().getMetas() == null) {
+                return ConsumerRecords.emptyRecord();
+            }
+
             ConsumerRecords<V> records = new ConsumerRecords<>();
 
             for (Meta meta : fetchJsonMetaResp.getData().getMetas()){
@@ -170,9 +176,6 @@ private ConsumerRecords<V> doPoll(Duration timeout, Deserializer<V> deserializer
             return ConsumerRecords.emptyRecord();
         }
 
-        messageId = pollResp.getMessageId();
-
-
         ConsumerRecords<V> records = new ConsumerRecords<>();
         try (WSConsumerResultSet rs = new WSConsumerResultSet(transport, factory, pollResp.getMessageId(), pollResp.getDatabase())) {
             if (deserializer instanceof MapEnhanceDeserializer){