### Question:
UpgradeCatalog251 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"); if(!newListenersPropertyValue.equals(listenersPropertyValue)) { updateConfigurationProperties(KAFKA_BROKER_CONFIG, Collections.singletonMap("listeners", newListenersPropertyValue), true, false); } } } } } } } } @Inject  UpgradeCatalog251(Injector injector); @Override String getSourceVersion(); @Override String getTargetVersion();  }### Answer:
@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); Map<String, String> initialProperties = Collections.singletonMap("listeners", "PLAINTEXT: Map<String, String> expectedUpdates = Collections.singletonMap("listeners", "PLAINTEXTSASL: final Config kafkaBroker = easyMockSupport.createNiceMock(Config.class); expect(kafkaBroker.getProperties()).andReturn(initialProperties).times(1); expect(kafkaBroker.getProperties()).andReturn(expectedUpdates).times(1); final Injector mockInjector = getInjector(mockClusters, mockAmbariManagementController); expect(mockAmbariManagementController.getClusters()).andReturn(mockClusters).atLeastOnce(); expect(mockClusters.getClusters()).andReturn(Collections.singletonMap("normal", mockClusterExpected)).atLeastOnce(); expect(mockClusterExpected.getDesiredConfigByType("kafka-broker")).andReturn(kafkaBroker).atLeastOnce(); expect(mockClusterExpected.getSecurityType()).andReturn(SecurityType.KERBEROS).atLeastOnce(); expect(mockClusterExpected.getServices()).andReturn(Collections.singletonMap("KAFKA", null)).atLeastOnce(); UpgradeCatalog251 upgradeCatalog251 = createMockBuilder(UpgradeCatalog251.class) .withConstructor(Injector.class) .withArgs(mockInjector) .addMockedMethod("updateConfigurationProperties", String.class, Map.class, boolean.class, boolean.class) .createMock(); upgradeCatalog251.updateConfigurationProperties("kafka-broker", expectedUpdates, true, false); expectLastCall().once(); easyMockSupport.replayAll(); replay(upgradeCatalog251); upgradeCatalog251.updateKAFKAConfigs(); upgradeCatalog251.updateKAFKAConfigs(); easyMockSupport.verifyAll(); }