code: @Override
      public boolean subscribeChange(ClientIdentity identity) {
          if (StringUtils.isNotEmpty(identity.getFilter())) {
              logger.info("subscribe filter change to " + identity.getFilter());
              AviaterRegexFilter aviaterFilter = new AviaterRegexFilter(identity.getFilter());
  
              boolean isGroup = (eventParser instanceof GroupEventParser);
              if (isGroup) {
                  // 处理group的模式
                  List<CanalEventParser> eventParsers = ((GroupEventParser) eventParser).getEventParsers();
                  for (CanalEventParser singleEventParser : eventParsers) {// 需要遍历启动
                      ((AbstractEventParser) singleEventParser).setEventFilter(aviaterFilter);
                  }
              } else {
                  ((AbstractEventParser) eventParser).setEventFilter(aviaterFilter);
              }
  
          }
  
          // filter的处理规则
          // a. parser处理数据过滤处理
          // b. sink处理数据的路由&分发,一份parse数据经过sink后可以分发为多份，每份的数据可以根据自己的过滤规则不同而有不同的数据
          // 后续内存版的一对多分发，可以考虑
          return true;
      }
docs: mq的配置