docs: @Override
      public void subscribe(ClientIdentity clientIdentity) throws CanalServerException {
          checkStart(clientIdentity.getDestination());
  
          CanalInstance canalInstance = canalInstances.get(clientIdentity.getDestination());
          if (!canalInstance.getMetaManager().isStart()) {
              canalInstance.getMetaManager().start();
          }
  
          canalInstance.getMetaManager().subscribe(clientIdentity); // 执行一下meta订阅
  
          Position position = canalInstance.getMetaManager().getCursor(clientIdentity);
          if (position == null) {
              position = canalInstance.getEventStore().getFirstPosition();// 获取一下store中的第一条
              if (position != null) {
                  canalInstance.getMetaManager().updateCursor(clientIdentity, position); // 更新一下cursor
              }
              logger.info("subscribe successfully, {} with first position:{} ", clientIdentity, position);
          } else {
              logger.info("subscribe successfully, use last cursor position:{} ", clientIdentity, position);
          }
  
          // 通知下订阅关系变化
          canalInstance.subscribeChange(clientIdentity);
      }