content: package com.alibaba.otter.canal.connector.core.producer;
  
  import java.util.ArrayList;
  import java.util.Collections;
  import java.util.HashMap;
  import java.util.HashSet;
  import java.util.LinkedHashMap;
  import java.util.List;
  import java.util.Map;
  import java.util.Set;
  import java.util.concurrent.ThreadPoolExecutor;
  
  import org.apache.commons.lang.StringUtils;
  
  import com.alibaba.otter.canal.common.utils.ExecutorTemplate;
  import com.alibaba.otter.canal.connector.core.filter.AviaterRegexFilter;
  import com.alibaba.otter.canal.protocol.CanalEntry;
  import com.alibaba.otter.canal.protocol.CanalEntry.Entry;
  import com.alibaba.otter.canal.protocol.CanalEntry.RowChange;
  import com.alibaba.otter.canal.protocol.FlatMessage;
  import com.alibaba.otter.canal.protocol.Message;
  import com.google.common.cache.CacheBuilder;
  import com.google.common.collect.Lists;
  import com.google.common.collect.MigrateMap;
  import com.google.protobuf.ByteString;
  import com.google.protobuf.InvalidProtocolBufferException;
  
  /**
   * process MQ Message utils
   *
   * @author agapple 2019年9月29日 下午12:36:26
   * @since 5.0.0
   */
  public class MQMessageUtils {
  
      private static Map<String, List<PartitionData>>    partitionDatas    = MigrateMap.makeComputingMap(CacheBuilder.newBuilder()
                                                                               .softValues(),
                                                                               pkHashConfigs -> {
                                                                                   List<PartitionData> datas = new ArrayList<>();
  
                                                                                   String[] pkHashConfigArray = StringUtils.split(StringUtils.replace(pkHashConfigs,
                                                                                       ",",
                                                                                       ";"),
                                                                                       ";");
                                                                                   // schema.table:id^name
                                                                                   for (String pkHashConfig : pkHashConfigArray) {
                                                                                       PartitionData data = new PartitionData();
                                                                                       int i = pkHashConfig.lastIndexOf(":");
                                                                                       if (i > 0) {
                                                                                           String pkStr = pkHashConfig.substring(i + 1);
                                                                                           if (pkStr.equalsIgnoreCase("$pk$")) {
                                                                                               data.hashMode.autoPkHash = true;
                                                                                           } else {
                                                                                               data.hashMode.pkNames = Lists.newArrayList(StringUtils.split(pkStr,
                                                                                                   '^'));
                                                                                           }
  
                                                                                           pkHashConfig = pkHashConfig.substring(0,
                                                                                               i);
                                                                                       } else {
                                                                                           data.hashMode.tableHash = true;
                                                                                       }
  
                                                                                       if (!isWildCard(pkHashConfig)) {
                                                                                           data.simpleName = pkHashConfig;
                                                                                       } else {
                                                                                           data.regexFilter = new AviaterRegexFilter(pkHashConfig);
                                                                                       }
                                                                                       datas.add(data);
                                                                                   }
  
                                                                                   return datas;
                                                                               });
  
      private static Map<String, List<DynamicTopicData>> dynamicTopicDatas = MigrateMap.makeComputingMap(CacheBuilder.newBuilder()
                                                                               .softValues(),
                                                                                pkHashConfigs -> {
                                                                                  List<DynamicTopicData> datas = new ArrayList<>();
                                                                                  String[] dynamicTopicArray = StringUtils.split(StringUtils.replace(pkHashConfigs,
                                                                                      ",",
                                                                                      ";"),
                                                                                      ";");
                                                                                  // schema.table
                                                                                  for (String dynamicTopic : dynamicTopicArray) {
                                                                                      DynamicTopicData data = new DynamicTopicData();
  
                                                                                      if (!isWildCard(dynamicTopic)) {
                                                                                          data.simpleName = dynamicTopic;
                                                                                      } else {
                                                                                          if (dynamicTopic.contains("\\.")) {
                                                                                              data.tableRegexFilter = new AviaterRegexFilter(dynamicTopic);
                                                                                          } else {
                                                                                              data.schemaRegexFilter = new AviaterRegexFilter(dynamicTopic);
                                                                                          }
                                                                                      }
                                                                                      datas.add(data);
                                                                                  }
  
                                                                                  return datas;
                                                                              });
  
      private static Map<String, List<TopicPartitionData>> topicPartitionDatas = MigrateMap.makeComputingMap(CacheBuilder.newBuilder()
                                                                                  .softValues(),
                                                                                  tPConfigs -> {
                                                                                      List<TopicPartitionData> datas = new ArrayList<>();
                                                                                      String[] tPArray = StringUtils.split(StringUtils.replace(tPConfigs,
                                                                                              ",",
                                                                                              ";"),
                                                                                              ";");
                                                                                      for (String tPConfig : tPArray) {
                                                                                          TopicPartitionData data = new TopicPartitionData();
                                                                                          int i = tPConfig.lastIndexOf(":");
                                                                                          if (i > 0) {
                                                                                              String tStr = tPConfig.substring(0, i);
                                                                                              String pStr = tPConfig.substring(i + 1);
                                                                                              if (!isWildCard(tStr)) {
                                                                                                  data.simpleName = tStr;
                                                                                              } else {
                                                                                                  data.regexFilter = new AviaterRegexFilter(tStr);
                                                                                              }
                                                                                              if (!StringUtils.isEmpty(pStr) && StringUtils.isNumeric(pStr)) {
                                                                                                  data.partitionNum = Integer.valueOf(pStr);
                                                                                              }
                                                                                              datas.add(data);
                                                                                          }
                                                                                      }
  
                                                                                      return datas;
                                                                                  });
  
      /**
       * 按 schema 或者 schema+table 将 message 分配到对应topic
       *
       * @param message 原message
       * @param defaultTopic 默认topic
       * @param dynamicTopicConfigs 动态topic规则
       * @return 分隔后的message map
       */
      public static Map<String, Message> messageTopics(Message message, String defaultTopic, String dynamicTopicConfigs) {
          List<CanalEntry.Entry> entries;
          if (message.isRaw()) {
              List<ByteString> rawEntries = message.getRawEntries();
              entries = new ArrayList<>(rawEntries.size());
              for (ByteString byteString : rawEntries) {
                  CanalEntry.Entry entry;
                  try {
                      entry = CanalEntry.Entry.parseFrom(byteString);
                  } catch (InvalidProtocolBufferException e) {
                      throw new RuntimeException(e);
                  }
                  entries.add(entry);
              }
          } else {
              entries = message.getEntries();
          }
          Map<String, Message> messages = new HashMap<>();
          for (CanalEntry.Entry entry : entries) {
              // 如果有topic路由,则忽略begin/end事件
              if (entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONBEGIN
                  || entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONEND) {
                  continue;
              }
  
              String schemaName = entry.getHeader().getSchemaName();
              String tableName = entry.getHeader().getTableName();
  
              if (StringUtils.isEmpty(schemaName) || StringUtils.isEmpty(tableName)) {
                  put2MapMessage(messages, message.getId(), defaultTopic, entry);
              } else {
                  Set<String> topics = matchTopics(schemaName + "." + tableName, dynamicTopicConfigs);
                  if (topics != null) {
                      for (String topic : topics) {
                          put2MapMessage(messages, message.getId(), topic, entry);
                      }
                  } else {
                      topics = matchTopics(schemaName, dynamicTopicConfigs);
                      if (topics != null) {
                          for (String topic : topics) {
                              put2MapMessage(messages, message.getId(), topic, entry);
                          }
                      } else {
                          put2MapMessage(messages, message.getId(), defaultTopic, entry);
                      }
                  }
              }
          }
          return messages;
      }
  
      /**
       * 多线程构造message的rowChanged对象，比如为partition/flastMessage转化等处理 </br>
       * 因为protobuf对象的序列化和反序列化是cpu密集型，串行执行会有代价
       */
      public static EntryRowData[] buildMessageData(Message message, ThreadPoolExecutor executor) {
          ExecutorTemplate template = new ExecutorTemplate(executor);
          if (message.isRaw()) {
              List<ByteString> rawEntries = message.getRawEntries();
              final EntryRowData[] datas = new EntryRowData[rawEntries.size()];
              int i = 0;
              for (ByteString byteString : rawEntries) {
                  final int index = i;
                  template.submit(() -> {
                      try {
                          Entry entry = Entry.parseFrom(byteString);
                          RowChange rowChange = RowChange.parseFrom(entry.getStoreValue());
                          datas[index] = new EntryRowData();
                          datas[index].entry = entry;
                          datas[index].rowChange = rowChange;
                      } catch (InvalidProtocolBufferException e) {
                          throw new RuntimeException(e);
                      }
                  });
  
                  i++;
              }
  
              template.waitForResult();
              return datas;
          } else {
              final EntryRowData[] datas = new EntryRowData[message.getEntries().size()];
              int i = 0;
              for (Entry entry : message.getEntries()) {
                  final int index = i;
                  template.submit(() -> {
                      try {
                          RowChange rowChange = RowChange.parseFrom(entry.getStoreValue());
                          datas[index] = new EntryRowData();
                          datas[index].entry = entry;
                          datas[index].rowChange = rowChange;
                      } catch (InvalidProtocolBufferException e) {
                          throw new RuntimeException(e);
                      }
                  });
  
                  i++;
              }
  
              template.waitForResult();
              return datas;
          }
      }
  
      /**
       * 将 message 分区
       *
       * @param partitionsNum 分区数
       * @param pkHashConfigs 分区库表主键正则表达式
       * @param databaseHash 是否取消根据database进行hash
       * @return 分区message数组
       */
      @SuppressWarnings("unchecked")
      public static Message[] messagePartition(EntryRowData[] datas, long id, Integer partitionsNum,
                                               String pkHashConfigs, boolean databaseHash) {
          if (partitionsNum == null) {
              partitionsNum = 1;
          }
          Message[] partitionMessages = new Message[partitionsNum];
          List<Entry>[] partitionEntries = new List[partitionsNum];
          for (int i = 0; i < partitionsNum; i++) {
              // 注意一下并发
              partitionEntries[i] = Collections.synchronizedList(new ArrayList<>());
          }
  
          for (EntryRowData data : datas) {
              CanalEntry.Entry entry = data.entry;
              CanalEntry.RowChange rowChange = data.rowChange;
              // 如果有分区路由,则忽略begin/end事件
              if (entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONBEGIN
                  || entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONEND) {
                  continue;
              }
  
              if (rowChange.getIsDdl()) {
                  partitionEntries[0].add(entry);
              } else {
                  if (rowChange.getRowDatasList() != null && !rowChange.getRowDatasList().isEmpty()) {
                      String database = entry.getHeader().getSchemaName();
                      String table = entry.getHeader().getTableName();
                      HashMode hashMode = getPartitionHashColumns(database + "." + table, pkHashConfigs);
                      if (hashMode == null) {
                          // 如果都没有匹配，发送到第一个分区
                          partitionEntries[0].add(entry);
                      } else if (hashMode.tableHash) {
                          int hashCode = table.hashCode();
                          int pkHash = Math.abs(hashCode) % partitionsNum;
                          pkHash = Math.abs(pkHash);
                          // tableHash not need split entry message
                          partitionEntries[pkHash].add(entry);
                      } else {
                          // build new entry
                          Entry.Builder builder = Entry.newBuilder(entry);
                          RowChange.Builder rowChangeBuilder = RowChange.newBuilder(rowChange);
  
                          for (CanalEntry.RowData rowData : rowChange.getRowDatasList()) {
                              int hashCode = 0;
                              if (databaseHash) {
                                  hashCode = database.hashCode();
                              }
                              CanalEntry.EventType eventType = rowChange.getEventType();
                              List<CanalEntry.Column> columns = null;
                              if (eventType == CanalEntry.EventType.DELETE) {
                                  columns = rowData.getBeforeColumnsList();
                              } else {
                                  columns = rowData.getAfterColumnsList();
                              }
  
                              if (hashMode.autoPkHash) {
                                  // isEmpty use default pkNames
                                  for (CanalEntry.Column column : columns) {
                                      if (column.getIsKey()) {
                                          hashCode = hashCode ^ column.getValue().hashCode();
                                      }
                                  }
                              } else {
                                  for (CanalEntry.Column column : columns) {
                                      if (checkPkNamesHasContain(hashMode.pkNames, column.getName())) {
                                          hashCode = hashCode ^ column.getValue().hashCode();
                                      }
                                  }
                              }
  
                              int pkHash = Math.abs(hashCode) % partitionsNum;
                              pkHash = Math.abs(pkHash);
                              // clear rowDatas
                              rowChangeBuilder.clearRowDatas();
                              rowChangeBuilder.addRowDatas(rowData);
                              builder.clearStoreValue();
                              builder.setStoreValue(rowChangeBuilder.build().toByteString());
                              partitionEntries[pkHash].add(builder.build());
                          }
                      }
                  } else {
                      // 针对stmt/mixed binlog格式的query事件
                      partitionEntries[0].add(entry);
                  }
              }
          }
  
          for (int i = 0; i < partitionsNum; i++) {
              List<Entry> entriesTmp = partitionEntries[i];
              if (!entriesTmp.isEmpty()) {
                  partitionMessages[i] = new Message(id, entriesTmp);
              }
          }
  
          return partitionMessages;
      }
  
      /**
       * 将Message转换为FlatMessage
       *
       * @return FlatMessage列表
       * @author agapple 2018年12月11日 下午1:28:32
       */
      public static List<FlatMessage> messageConverter(EntryRowData[] datas, long id) {
          List<FlatMessage> flatMessages = new ArrayList<>();
          for (EntryRowData entryRowData : datas) {
              CanalEntry.Entry entry = entryRowData.entry;
              CanalEntry.RowChange rowChange = entryRowData.rowChange;
              // 如果有分区路由,则忽略begin/end事件
              if (entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONBEGIN
                  || entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONEND) {
                  continue;
              }
  
              // build flatMessage
              CanalEntry.EventType eventType = rowChange.getEventType();
              FlatMessage flatMessage = new FlatMessage(id);
              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());
              flatMessage.setGtid(entry.getHeader().getGtid());
  
              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操作将记录修改前的值
                          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;
      }
  
      /**
       * 将FlatMessage按指定的字段值hash拆分
       *
       * @param flatMessage flatMessage
       * @param partitionsNum 分区数量
       * @param pkHashConfigs hash映射
       * @param databaseHash 是否取消根据database进行hash
       * @return 拆分后的flatMessage数组
       */
      public static FlatMessage[] messagePartition(FlatMessage flatMessage, Integer partitionsNum, String pkHashConfigs,
                                                   boolean databaseHash) {
          if (partitionsNum == null) {
              partitionsNum = 1;
          }
          FlatMessage[] partitionMessages = new FlatMessage[partitionsNum];
  
          if (flatMessage.getIsDdl()) {
              partitionMessages[0] = flatMessage;
          } else {
              if (flatMessage.getData() != null && !flatMessage.getData().isEmpty()) {
                  String database = flatMessage.getDatabase();
                  String table = flatMessage.getTable();
                  HashMode hashMode = getPartitionHashColumns(database + "." + table, pkHashConfigs);
                  if (hashMode == null) {
                      // 如果都没有匹配，发送到第一个分区
                      partitionMessages[0] = flatMessage;
                  } else if (hashMode.tableHash) {
                      int hashCode = table.hashCode();
                      int pkHash = Math.abs(hashCode) % partitionsNum;
                      // math.abs可能返回负值，这里再取反，把出现负值的数据还是写到固定的分区，仍然可以保证消费顺序
                      pkHash = Math.abs(pkHash);
                      partitionMessages[pkHash] = flatMessage;
                  } else {
                      List<String> pkNames = hashMode.pkNames;
                      if (hashMode.autoPkHash) {
                          pkNames = flatMessage.getPkNames();
                      }
  
                      int idx = 0;
                      for (Map<String, String> row : flatMessage.getData()) {
                          int hashCode = 0;
                          if (databaseHash) {
                              hashCode = database.hashCode();
                          }
                          if (pkNames != null) {
                              for (String pkName : pkNames) {
                                  String value = row.get(pkName);
                                  if (value == null) {
                                      value = "";
                                  }
                                  hashCode = hashCode ^ value.hashCode();
                              }
                          }
  
                          int pkHash = Math.abs(hashCode) % partitionsNum;
                          // math.abs可能返回负值，这里再取反，把出现负值的数据还是写到固定的分区，仍然可以保证消费顺序
                          pkHash = Math.abs(pkHash);
  
                          FlatMessage flatMessageTmp = partitionMessages[pkHash];
                          if (flatMessageTmp == null) {
                              flatMessageTmp = new FlatMessage(flatMessage.getId());
                              partitionMessages[pkHash] = flatMessageTmp;
                              flatMessageTmp.setDatabase(flatMessage.getDatabase());
                              flatMessageTmp.setTable(flatMessage.getTable());
                              flatMessageTmp.setIsDdl(flatMessage.getIsDdl());
                              flatMessageTmp.setType(flatMessage.getType());
                              flatMessageTmp.setSql(flatMessage.getSql());
                              flatMessageTmp.setSqlType(flatMessage.getSqlType());
                              flatMessageTmp.setMysqlType(flatMessage.getMysqlType());
                              flatMessageTmp.setEs(flatMessage.getEs());
                              flatMessageTmp.setTs(flatMessage.getTs());
                              flatMessageTmp.setPkNames(flatMessage.getPkNames());
                              flatMessageTmp.setGtid(flatMessage.getGtid());
                          }
                          List<Map<String, String>> data = flatMessageTmp.getData();
                          if (data == null) {
                              data = new ArrayList<>();
                              flatMessageTmp.setData(data);
                          }
                          data.add(row);
                          if (flatMessage.getOld() != null && !flatMessage.getOld().isEmpty()) {
                              List<Map<String, String>> old = flatMessageTmp.getOld();
                              if (old == null) {
                                  old = new ArrayList<>();
                                  flatMessageTmp.setOld(old);
                              }
                              old.add(flatMessage.getOld().get(idx));
                          }
                          idx++;
                      }
                  }
              } else {
                  // 针对stmt/mixed binlog格式的query事件
                  partitionMessages[0] = flatMessage;
              }
          }
          return partitionMessages;
      }
  
      /**
       * match return List , not match return null
       */
      public static HashMode getPartitionHashColumns(String name, String pkHashConfigs) {
          if (StringUtils.isEmpty(pkHashConfigs)) {
              return null;
          }
  
          List<PartitionData> datas = partitionDatas.get(pkHashConfigs);
          for (PartitionData data : datas) {
              if (data.simpleName != null) {
                  if (data.simpleName.equalsIgnoreCase(name)) {
                      return data.hashMode;
                  }
              } else {
                  if (data.regexFilter.filter(name)) {
                      return data.hashMode;
                  }
              }
          }
  
          return null;
      }
  
      private static Set<String> matchTopics(String name, String dynamicTopicConfigs) {
          String[] router = StringUtils.split(StringUtils.replace(dynamicTopicConfigs, ",", ";"), ";");
          Set<String> topics = new HashSet<>();
          for (String item : router) {
              int i = item.indexOf(":");
              if (i > -1) {
                  String topic = item.substring(0, i).trim();
                  String topicConfigs = item.substring(i + 1).trim();
                  if (matchDynamicTopic(name, topicConfigs)) {
                      topics.add(topic);
                      // 匹配了一个就退出
                      break;
                  }
              } else if (matchDynamicTopic(name, item)) {
                  // 匹配了一个就退出
                  topics.add(name.toLowerCase());
                  break;
              }
          }
          return topics.isEmpty() ? null : topics;
      }
  
      public static boolean matchDynamicTopic(String name, String dynamicTopicConfigs) {
          if (StringUtils.isEmpty(dynamicTopicConfigs)) {
              return false;
          }
  
          boolean res = false;
          List<DynamicTopicData> datas = dynamicTopicDatas.get(dynamicTopicConfigs);
          for (DynamicTopicData data : datas) {
              if (data.simpleName != null) {
                  if (data.simpleName.equalsIgnoreCase(name)) {
                      res = true;
                      break;
                  }
              } else if (name.contains(".")) {
                  if (data.tableRegexFilter != null && data.tableRegexFilter.filter(name)) {
                      res = true;
                      break;
                  }
              } else {
                  if (data.schemaRegexFilter != null && data.schemaRegexFilter.filter(name)) {
                      res = true;
                      break;
                  }
              }
          }
          return res;
      }
  
      public static boolean checkPkNamesHasContain(List<String> pkNames, String name) {
          for (String pkName : pkNames) {
              if (pkName.equalsIgnoreCase(name)) {
                  return true;
              }
          }
  
          return false;
      }
  
      public static Integer parseDynamicTopicPartition(String name, String tPConfigs) {
          if (!StringUtils.isEmpty(tPConfigs)) {
              List<TopicPartitionData> datas = topicPartitionDatas.get(tPConfigs);
              for (TopicPartitionData data : datas) {
                  if (data.simpleName != null) {
                      if (data.simpleName.equalsIgnoreCase(name)) {
                          return data.partitionNum;
                      }
                  } else {
                      if (data.regexFilter.filter(name)) {
                          return data.partitionNum;
                      }
                  }
              }
          }
          return null;
      }
  
      private static boolean isWildCard(String value) {
          // not contaiins '.' ?
          return StringUtils.containsAny(value, new char[] { '*', '?', '+', '|', '(', ')', '{', '}', '[', ']', '\\', '$',
                  '^' });
      }
  
      private static void put2MapMessage(Map<String, Message> messageMap, Long messageId, String topicName,
                                         CanalEntry.Entry entry) {
          Message message = messageMap.get(topicName);
          if (message == null) {
              message = new Message(messageId, new ArrayList<>());
              messageMap.put(topicName, message);
          }
          message.getEntries().add(entry);
      }
  
      public static class PartitionData {
          /**
           * 当{schemaName.tableName}没有正则时，就匹配{schemaName.tableName}
           */
          public String             simpleName;
          /**
           * 当{schemaName.tableName}有正则时，匹配正则
           */
          public AviaterRegexFilter regexFilter;
          /**
           * hash模式
           */
          public HashMode           hashMode = new HashMode();
      }
  
      public static class HashMode {
          /**
           * 当{schemaName.tableName}:$pk$时，使用自动主键hash
           */
          public boolean      autoPkHash = false;
          /**
           * 当表达式仅为{schemaName.tableName}，没有:时，仅使用table hash
           */
          public boolean      tableHash  = false;
  
          /**
           * 当表达式为{schemaName.tableName}:id^name^age时，pkNames为：id, name, age三个
           */
          public List<String> pkNames    = new ArrayList<>();
      }
  
      public static class DynamicTopicData {
  
          public String             simpleName;
          public AviaterRegexFilter schemaRegexFilter;
          public AviaterRegexFilter tableRegexFilter;
      }
  
      public static class TopicPartitionData {
  
          public String             simpleName;
          public AviaterRegexFilter regexFilter;
          public Integer            partitionNum;
      }
  
      public static class EntryRowData {
  
          public Entry     entry;
          public RowChange rowChange;
      }
  }
  