docs: public void init() {
          loader = ExtensionLoader.getExtensionLoader(OuterAdapter.class);
  
          String canalServerHost = this.canalClientConfig.getCanalServerHost();
          SocketAddress sa = null;
          if (canalServerHost != null) {
              String[] ipPort = canalServerHost.split(":");
              sa = new InetSocketAddress(ipPort[0], Integer.parseInt(ipPort[1]));
          }
          String zkHosts = this.canalClientConfig.getZookeeperHosts();
  
          if ("tcp".equalsIgnoreCase(canalClientConfig.getMode())) {
              // 初始化canal-client的适配器
              for (CanalClientConfig.CanalAdapter canalAdapter : canalClientConfig.getCanalAdapters()) {
                  List<List<OuterAdapter>> canalOuterAdapterGroups = new ArrayList<>();
  
                  for (CanalClientConfig.Group connectorGroup : canalAdapter.getGroups()) {
                      List<OuterAdapter> canalOutConnectors = new ArrayList<>();
                      for (OuterAdapterConfig c : connectorGroup.getOuterAdapters()) {
                          loadAdapter(c, canalOutConnectors);
                      }
                      canalOuterAdapterGroups.add(canalOutConnectors);
                  }
                  CanalAdapterWorker worker;
                  if (sa != null) {
                      worker = new CanalAdapterWorker(canalClientConfig,
                          canalAdapter.getInstance(),
                          sa,
                          canalOuterAdapterGroups);
                  } else if (zkHosts != null) {
                      worker = new CanalAdapterWorker(canalClientConfig,
                          canalAdapter.getInstance(),
                          zkHosts,
                          canalOuterAdapterGroups);
                  } else {
                      throw new RuntimeException("No canal server connector found");
                  }
                  canalWorkers.put(canalAdapter.getInstance(), worker);
                  worker.start();
                  logger.info("Start adapter for canal instance: {} succeed", canalAdapter.getInstance());
              }
          } else if ("kafka".equalsIgnoreCase(canalClientConfig.getMode())) {
              // 初始化canal-client-kafka的适配器
              for (CanalClientConfig.CanalAdapter canalAdapter : canalClientConfig.getCanalAdapters()) {
                  for (CanalClientConfig.Group group : canalAdapter.getGroups()) {
                      List<List<OuterAdapter>> canalOuterAdapterGroups = new ArrayList<>();
                      List<OuterAdapter> canalOuterAdapters = new ArrayList<>();
                      for (OuterAdapterConfig config : group.getOuterAdapters()) {
                          loadAdapter(config, canalOuterAdapters);
                      }
                      canalOuterAdapterGroups.add(canalOuterAdapters);
  
                      CanalAdapterKafkaWorker canalKafkaWorker = new CanalAdapterKafkaWorker(canalClientConfig,
                          canalClientConfig.getMqServers(),
                          canalAdapter.getInstance(),
                          group.getGroupId(),
                          canalOuterAdapterGroups,
                          canalClientConfig.getFlatMessage());
                      canalMQWorker.put(canalAdapter.getInstance() + "-kafka-" + group.getGroupId(), canalKafkaWorker);
                      canalKafkaWorker.start();
                      logger.info("Start adapter for canal-client mq topic: {} succeed",
                          canalAdapter.getInstance() + "-" + group.getGroupId());
                  }
              }
          } else if ("rocketMQ".equalsIgnoreCase(canalClientConfig.getMode())) {
              // 初始化canal-client-rocketMQ的适配器
              for (CanalClientConfig.CanalAdapter canalAdapter : canalClientConfig.getCanalAdapters()) {
                  for (CanalClientConfig.Group group : canalAdapter.getGroups()) {
                      List<List<OuterAdapter>> canalOuterAdapterGroups = new ArrayList<>();
                      List<OuterAdapter> canalOuterAdapters = new ArrayList<>();
                      for (OuterAdapterConfig config : group.getOuterAdapters()) {
                          loadAdapter(config, canalOuterAdapters);
                      }
                      canalOuterAdapterGroups.add(canalOuterAdapters);
                      CanalAdapterRocketMQWorker rocketMQWorker = new CanalAdapterRocketMQWorker(canalClientConfig,
                          canalClientConfig.getMqServers(),
                          canalAdapter.getInstance(),
                          group.getGroupId(),
                          canalOuterAdapterGroups,
                          canalClientConfig.getAccessKey(),
                          canalClientConfig.getSecretKey(),
                          canalClientConfig.getFlatMessage());
                      canalMQWorker.put(canalAdapter.getInstance() + "-rocketmq-" + group.getGroupId(), rocketMQWorker);
                      rocketMQWorker.start();
  
                      logger.info("Start adapter for canal-client mq topic: {} succeed",
                          canalAdapter.getInstance() + "-" + group.getGroupId());
                  }
              }
          }
      }