Resolve the situation where the field value contains commas in line formatter. (#16609)

Author: wenyanshi-123

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

@@ -72,8 +72,19 @@ 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/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,16 @@ 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