UpgradeCatalog250 extends AbstractUpgradeCatalog { protected void updateKafkaConfigs() throws AmbariException { AmbariManagementController ambariManagementController = injector.getInstance(AmbariManagementController.class); Clusters clusters = ambariManagementController.getClusters(); if (clusters != null) { Map<String, Cluster> clusterMap = clusters.getClusters(); if (clusterMap != null && !clusterMap.isEmpty()) { for (final Cluster cluster : clusterMap.values()) { Config kafkaBrokerConfig = cluster.getDesiredConfigByType(KAFKA_BROKER); if (kafkaBrokerConfig != null) { Map<String, String> kafkaBrokerProperties = kafkaBrokerConfig.getProperties(); if (kafkaBrokerProperties != null && kafkaBrokerProperties.containsKey(KAFKA_TIMELINE_METRICS_HOST)) { LOG.info("Removing kafka.timeline.metrics.host from kafka-broker"); removeConfigurationPropertiesFromCluster(cluster, KAFKA_BROKER, Collections.singleton("kafka.timeline.metrics.host")); } } } } } } @Inject  UpgradeCatalog250(Injector injector); @Override String getTargetVersion(); @Override String getSourceVersion(); static final String COMPONENT_TABLE; static final String COMPONENT_VERSION_TABLE; static final String COMPONENT_VERSION_PK; static final String COMPONENT_VERSION_FK_COMPONENT; static final String COMPONENT_VERSION_FK_REPO_VERSION; }
@Test public void testKafkaUpdateConfigs() throws Exception { Map<String, String> oldProperties = new HashMap<String, String>() { { put("kafka.timeline.metrics.host", "{{metric_collector_host}}"); put("kafka.timeline.metrics.port", "{{metric_collector_port}}"); } }; Map<String, String> newProperties = new HashMap<String, String>() { { put("kafka.timeline.metrics.port", "{{metric_collector_port}}"); } }; EasyMockSupport easyMockSupport = new EasyMockSupport(); Config mockKafkaBroker = easyMockSupport.createNiceMock(Config.class); reset(clusters, cluster); expect(clusters.getClusters()).andReturn(new HashMap<String, Cluster>() {{ put("normal", cluster); }}).once(); expect(cluster.getDesiredConfigByType("kafka-broker")).andReturn(mockKafkaBroker).atLeastOnce(); expect(mockKafkaBroker.getProperties()).andReturn(oldProperties).anyTimes(); replay(clusters, mockKafkaBroker, cluster); AmbariManagementControllerImpl controller = createMockBuilder(AmbariManagementControllerImpl.class) .addMockedMethod("createConfiguration") .addMockedMethod("getClusters", new Class[]{}) .addMockedMethod("createConfig") .withConstructor(actionManager, clusters, injector) .createNiceMock(); Injector injector2 = easyMockSupport.createNiceMock(Injector.class); Capture<Map<String, String>> propertiesCapture = EasyMock.newCapture(); expect(injector2.getInstance(AmbariManagementController.class)).andReturn(controller).anyTimes(); expect(controller.getClusters()).andReturn(clusters).anyTimes(); expect(controller.createConfig(anyObject(Cluster.class), anyString(), capture(propertiesCapture), anyString(), EasyMock.<Map<String, Map<String, String>>>anyObject())).andReturn(config).once(); replay(controller, injector2); new UpgradeCatalog250(injector2).updateKafkaConfigs(); easyMockSupport.verifyAll(); Map<String, String> updatedProperties = propertiesCapture.getValue(); assertTrue(Maps.difference(newProperties, updatedProperties).areEqual()); }