HostVersionOutOfSyncListener { @Subscribe @Transactional public void onServiceEvent(ServiceInstalledEvent event) { if (LOG.isDebugEnabled()) { LOG.debug(event.toString()); } try { Cluster cluster = clusters.get().getClusterById(event.getClusterId()); Set<RepositoryVersionEntity> changedRepositoryVersions = new HashSet<RepositoryVersionEntity>(); Map<String, ServiceComponent> serviceComponents = cluster.getService(event.getServiceName()).getServiceComponents(); Map<String, List<ServiceComponent>> affectedHosts = new HashMap<String, List<ServiceComponent>>(); for (ServiceComponent component : serviceComponents.values()) { for (String hostname : component.getServiceComponentHosts().keySet()) { if (! affectedHosts.containsKey(hostname)) { affectedHosts.put(hostname, new ArrayList<ServiceComponent>()); } affectedHosts.get(hostname).add(component); } } for (String hostName : affectedHosts.keySet()) { List<HostVersionEntity> hostVersionEntities = hostVersionDAO.get().findByClusterAndHost(cluster.getClusterName(), hostName); for (HostVersionEntity hostVersionEntity : hostVersionEntities) { RepositoryVersionEntity repositoryVersion = hostVersionEntity.getRepositoryVersion(); boolean hasChangedComponentsWithVersions = false; String serviceName = event.getServiceName(); for (ServiceComponent comp : affectedHosts.get(hostName)) { String componentName = comp.getName(); ComponentInfo component = ami.get().getComponent(repositoryVersion.getStackName(), repositoryVersion.getStackVersion(), serviceName, componentName); if (component.isVersionAdvertised()) { hasChangedComponentsWithVersions = true; } } if (! hasChangedComponentsWithVersions) { continue; } if (hostVersionEntity.getState().equals(RepositoryVersionState.INSTALLED)) { hostVersionEntity.setState(RepositoryVersionState.OUT_OF_SYNC); hostVersionDAO.get().merge(hostVersionEntity); changedRepositoryVersions.add(repositoryVersion); } } } for (RepositoryVersionEntity repositoryVersion : changedRepositoryVersions) { cluster.recalculateClusterVersionState(repositoryVersion); } } catch (AmbariException e) { LOG.error("Can not update hosts about out of sync", e); } } @Inject  HostVersionOutOfSyncListener(AmbariEventPublisher ambariEventPublisher, LockFactory lockFactory); @Subscribe @Transactional void onServiceComponentEvent(ServiceComponentInstalledEvent event); @Subscribe @Transactional void onServiceComponentHostEvent(ServiceComponentUninstalledEvent event); @Subscribe @Transactional void onServiceEvent(ServiceInstalledEvent event); @Subscribe @Transactional void onHostEvent(HostsAddedEvent event); @Subscribe @Transactional void onHostEvent(HostsRemovedEvent event);  }
@Test public void testOnServiceEvent() throws AmbariException { String INSTALLED_VERSION = "2.2.0-1000"; String INSTALLED_VERSION_2 = "2.1.1-2000"; StackId stackId = new StackId(this.stackId); StackId yaStackId = new StackId(yetAnotherStackId); createClusterAndHosts(INSTALLED_VERSION, stackId); addRepoVersion(INSTALLED_VERSION_2, yaStackId); List<String> hostList = new ArrayList<String>(); hostList.add("h1"); hostList.add("h2"); hostList.add("h3"); Map<String, List<Integer>> hdfsTopology = new HashMap<String, List<Integer>>(); hdfsTopology.put("NAMENODE", Collections.singletonList(0)); hdfsTopology.put("SECONDARY_NAMENODE", Collections.singletonList(1)); List<Integer> datanodeHosts = Arrays.asList(0, 1); hdfsTopology.put("DATANODE", new ArrayList<Integer>(datanodeHosts)); addService(c1, hostList, hdfsTopology, "HDFS"); Set<String> changedHosts = new HashSet<String>(); changedHosts.add("h1"); changedHosts.add("h2"); List<HostVersionEntity> hostVersions = hostVersionDAO.findAll(); assertRepoVersionState(stackId.getStackId(), INSTALLED_VERSION, RepositoryVersionState.INSTALLED); assertRepoVersionState(yaStackId.getStackId(), INSTALLED_VERSION_2, RepositoryVersionState.INSTALLED); for (HostVersionEntity hostVersionEntity : hostVersions) { if (hostVersionEntity.getRepositoryVersion().getVersion().equals(INSTALLED_VERSION) || hostVersionEntity.getRepositoryVersion().getVersion().equals(INSTALLED_VERSION_2)) { if (changedHosts.contains(hostVersionEntity.getHostName())) { assertEquals(hostVersionEntity.getState(), RepositoryVersionState.OUT_OF_SYNC); } else { assertEquals(hostVersionEntity.getState(), RepositoryVersionState.INSTALLED); } } } }