docs: public static List<FlatMessage> messageConverter(Message message) {
          try {
              if (message == null) {
                  return null;
              }
  
              List<FlatMessage> flatMessages = new ArrayList<>();
              List<CanalEntry.Entry> entrys = null;
              if (message.isRaw()) {
                  List<ByteString> rawEntries = message.getRawEntries();
                  entrys = new ArrayList<CanalEntry.Entry>(rawEntries.size());
                  for (ByteString byteString : rawEntries) {
                      CanalEntry.Entry entry = CanalEntry.Entry.parseFrom(byteString);
                      entrys.add(entry);
                  }
              } else {
                  entrys = message.getEntries();
              }
  
              for (CanalEntry.Entry entry : entrys) {
                  if (entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONBEGIN
                      || entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONEND) {
                      continue;
                  }
  
                  CanalEntry.RowChange rowChange;
                  try {
                      rowChange = CanalEntry.RowChange.parseFrom(entry.getStoreValue());
                  } catch (Exception e) {
                      throw new RuntimeException("ERROR ## parser of eromanga-event has an error , data:"
                                                 + entry.toString(), e);
                  }
  
                  CanalEntry.EventType eventType = rowChange.getEventType();
  
                  FlatMessage flatMessage = new FlatMessage(message.getId());
                  flatMessages.add(flatMessage);
                  flatMessage.setDatabase(entry.getHeader().getSchemaName());
                  flatMessage.setTable(entry.getHeader().getTableName());
                  flatMessage.setIsDdl(rowChange.getIsDdl());
                  flatMessage.setType(eventType.toString());
                  flatMessage.setEs(entry.getHeader().getExecuteTime());
                  flatMessage.setTs(System.currentTimeMillis());
                  flatMessage.setSql(rowChange.getSql());
  
                  if (!rowChange.getIsDdl()) {
                      Map<String, Integer> sqlType = new LinkedHashMap<>();
                      Map<String, String> mysqlType = new LinkedHashMap<>();
                      List<Map<String, String>> data = new ArrayList<>();
                      List<Map<String, String>> old = new ArrayList<>();
  
                      Set<String> updateSet = new HashSet<>();
                      boolean hasInitPkNames = false;
                      for (CanalEntry.RowData rowData : rowChange.getRowDatasList()) {
                          if (eventType != CanalEntry.EventType.INSERT && eventType != CanalEntry.EventType.UPDATE
                              && eventType != CanalEntry.EventType.DELETE) {
                              continue;
                          }
  
                          Map<String, String> row = new LinkedHashMap<>();
                          List<CanalEntry.Column> columns;
  
                          if (eventType == CanalEntry.EventType.DELETE) {
                              columns = rowData.getBeforeColumnsList();
                          } else {
                              columns = rowData.getAfterColumnsList();
                          }
  
                          for (CanalEntry.Column column : columns) {
                              if (!hasInitPkNames && column.getIsKey()) {
                                  flatMessage.addPkName(column.getName());
                              }
                              sqlType.put(column.getName(), column.getSqlType());
                              mysqlType.put(column.getName(), column.getMysqlType());
                              if (column.getIsNull()) {
                                  row.put(column.getName(), null);
                              } else {
                                  row.put(column.getName(), column.getValue());
                              }
                              // 获取update为true的字段
                              if (column.getUpdated()) {
                                  updateSet.add(column.getName());
                              }
                          }
  
                          hasInitPkNames = true;
                          if (!row.isEmpty()) {
                              data.add(row);
                          }
  
                          if (eventType == CanalEntry.EventType.UPDATE) {
                              Map<String, String> rowOld = new LinkedHashMap<>();
                              for (CanalEntry.Column column : rowData.getBeforeColumnsList()) {
                                  if (updateSet.contains(column.getName())) {
                                      if (column.getIsNull()) {
                                          rowOld.put(column.getName(), null);
                                      } else {
                                          rowOld.put(column.getName(), column.getValue());
                                      }
                                  }
                              }
                              // update操作将记录修改前的值
                              if (!rowOld.isEmpty()) {
                                  old.add(rowOld);
                              }
                          }
                      }
                      if (!sqlType.isEmpty()) {
                          flatMessage.setSqlType(sqlType);
                      }
                      if (!mysqlType.isEmpty()) {
                          flatMessage.setMysqlType(mysqlType);
                      }
                      if (!data.isEmpty()) {
                          flatMessage.setData(data);
                      }
                      if (!old.isEmpty()) {
                          flatMessage.setOld(old);
                      }
                  }
              }
              return flatMessages;
          } catch (Exception e) {
              throw new RuntimeException(e);
          }
      }