当同步任务,服务名配置为ALL时,关闭同步任务,目标节点的同步心跳不会停止,只能关闭服务

[dragonTalon]

版本:0.4.9
同步任务:nacos(1.4)->nacos(2.1)

当配置ALL时,serviceName的同步是通过com.alibaba.nacossync.timer.CheckRunningStatusAllThread类进行同步的,存放在
com.alibaba.nacossync.extension.impl.NacosSyncToNacosServiceImpl#serviceClient 格式是 taskId+serviceName(具体的serviceName)

同步任务

 public void run() {
          Long startTime = System.currentTimeMillis();
          try {
              //获取ALL配置的任务
              List<TaskDO> taskDOS = taskAccessService.findServiceNameIsNull()
                      .stream().filter(t -> t.getStatus() == null || t.getStatus() == 0)
                      .collect(Collectors.toList());
              if (CollectionUtils.isEmpty(taskDOS)) {
                  return;
              }
              for (TaskDO taskDO : taskDOS) {
                  List<String> serviceNameList = getServiceNameList(taskDO);
                  if (CollectionUtils.isEmpty(serviceNameList)) {
                     continue;
                  }
                  //获取所有的service名称不为空的对象
                  //如果是null，证明此时没有处理完成
                  List<String> filterService = serviceNameList.stream()
                          .filter(serviceName -> skyWalkerCacheServices.getFinishedTask(taskDO.getTaskId() + serviceName ) == null)
                          .collect(Collectors.toList());
                  
                  if (CollectionUtils.isEmpty(filterService)) {
                      continue;
                  }
                  
                  // 当删除任务后，此时任务的状态为DELETE,不会执行数据同步
                  if (TaskStatusEnum.SYNC.getCode().equals(taskDO.getTaskStatus())) {
                      fastSyncHelper.syncWithThread(taskDO, filterService);
                  }
              }
          }catch (Exception e) {
              log.warn("CheckRunningStatusThread Exception ", e);
          }
          metricsManager.record(MetricsStatisticsType.DISPATCHER_TASK, System.currentTimeMillis() - startTime);
      }

com.alibaba.nacossync.timer.FastSyncHelper#syncByIndex
copy一个临时实例的task任务

    private void syncByIndex(TaskDO taskDO, String serviceName, int index) {
        long startTime = System.currentTimeMillis();
        TaskDO task = new TaskDO();
        BeanUtils.copyProperties(taskDO, task);
        task.setServiceName(serviceName);
        task.setOperationId(taskDO.getTaskId() + serviceName);
        if (syncManagerService.sync(task, index)) {
            skyWalkerCacheServices.addFinishedTask(task);
            log.info("sync thread : {} sync finish ,time consuming ：{}", Thread.currentThread().getId(),
                    System.currentTimeMillis() - startTime);
            metricsManager.record(MetricsStatisticsType.SYNC_TASK_RT, System.currentTimeMillis() - startTime);
        } else {
            log.warn("listenerSyncTaskEvent sync failure.");
        }
    }

但是删除的逻辑是根据数据库中配置的task类进行delete,删除的key是taskId:ALL这样导致所有通过CheckRunningStatusAllThread逻辑生成的临时task类,产生的心跳无法stop,导致这些实例在同步任务关闭之后还是存活在目标nacos中

com.alibaba.nacossync.extension.impl.NacosSyncToNacosServiceImpl#popNamingService

    public NamingService popNamingService(TaskDO taskDO) {
        String key = buildClientKey(taskDO);
        Set<NamingService> namingServices = serviceClient.get(key);
        if (CollectionUtils.isNotEmpty(namingServices)) {
            return namingServices.iterator().next();
        }
        log.warn("{} 无可用 namingservice", key);
        return null;
    }

nacos-sync是不是缺少一个和CheckRunningStatusAllThread对应的delete任务

[dragonTalon]

#347

[paderlol]

merged