UpgradeCatalog240 extends AbstractUpgradeCatalog { protected void updateKAFKAConfigs() throws AmbariException { AmbariManagementController ambariManagementController = injector.getInstance(AmbariManagementController.class); Clusters clusters = ambariManagementController.getClusters(); if (clusters != null) { Map<String, Cluster> clusterMap = getCheckedClusterMap(clusters); if (clusterMap != null && !clusterMap.isEmpty()) { for (final Cluster cluster : clusterMap.values()) { Set<String> installedServices = cluster.getServices().keySet(); if (installedServices.contains("KAFKA") && cluster.getSecurityType() == SecurityType.KERBEROS) { Config kafkaBroker = cluster.getDesiredConfigByType(KAFKA_BROKER_CONFIG); if (kafkaBroker != null) { String listenersPropertyValue = kafkaBroker.getProperties().get("listeners"); if (StringUtils.isNotEmpty(listenersPropertyValue)) { String newListenersPropertyValue = listenersPropertyValue.replaceAll("\\bPLAINTEXT\\b", "PLAINTEXTSASL"); updateConfigurationProperties(KAFKA_BROKER_CONFIG, Collections.singletonMap("listeners", newListenersPropertyValue), true, false); } } } } } } } @Inject  UpgradeCatalog240(Injector injector); @Override String getTargetVersion(); @Override String getSourceVersion(); static final String DESIRED_VERSION_COLUMN_NAME; static final String BLUEPRINT_SETTING_TABLE; static final String BLUEPRINT_NAME_COL; static final String SETTING_NAME_COL; static final String SETTING_DATA_COL; static final String ID; static final String BLUEPRINT_TABLE; static final String VIEWINSTANCE_TABLE; static final String SHORT_URL_COLUMN; static final String CLUSTER_HANDLE_COLUMN; static final String REQUESTSCHEDULE_TABLE; static final String AUTHENTICATED_USER_ID_COLUMN; static final String TIMELINE_METRICS_SINK_COLLECTION_PERIOD; static final String ONE_DIR_PER_PARITION_PROPERTY; static final String VIEWURL_TABLE; static final String URL_ID_COLUMN; static final String PRECISION_TABLE_TTL_PROPERTY; static final String CLUSTER_SECOND_TABLE_TTL_PROPERTY; static final String CLUSTER_TYPE_COLUMN; static final String REMOTE_AMBARI_CLUSTER_TABLE; static final String REMOTE_AMBARI_CLUSTER_SERVICE_TABLE; static final String CLUSTER_ID; static final String SERVICE_NAME; static final String CLUSTER_NAME; }
@Test public void testUpdateKAFKAConfigs() throws Exception{ EasyMockSupport easyMockSupport = new EasyMockSupport(); final AmbariManagementController mockAmbariManagementController = easyMockSupport.createNiceMock(AmbariManagementController.class); final Clusters mockClusters = easyMockSupport.createStrictMock(Clusters.class); final Cluster mockClusterExpected = easyMockSupport.createNiceMock(Cluster.class); final Config kafkaBroker = easyMockSupport.createNiceMock(Config.class); expect(kafkaBroker.getProperties()).andReturn(new HashMap<String, String>(){{ put("listeners", "PLAINTEXT: }} ).anyTimes(); final Injector mockInjector = Guice.createInjector(new AbstractModule() { @Override protected void configure() { bind(AmbariManagementController.class).toInstance(mockAmbariManagementController); bind(Clusters.class).toInstance(mockClusters); bind(EntityManager.class).toInstance(entityManager); bind(DBAccessor.class).toInstance(createNiceMock(DBAccessor.class)); bind(OsFamily.class).toInstance(createNiceMock(OsFamily.class)); bind(PasswordEncoder.class).toInstance(createNiceMock(PasswordEncoder.class)); bind(HookContextFactory.class).toInstance(createMock(HookContextFactory.class)); bind(HookService.class).toInstance(createMock(HookService.class)); } }); expect(mockAmbariManagementController.getClusters()).andReturn(mockClusters).once(); expect(mockClusters.getClusters()).andReturn(new HashMap<String, Cluster>() {{ put("normal", mockClusterExpected); }}).atLeastOnce(); expect(mockClusterExpected.getDesiredConfigByType("kafka-broker")).andReturn(kafkaBroker).atLeastOnce(); expect(mockClusterExpected.getSecurityType()).andReturn(SecurityType.KERBEROS); expect(mockClusterExpected.getServices()).andReturn(new HashMap<String, Service>() { { put("KAFKA", null); } }).atLeastOnce(); UpgradeCatalog240 upgradeCatalog240 = createMockBuilder(UpgradeCatalog240.class) .withConstructor(Injector.class) .withArgs(mockInjector) .addMockedMethod("updateConfigurationProperties", String.class, Map.class, boolean.class, boolean.class) .createMock(); Map<String, String> expectedUpdates = new HashMap<>(); expectedUpdates.put("listeners", "PLAINTEXTSASL: upgradeCatalog240.updateConfigurationProperties("kafka-broker", expectedUpdates, true, false); expectLastCall().once(); easyMockSupport.replayAll(); replay(upgradeCatalog240); upgradeCatalog240.updateKAFKAConfigs(); easyMockSupport.verifyAll(); }