@Test(expected = Exception.class) public void testNacosSyncToZookeeperWithException() throws Exception { Assert.assertFalse(nacosSyncToZookeeperService.sync(null)); }
@Override public boolean sync(TaskDO taskDO) { try { NamingService sourceNamingService = nacosServerHolder.get(taskDO.getSourceClusterId(), taskDO.getGroupName()); CuratorFramework client = zookeeperServerHolder.get(taskDO.getDestClusterId(), taskDO.getGroupName()); nacosListenerMap.putIfAbsent(taskDO.getTaskId(), event -> { if (event instanceof NamingEvent) { try { List<Instance> sourceInstances = sourceNamingService.getAllInstances(taskDO.getServiceName()); Set<String> newInstanceUrlSet = getWaitingToAddInstance(taskDO, client, sourceInstances); deleteInvalidInstances(taskDO, client, newInstanceUrlSet); instanceBackupMap.put(taskDO.getTaskId(), newInstanceUrlSet); tryToCompensate(taskDO, sourceNamingService, sourceInstances); } catch (Exception e) { log.error("event process fail, taskId:{}", taskDO.getTaskId(), e); metricsManager.recordError(MetricsStatisticsType.SYNC_ERROR); } } }); sourceNamingService.subscribe(taskDO.getServiceName(), nacosListenerMap.get(taskDO.getTaskId())); } catch (Exception e) { log.error("sync task from nacos to zk was failed, taskId:{}", taskDO.getTaskId(), e); metricsManager.recordError(MetricsStatisticsType.SYNC_ERROR); return false; } return true; }
NacosSyncToZookeeperServiceImpl implements SyncService { @Override public boolean sync(TaskDO taskDO) { try { NamingService sourceNamingService = nacosServerHolder.get(taskDO.getSourceClusterId(), taskDO.getGroupName()); CuratorFramework client = zookeeperServerHolder.get(taskDO.getDestClusterId(), taskDO.getGroupName()); nacosListenerMap.putIfAbsent(taskDO.getTaskId(), event -> { if (event instanceof NamingEvent) { try { List<Instance> sourceInstances = sourceNamingService.getAllInstances(taskDO.getServiceName()); Set<String> newInstanceUrlSet = getWaitingToAddInstance(taskDO, client, sourceInstances); deleteInvalidInstances(taskDO, client, newInstanceUrlSet); instanceBackupMap.put(taskDO.getTaskId(), newInstanceUrlSet); tryToCompensate(taskDO, sourceNamingService, sourceInstances); } catch (Exception e) { log.error("event process fail, taskId:{}", taskDO.getTaskId(), e); metricsManager.recordError(MetricsStatisticsType.SYNC_ERROR); } } }); sourceNamingService.subscribe(taskDO.getServiceName(), nacosListenerMap.get(taskDO.getTaskId())); } catch (Exception e) { log.error("sync task from nacos to zk was failed, taskId:{}", taskDO.getTaskId(), e); metricsManager.recordError(MetricsStatisticsType.SYNC_ERROR); return false; } return true; } }
NacosSyncToZookeeperServiceImpl implements SyncService { @Override public boolean sync(TaskDO taskDO) { try { NamingService sourceNamingService = nacosServerHolder.get(taskDO.getSourceClusterId(), taskDO.getGroupName()); CuratorFramework client = zookeeperServerHolder.get(taskDO.getDestClusterId(), taskDO.getGroupName()); nacosListenerMap.putIfAbsent(taskDO.getTaskId(), event -> { if (event instanceof NamingEvent) { try { List<Instance> sourceInstances = sourceNamingService.getAllInstances(taskDO.getServiceName()); Set<String> newInstanceUrlSet = getWaitingToAddInstance(taskDO, client, sourceInstances); deleteInvalidInstances(taskDO, client, newInstanceUrlSet); instanceBackupMap.put(taskDO.getTaskId(), newInstanceUrlSet); tryToCompensate(taskDO, sourceNamingService, sourceInstances); } catch (Exception e) { log.error("event process fail, taskId:{}", taskDO.getTaskId(), e); metricsManager.recordError(MetricsStatisticsType.SYNC_ERROR); } } }); sourceNamingService.subscribe(taskDO.getServiceName(), nacosListenerMap.get(taskDO.getTaskId())); } catch (Exception e) { log.error("sync task from nacos to zk was failed, taskId:{}", taskDO.getTaskId(), e); metricsManager.recordError(MetricsStatisticsType.SYNC_ERROR); return false; } return true; } @Autowired  NacosSyncToZookeeperServiceImpl(SkyWalkerCacheServices skyWalkerCacheServices,
        NacosServerHolder nacosServerHolder, ZookeeperServerHolder zookeeperServerHolder); }
NacosSyncToZookeeperServiceImpl implements SyncService { @Override public boolean sync(TaskDO taskDO) { try { NamingService sourceNamingService = nacosServerHolder.get(taskDO.getSourceClusterId(), taskDO.getGroupName()); CuratorFramework client = zookeeperServerHolder.get(taskDO.getDestClusterId(), taskDO.getGroupName()); nacosListenerMap.putIfAbsent(taskDO.getTaskId(), event -> { if (event instanceof NamingEvent) { try { List<Instance> sourceInstances = sourceNamingService.getAllInstances(taskDO.getServiceName()); Set<String> newInstanceUrlSet = getWaitingToAddInstance(taskDO, client, sourceInstances); deleteInvalidInstances(taskDO, client, newInstanceUrlSet); instanceBackupMap.put(taskDO.getTaskId(), newInstanceUrlSet); tryToCompensate(taskDO, sourceNamingService, sourceInstances); } catch (Exception e) { log.error("event process fail, taskId:{}", taskDO.getTaskId(), e); metricsManager.recordError(MetricsStatisticsType.SYNC_ERROR); } } }); sourceNamingService.subscribe(taskDO.getServiceName(), nacosListenerMap.get(taskDO.getTaskId())); } catch (Exception e) { log.error("sync task from nacos to zk was failed, taskId:{}", taskDO.getTaskId(), e); metricsManager.recordError(MetricsStatisticsType.SYNC_ERROR); return false; } return true; } @Autowired  NacosSyncToZookeeperServiceImpl(SkyWalkerCacheServices skyWalkerCacheServices,
        NacosServerHolder nacosServerHolder, ZookeeperServerHolder zookeeperServerHolder); @Override boolean delete(TaskDO taskDO); @Override boolean sync(TaskDO taskDO); }
NacosSyncToZookeeperServiceImpl implements SyncService { @Override public boolean sync(TaskDO taskDO) { try { NamingService sourceNamingService = nacosServerHolder.get(taskDO.getSourceClusterId(), taskDO.getGroupName()); CuratorFramework client = zookeeperServerHolder.get(taskDO.getDestClusterId(), taskDO.getGroupName()); nacosListenerMap.putIfAbsent(taskDO.getTaskId(), event -> { if (event instanceof NamingEvent) { try { List<Instance> sourceInstances = sourceNamingService.getAllInstances(taskDO.getServiceName()); Set<String> newInstanceUrlSet = getWaitingToAddInstance(taskDO, client, sourceInstances); deleteInvalidInstances(taskDO, client, newInstanceUrlSet); instanceBackupMap.put(taskDO.getTaskId(), newInstanceUrlSet); tryToCompensate(taskDO, sourceNamingService, sourceInstances); } catch (Exception e) { log.error("event process fail, taskId:{}", taskDO.getTaskId(), e); metricsManager.recordError(MetricsStatisticsType.SYNC_ERROR); } } }); sourceNamingService.subscribe(taskDO.getServiceName(), nacosListenerMap.get(taskDO.getTaskId())); } catch (Exception e) { log.error("sync task from nacos to zk was failed, taskId:{}", taskDO.getTaskId(), e); metricsManager.recordError(MetricsStatisticsType.SYNC_ERROR); return false; } return true; } @Autowired  NacosSyncToZookeeperServiceImpl(SkyWalkerCacheServices skyWalkerCacheServices,
        NacosServerHolder nacosServerHolder, ZookeeperServerHolder zookeeperServerHolder); @Override boolean delete(TaskDO taskDO); @Override boolean sync(TaskDO taskDO);  }