Fixmqtt1017

example/mqtt/src/main/java/org/apache/iotdb/mqtt/MQTTClient.java

@@ -72,8 +72,20 @@ private static void jsonPayloadFormatter(BlockingConnection connection) throws E
 
   // The database must be created in advance
   private static void linePayloadFormatter(BlockingConnection connection) throws Exception {
+    // myTable,tag1=t1,tag2=t2 fieldKey1="1,2,3" 1740109006001
+    String payload = "myTable,tag1=t1,tag2=t2 fieldKey1=\"1,2,3\" 1740109006001";
+    connection.publish(DATABASE + "/myTopic", payload.getBytes(), QoS.AT_LEAST_ONCE, false);
+    Thread.sleep(10);
+
+    payload = "myTable,tag1=t1,tag2=t2 fieldKey1=\"1,2,3\" 1740109006002";
+    connection.publish(DATABASE + "/myTopic", payload.getBytes(), QoS.AT_LEAST_ONCE, false);
+    Thread.sleep(10);
 
-    String payload =
+    payload = "myTable,tag1=t1,tag2=t2 fieldKey1=\"1,2,3\" 1740109006003";
+    connection.publish(DATABASE + "/myTopic", payload.getBytes(), QoS.AT_LEAST_ONCE, false);
+    Thread.sleep(10);
+
+    payload =
         "test1,tag1=t1,tag2=t2 attr3=a5,attr4=a4 field1=\"fieldValue1\",field2=1i,field3=1u 1";
     connection.publish(DATABASE + "/myTopic", payload.getBytes(), QoS.AT_LEAST_ONCE, false);
     Thread.sleep(10);

iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/source/ConfigRegionListeningQueue.java

@@ -21,6 +21,7 @@
 
 import org.apache.iotdb.commons.auth.AuthException;
 import org.apache.iotdb.commons.auth.user.LocalFileUserAccessor;
+import org.apache.iotdb.commons.auth.user.UserId;
 import org.apache.iotdb.commons.conf.IoTDBConstant;
 import org.apache.iotdb.commons.exception.MetadataException;
 import org.apache.iotdb.commons.pipe.datastructure.queue.listening.AbstractPipeListeningQueue;
@@ -157,15 +158,21 @@ public synchronized void tryListenToSnapshots(
                   : null,
               snapshotPathInfo.getRight());
       if (type == CNSnapshotFileType.USER_ROLE) {
-        long userId = Long.parseLong(snapshotPath.toFile().getName().split("_")[0]);
+        UserId userId = UserId.parse(snapshotPath.toFile().getName().split("_")[0]);
         try {
-          curEvent.setAuthUserName(
-              ConfigNode.getInstance()
-                  .getConfigManager()
-                  .getPermissionManager()
-                  .getUserName(userId));
+          if (userId.isLong()) {
+            curEvent.setAuthUserName(
+                ConfigNode.getInstance()
+                    .getConfigManager()
+                    .getPermissionManager()
+                    .getUserName(userId.longValue));
+          } else if (userId.isString()) {
+            curEvent.setAuthUserName(userId.strValue);
+          }
         } catch (AuthException e) {
-          LOGGER.warn("Failed to collect user name for user id {}", userId, e);
+          if (userId.isLong()) {
+            LOGGER.warn("Failed to collect user name for user id {}", userId.longValue, e);
+          }
         }
       }
       events.add(curEvent);

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/mqtt/LinePayloadFormatter.java

@@ -31,6 +31,7 @@
 import java.util.List;
 import java.util.regex.Matcher;
 import java.util.regex.Pattern;
+import java.util.stream.Stream;
 
 /**
  * The Line payload formatter. myTable,tag1=value1,tag2=value2 attr1=value1,attr2=value2
@@ -190,7 +191,7 @@ private boolean setFields(Matcher matcher, TableMessage message) {
     List<Object> values = new ArrayList<>();
     String fieldsGroup = matcher.group(FIELDS);
     if (fieldsGroup != null && !fieldsGroup.isEmpty()) {
-      String[] fieldPairs = fieldsGroup.split(COMMA);
+      String[] fieldPairs = splitFieldPairs(fieldsGroup);
       for (String fieldPair : fieldPairs) {
         if (!fieldPair.isEmpty()) {
           String[] keyValue = fieldPair.split(EQUAL);
@@ -213,6 +214,15 @@ private boolean setFields(Matcher matcher, TableMessage message) {
     }
   }
 
+  private String[] splitFieldPairs(String fieldsGroup) {
+    if (fieldsGroup == null || fieldsGroup.isEmpty()) return new String[0];
+    Matcher m = Pattern.compile("\\w+=\"[^\"]*\"|\\w+=[^,]*").matcher(fieldsGroup);
+    Stream.Builder<String> builder = Stream.builder();
+
+    while (m.find()) builder.add(m.group());
+    return builder.build().toArray(String[]::new);
+  }
+
   private Pair<TSDataType, Object> analyticValue(String value) {
     if (value.startsWith("\"") && value.endsWith("\"")) {
       // String

iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/auth/user/UserId.java

@@ -0,0 +1,27 @@
+package org.apache.iotdb.commons.auth.user;
+
+public class UserId {
+  public final String strValue;
+  public final Long longValue;
+
+  private UserId(String str, Long lng) {
+    this.strValue = str;
+    this.longValue = lng;
+  }
+
+  public static UserId parse(String input) {
+    try {
+      return new UserId(null, Long.parseLong(input));
+    } catch (NumberFormatException e) {
+      return new UserId(input, null);
+    }
+  }
+
+  public boolean isLong() {
+    return longValue != null;
+  }
+
+  public boolean isString() {
+    return strValue != null;
+  }
+}