@Test(expected = Exception.class) public void testNacosSyncToNacosWithException() throws Exception { Assert.assertFalse(nacosSyncToNacosService.sync(null)); }
@Override public boolean sync(TaskDO taskDO) { try { NamingService sourceNamingService = nacosServerHolder.get(taskDO.getSourceClusterId(), taskDO.getGroupName()); NamingService destNamingService = nacosServerHolder.get(taskDO.getDestClusterId(), taskDO.getGroupName()); nacosListenerMap.putIfAbsent(taskDO.getTaskId(), event -> { if (event instanceof NamingEvent) { try { Set instanceKeySet = new HashSet(); List<Instance> sourceInstances = sourceNamingService.getAllInstances(taskDO.getServiceName()); for (Instance instance : sourceInstances) { if (needSync(instance.getMetadata())) { destNamingService.registerInstance(taskDO.getServiceName(), buildSyncInstance(instance, taskDO)); instanceKeySet.add(composeInstanceKey(instance)); } } List<Instance> destInstances = destNamingService.getAllInstances(taskDO.getServiceName()); for (Instance instance : destInstances) { if (needDelete(instance.getMetadata(), taskDO) && !instanceKeySet.contains(composeInstanceKey(instance))) { destNamingService.deregisterInstance(taskDO.getServiceName(), instance.getIp(), instance.getPort()); } } } 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 nacos was failed, taskId:{}", taskDO.getTaskId(), e); metricsManager.recordError(MetricsStatisticsType.SYNC_ERROR); return false; } return true; }
NacosSyncToNacosServiceImpl implements SyncService { @Override public boolean sync(TaskDO taskDO) { try { NamingService sourceNamingService = nacosServerHolder.get(taskDO.getSourceClusterId(), taskDO.getGroupName()); NamingService destNamingService = nacosServerHolder.get(taskDO.getDestClusterId(), taskDO.getGroupName()); nacosListenerMap.putIfAbsent(taskDO.getTaskId(), event -> { if (event instanceof NamingEvent) { try { Set instanceKeySet = new HashSet(); List<Instance> sourceInstances = sourceNamingService.getAllInstances(taskDO.getServiceName()); for (Instance instance : sourceInstances) { if (needSync(instance.getMetadata())) { destNamingService.registerInstance(taskDO.getServiceName(), buildSyncInstance(instance, taskDO)); instanceKeySet.add(composeInstanceKey(instance)); } } List<Instance> destInstances = destNamingService.getAllInstances(taskDO.getServiceName()); for (Instance instance : destInstances) { if (needDelete(instance.getMetadata(), taskDO) && !instanceKeySet.contains(composeInstanceKey(instance))) { destNamingService.deregisterInstance(taskDO.getServiceName(), instance.getIp(), instance.getPort()); } } } 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 nacos was failed, taskId:{}", taskDO.getTaskId(), e); metricsManager.recordError(MetricsStatisticsType.SYNC_ERROR); return false; } return true; } }
NacosSyncToNacosServiceImpl implements SyncService { @Override public boolean sync(TaskDO taskDO) { try { NamingService sourceNamingService = nacosServerHolder.get(taskDO.getSourceClusterId(), taskDO.getGroupName()); NamingService destNamingService = nacosServerHolder.get(taskDO.getDestClusterId(), taskDO.getGroupName()); nacosListenerMap.putIfAbsent(taskDO.getTaskId(), event -> { if (event instanceof NamingEvent) { try { Set instanceKeySet = new HashSet(); List<Instance> sourceInstances = sourceNamingService.getAllInstances(taskDO.getServiceName()); for (Instance instance : sourceInstances) { if (needSync(instance.getMetadata())) { destNamingService.registerInstance(taskDO.getServiceName(), buildSyncInstance(instance, taskDO)); instanceKeySet.add(composeInstanceKey(instance)); } } List<Instance> destInstances = destNamingService.getAllInstances(taskDO.getServiceName()); for (Instance instance : destInstances) { if (needDelete(instance.getMetadata(), taskDO) && !instanceKeySet.contains(composeInstanceKey(instance))) { destNamingService.deregisterInstance(taskDO.getServiceName(), instance.getIp(), instance.getPort()); } } } 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 nacos was failed, taskId:{}", taskDO.getTaskId(), e); metricsManager.recordError(MetricsStatisticsType.SYNC_ERROR); return false; } return true; }  }
NacosSyncToNacosServiceImpl implements SyncService { @Override public boolean sync(TaskDO taskDO) { try { NamingService sourceNamingService = nacosServerHolder.get(taskDO.getSourceClusterId(), taskDO.getGroupName()); NamingService destNamingService = nacosServerHolder.get(taskDO.getDestClusterId(), taskDO.getGroupName()); nacosListenerMap.putIfAbsent(taskDO.getTaskId(), event -> { if (event instanceof NamingEvent) { try { Set instanceKeySet = new HashSet(); List<Instance> sourceInstances = sourceNamingService.getAllInstances(taskDO.getServiceName()); for (Instance instance : sourceInstances) { if (needSync(instance.getMetadata())) { destNamingService.registerInstance(taskDO.getServiceName(), buildSyncInstance(instance, taskDO)); instanceKeySet.add(composeInstanceKey(instance)); } } List<Instance> destInstances = destNamingService.getAllInstances(taskDO.getServiceName()); for (Instance instance : destInstances) { if (needDelete(instance.getMetadata(), taskDO) && !instanceKeySet.contains(composeInstanceKey(instance))) { destNamingService.deregisterInstance(taskDO.getServiceName(), instance.getIp(), instance.getPort()); } } } 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 nacos was failed, taskId:{}", taskDO.getTaskId(), e); metricsManager.recordError(MetricsStatisticsType.SYNC_ERROR); return false; } return true; }  @Override boolean delete(TaskDO taskDO); @Override boolean sync(TaskDO taskDO); Instance buildSyncInstance(Instance instance, TaskDO taskDO); }
NacosSyncToNacosServiceImpl implements SyncService { @Override public boolean sync(TaskDO taskDO) { try { NamingService sourceNamingService = nacosServerHolder.get(taskDO.getSourceClusterId(), taskDO.getGroupName()); NamingService destNamingService = nacosServerHolder.get(taskDO.getDestClusterId(), taskDO.getGroupName()); nacosListenerMap.putIfAbsent(taskDO.getTaskId(), event -> { if (event instanceof NamingEvent) { try { Set instanceKeySet = new HashSet(); List<Instance> sourceInstances = sourceNamingService.getAllInstances(taskDO.getServiceName()); for (Instance instance : sourceInstances) { if (needSync(instance.getMetadata())) { destNamingService.registerInstance(taskDO.getServiceName(), buildSyncInstance(instance, taskDO)); instanceKeySet.add(composeInstanceKey(instance)); } } List<Instance> destInstances = destNamingService.getAllInstances(taskDO.getServiceName()); for (Instance instance : destInstances) { if (needDelete(instance.getMetadata(), taskDO) && !instanceKeySet.contains(composeInstanceKey(instance))) { destNamingService.deregisterInstance(taskDO.getServiceName(), instance.getIp(), instance.getPort()); } } } 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 nacos was failed, taskId:{}", taskDO.getTaskId(), e); metricsManager.recordError(MetricsStatisticsType.SYNC_ERROR); return false; } return true; }  @Override boolean delete(TaskDO taskDO); @Override boolean sync(TaskDO taskDO); Instance buildSyncInstance(Instance instance, TaskDO taskDO);  }