StackVersionListener { @Subscribe @AllowConcurrentEvents public void onAmbariEvent(HostComponentVersionAdvertisedEvent event) { LOG.debug("Received event {}", event); Cluster cluster = event.getCluster(); ServiceComponentHost sch = event.getServiceComponentHost(); String newVersion = event.getVersion(); if (StringUtils.isEmpty(newVersion)) { return; } m_stackVersionLock.lock(); if (null != event.getRepositoryVersionId() && null == cluster.getUpgradeInProgress()) { RepositoryVersionEntity rve = repositoryVersionDAO.findByPK(event.getRepositoryVersionId()); if (null != rve) { String currentRepoVersion = rve.getVersion(); if (!StringUtils.equals(currentRepoVersion, newVersion)) { rve.setVersion(newVersion); repositoryVersionDAO.merge(rve); } } } try { AmbariMetaInfo metaInfo = ambariMetaInfo.get(); ComponentInfo componentInfo = metaInfo.getComponent(cluster.getDesiredStackVersion().getStackName(), cluster.getDesiredStackVersion().getStackVersion(), sch.getServiceName(), sch.getServiceComponentName()); ServiceComponent sc = cluster.getService(sch.getServiceName()).getServiceComponent(sch.getServiceComponentName()); if (componentInfo.isVersionAdvertised() && StringUtils.isNotBlank(newVersion) && !UNKNOWN_VERSION.equalsIgnoreCase(newVersion)) { processComponentAdvertisedVersion(cluster, sch, newVersion, sc); } else if(!sc.isVersionAdvertised() && StringUtils.isNotBlank(newVersion) && !UNKNOWN_VERSION.equalsIgnoreCase(newVersion)) { LOG.debug("ServiceComponent {} doesn't advertise version, " + "however ServiceHostComponent {} on host {} advertised version as {}. Skipping version update", sc.getName(), sch.getServiceComponentName(), sch.getHostName(), newVersion); } else { if (UNKNOWN_VERSION.equals(sc.getDesiredVersion())) { processUnknownDesiredVersion(cluster, sc, sch, newVersion); } else { processComponentAdvertisedVersion(cluster, sch, newVersion, sc); } } } catch (Exception e) { LOG.error( "Unable to propagate version for ServiceHostComponent on component: {}, host: {}. Error: {}", sch.getServiceComponentName(), sch.getHostName(), e.getMessage()); } finally { m_stackVersionLock.unlock(); } } @Inject  StackVersionListener(VersionEventPublisher eventPublisher); @Subscribe @AllowConcurrentEvents void onAmbariEvent(HostComponentVersionAdvertisedEvent event); static final String UNKNOWN_VERSION; }
@Test public void testSetRepositoryVersion() throws Exception { expect(sch.getVersion()).andReturn(UNKNOWN_VERSION); expect(serviceComponent.isVersionAdvertised()).andReturn(Boolean.TRUE); RepositoryVersionDAO dao = createNiceMock(RepositoryVersionDAO.class); RepositoryVersionEntity entity = createNiceMock(RepositoryVersionEntity.class); expect(entity.getVersion()).andReturn("2.4.0.0").once(); expect(dao.findByPK(1L)).andReturn(entity).once(); expect(dao.merge(entity)).andReturn(entity).once(); replayAll(); String newVersion = VALID_NEW_VERSION; HostComponentVersionAdvertisedEvent event = new HostComponentVersionAdvertisedEvent(cluster, sch, newVersion, 1L); Field field = StackVersionListener.class.getDeclaredField("repositoryVersionDAO"); field.setAccessible(true); field.set(listener, dao); listener.onAmbariEvent(event); verifyAll(); }
@Test public void testRepositoryVersionNotSetDuringUpgrade() throws Exception { expect(cluster.getUpgradeInProgress()).andReturn(createNiceMock(UpgradeEntity.class)); RepositoryVersionDAO dao = createStrictMock(RepositoryVersionDAO.class); replayAll(); Field field = StackVersionListener.class.getDeclaredField("repositoryVersionDAO"); field.setAccessible(true); field.set(listener, dao); HostComponentVersionAdvertisedEvent event = new HostComponentVersionAdvertisedEvent(cluster, sch, VALID_NEW_VERSION, 1L); Assert.assertNotNull(event.getRepositoryVersionId()); listener.onAmbariEvent(event); verifyAll(); }