@Override
  public List<Message> format(String topic, ByteBuf payload) {
    List<Message> messages = new ArrayList<>();
    if (payload == null) {
      return messages;
    }

    String txt = payload.toString(StandardCharsets.UTF_8);
    String[] lines = txt.split(LINE_BREAK);
    // '/' previously defined as a database name
    String database = !topic.contains("/") ? topic : topic.substring(0, topic.indexOf("/"));
    for (String line : lines) {
      if (line.trim().startsWith(WELL)) {
        continue;
      }
      TableMessage message = new TableMessage();
      try {
        Matcher matcher = pattern.matcher(line.trim());
        if (!matcher.matches()) {
          log.warn("Invalid line protocol format ,line is {}", line);
          continue;
        }

        // Parsing Database Name
        message.setDatabase((database));

        // Parsing Table Names
        message.setTable(matcher.group(TABLE));

        // Parsing Tags
        if (!setTags(matcher, message)) {
          log.warn("The tags is error , line is {}", line);
          continue;
        }

        // Parsing Attributes
        if (!setAttributes(matcher, message)) {
          log.warn("The attributes is error , line is {}", line);
          continue;
        }

        // Parsing Fields
        if (!setFields(matcher, message)) {
          log.warn("The fields is error , line is {}", line);
          continue;
        }

        // Parsing timestamp
        if (!setTimestamp(matcher, message)) {
          log.warn("The timestamp is error , line is {}", line);
          continue;
        }

        messages.add(message);
      } catch (Exception e) {
        log.warn(
            "The line pattern parsing fails, and the failed line message is {} ,exception is",
            line,
            e);
      }
    }
    return messages;
  }
The Line payload formatter. myTable,tag1=value1,tag2=value2 attr1=value1,attr2=value2
fieldKey="fieldValue" 1740109006000 \n myTable,tag1=value1,tag2=value2 fieldKey="fieldValue"
1740109006000
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/mqtt/LinePayloadFormatter.java