KafkaMetadataService implements AutoCloseable { public static KafkaMetadataService newInstance( EnvironmentService environmentService, Long clusterId, SecurityContext securityContext) throws ServiceConfigurationNotFoundException, IOException, ServiceNotFoundException, ServiceComponentNotFoundException { final KafkaZkConnection kafkaZkConnection = KafkaZkConnection.newInstance( getZkStringRaw(environmentService, clusterId, AMBARI_JSON_CONFIG_KAFKA_BROKER)); final ZookeeperClient zkCli = ZookeeperClient.newInstance(kafkaZkConnection); zkCli.start(); return new KafkaMetadataService(zkCli, kafkaZkConnection, securityContext, getKafkaBrokerComponent(environmentService, clusterId), getKafkaBrokers(environmentService, clusterId), getServiceConfig(environmentService, clusterId, AMBARI_JSON_CONFIG_KAFKA_BROKER), getServiceConfig(environmentService, clusterId, AMBARI_JSON_CONFIG_KAFKA_ENV)); } KafkaMetadataService(ZookeeperClient zkCli, KafkaZkConnection kafkaZkConnection,
                         SecurityContext securityContext, Component kafkaBroker, Collection<ComponentProcess> kafkaBrokerProcesses,
                         ServiceConfiguration brokerConfig, ServiceConfiguration kafkaEnvConfig); static KafkaMetadataService newInstance(
            EnvironmentService environmentService, Long clusterId, SecurityContext securityContext); KafkaBrokersInfo<HostPort> getBrokerHostPortFromStreamsJson(); KafkaBrokersInfo<String> getBrokerInfoFromZk(); KafkaBrokersInfo<KafkaBrokersInfo.BrokerId> getBrokerIdsFromZk(); KafkaTopics getTopicsFromZk(); KafkaBrokerListeners getKafkaBrokerListeners(); @Override void close(); KafkaZkConnection getKafkaZkConnection(); Keytabs getKeytabs(); Principals getPrincipals(); Security getSecurity(); String getKafkaServiceName();  }
@Test public void test_KafkaZkConnection_wellInitialized() throws Exception { List<ComponentProcess> brokerProcesses = getBrokerComponentProcesses(); Deencapsulation.setField(kafkaMetadataService, "kafkaBrokerProcesses", brokerProcesses); for (String zkStr : zkStrs) { for (String chRoot : chRoots) { final String zkStrRaw = zkStr + chRoot; LOG.debug("zookeeper.connect=" + zkStrRaw); KafkaMetadataService.KafkaZkConnection kafkaZkConnection = KafkaMetadataService.KafkaZkConnection.newInstance(zkStrRaw); Assert.assertEquals(zkStr, kafkaZkConnection.getZkString()); Assert.assertEquals(chRoot.isEmpty() ? expectedChrootPath.get(0) : expectedChrootPath.get(1), kafkaZkConnection.getChRoot()); } } }
@Test public void test_KafkaZkConnection_createPath() throws Exception { List<ComponentProcess> brokerProcesses = getBrokerComponentProcesses(); Deencapsulation.setField(kafkaMetadataService, "kafkaBrokerProcesses", brokerProcesses); for (String zkStr : zkStrs) { for (String chRoot : chRoots) { final String zkStrRaw = zkStr + chRoot; LOG.debug("zookeeper.connect=" + zkStrRaw); KafkaMetadataService.KafkaZkConnection kafkaZkConnection = KafkaMetadataService.KafkaZkConnection.newInstance(zkStrRaw); final String zkPath = kafkaZkConnection.buildZkRootPath(ZK_RELATIVE_PATH_KAFKA_BROKERS_IDS); Assert.assertEquals(chRoot.isEmpty() ? expectedBrokerIdPath.get(0) : expectedBrokerIdPath.get(1), zkPath); } } }