@Test(expected = Exception.class) public void tesZookeeperSyncToNacosWithException() throws Exception { Assert.assertFalse(zookeeperSyncToNacosService.sync(null)); }
@Override public boolean sync(TaskDO taskDO) { try { if (treeCacheMap.containsKey(taskDO.getTaskId())) { return true; } TreeCache treeCache = getTreeCache(taskDO); NamingService destNamingService = nacosServerHolder.get(taskDO.getDestClusterId(), null); registerAllInstances(taskDO, destNamingService); Objects.requireNonNull(treeCache).getListenable().addListener((client, event) -> { try { String path = event.getData().getPath(); Map<String, String> queryParam = parseQueryString(path); if (isMatch(taskDO, queryParam) && needSync(queryParam)) { processEvent(taskDO, destNamingService, event, path, queryParam); } } catch (Exception e) { log.error("event process from Zookeeper to Nacos was failed, taskId:{}", taskDO.getTaskId(), e); metricsManager.recordError(MetricsStatisticsType.SYNC_ERROR); } }); } catch (Exception e) { log.error("sync task from Zookeeper to Nacos was failed, taskId:{}", taskDO.getTaskId(), e); metricsManager.recordError(MetricsStatisticsType.SYNC_ERROR); return false; } return true; }
ZookeeperSyncToNacosServiceImpl implements SyncService { @Override public boolean sync(TaskDO taskDO) { try { if (treeCacheMap.containsKey(taskDO.getTaskId())) { return true; } TreeCache treeCache = getTreeCache(taskDO); NamingService destNamingService = nacosServerHolder.get(taskDO.getDestClusterId(), null); registerAllInstances(taskDO, destNamingService); Objects.requireNonNull(treeCache).getListenable().addListener((client, event) -> { try { String path = event.getData().getPath(); Map<String, String> queryParam = parseQueryString(path); if (isMatch(taskDO, queryParam) && needSync(queryParam)) { processEvent(taskDO, destNamingService, event, path, queryParam); } } catch (Exception e) { log.error("event process from Zookeeper to Nacos was failed, taskId:{}", taskDO.getTaskId(), e); metricsManager.recordError(MetricsStatisticsType.SYNC_ERROR); } }); } catch (Exception e) { log.error("sync task from Zookeeper to Nacos was failed, taskId:{}", taskDO.getTaskId(), e); metricsManager.recordError(MetricsStatisticsType.SYNC_ERROR); return false; } return true; } }
ZookeeperSyncToNacosServiceImpl implements SyncService { @Override public boolean sync(TaskDO taskDO) { try { if (treeCacheMap.containsKey(taskDO.getTaskId())) { return true; } TreeCache treeCache = getTreeCache(taskDO); NamingService destNamingService = nacosServerHolder.get(taskDO.getDestClusterId(), null); registerAllInstances(taskDO, destNamingService); Objects.requireNonNull(treeCache).getListenable().addListener((client, event) -> { try { String path = event.getData().getPath(); Map<String, String> queryParam = parseQueryString(path); if (isMatch(taskDO, queryParam) && needSync(queryParam)) { processEvent(taskDO, destNamingService, event, path, queryParam); } } catch (Exception e) { log.error("event process from Zookeeper to Nacos was failed, taskId:{}", taskDO.getTaskId(), e); metricsManager.recordError(MetricsStatisticsType.SYNC_ERROR); } }); } catch (Exception e) { log.error("sync task from Zookeeper to Nacos was failed, taskId:{}", taskDO.getTaskId(), e); metricsManager.recordError(MetricsStatisticsType.SYNC_ERROR); return false; } return true; } @Autowired  ZookeeperSyncToNacosServiceImpl(ZookeeperServerHolder zookeeperServerHolder,
                                           NacosServerHolder nacosServerHolder, SkyWalkerCacheServices skyWalkerCacheServices); }
ZookeeperSyncToNacosServiceImpl implements SyncService { @Override public boolean sync(TaskDO taskDO) { try { if (treeCacheMap.containsKey(taskDO.getTaskId())) { return true; } TreeCache treeCache = getTreeCache(taskDO); NamingService destNamingService = nacosServerHolder.get(taskDO.getDestClusterId(), null); registerAllInstances(taskDO, destNamingService); Objects.requireNonNull(treeCache).getListenable().addListener((client, event) -> { try { String path = event.getData().getPath(); Map<String, String> queryParam = parseQueryString(path); if (isMatch(taskDO, queryParam) && needSync(queryParam)) { processEvent(taskDO, destNamingService, event, path, queryParam); } } catch (Exception e) { log.error("event process from Zookeeper to Nacos was failed, taskId:{}", taskDO.getTaskId(), e); metricsManager.recordError(MetricsStatisticsType.SYNC_ERROR); } }); } catch (Exception e) { log.error("sync task from Zookeeper to Nacos was failed, taskId:{}", taskDO.getTaskId(), e); metricsManager.recordError(MetricsStatisticsType.SYNC_ERROR); return false; } return true; } @Autowired  ZookeeperSyncToNacosServiceImpl(ZookeeperServerHolder zookeeperServerHolder,
                                           NacosServerHolder nacosServerHolder, SkyWalkerCacheServices skyWalkerCacheServices); @Override boolean sync(TaskDO taskDO); @Override boolean delete(TaskDO taskDO); }
ZookeeperSyncToNacosServiceImpl implements SyncService { @Override public boolean sync(TaskDO taskDO) { try { if (treeCacheMap.containsKey(taskDO.getTaskId())) { return true; } TreeCache treeCache = getTreeCache(taskDO); NamingService destNamingService = nacosServerHolder.get(taskDO.getDestClusterId(), null); registerAllInstances(taskDO, destNamingService); Objects.requireNonNull(treeCache).getListenable().addListener((client, event) -> { try { String path = event.getData().getPath(); Map<String, String> queryParam = parseQueryString(path); if (isMatch(taskDO, queryParam) && needSync(queryParam)) { processEvent(taskDO, destNamingService, event, path, queryParam); } } catch (Exception e) { log.error("event process from Zookeeper to Nacos was failed, taskId:{}", taskDO.getTaskId(), e); metricsManager.recordError(MetricsStatisticsType.SYNC_ERROR); } }); } catch (Exception e) { log.error("sync task from Zookeeper to Nacos was failed, taskId:{}", taskDO.getTaskId(), e); metricsManager.recordError(MetricsStatisticsType.SYNC_ERROR); return false; } return true; } @Autowired  ZookeeperSyncToNacosServiceImpl(ZookeeperServerHolder zookeeperServerHolder,
                                           NacosServerHolder nacosServerHolder, SkyWalkerCacheServices skyWalkerCacheServices); @Override boolean sync(TaskDO taskDO); @Override boolean delete(TaskDO taskDO);  }