import org.apache.kafka.clients.admin.AdminClientConfig; //导入依赖的package包/类
/**
 * Create a new AdminClient instance.
 * @param clusterConfig What cluster to connect to.
 * @param clientId What clientId to associate the connection with.
 */
public AdminClient create(final ClusterConfig clusterConfig, final String clientId) {
    // Create a map
    final Map<String, Object> config = new HashMap<>();
    config.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, clusterConfig.getConnectString());
    config.put(AdminClientConfig.CLIENT_ID_CONFIG, clientId);
    config.put(AdminClientConfig.REQUEST_TIMEOUT_MS_CONFIG, requestTimeout);

    if (clusterConfig.isUseSsl()) {
        config.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SSL");
        config.put(SslConfigs.SSL_KEYSTORE_LOCATION_CONFIG, keyStoreRootPath + "/" + clusterConfig.getKeyStoreFile());
        config.put(SslConfigs.SSL_KEYSTORE_PASSWORD_CONFIG, clusterConfig.getKeyStorePassword());
        config.put(SslConfigs.SSL_TRUSTSTORE_LOCATION_CONFIG, keyStoreRootPath + "/" + clusterConfig.getTrustStoreFile());
        config.put(SslConfigs.SSL_TRUSTSTORE_PASSWORD_CONFIG, clusterConfig.getTrustStorePassword());
    }

    return KafkaAdminClient.create(config);
}
 

import org.apache.kafka.clients.admin.AdminClientConfig; //导入依赖的package包/类
@Before
public void init() {
	String multiplier = System.getenv("KAFKA_TIMEOUT_MULTIPLIER");
	if (multiplier != null) {
		timeoutMultiplier = Double.parseDouble(multiplier);
	}

	BrokerAddress[] brokerAddresses = embeddedKafka.getBrokerAddresses();
	List<String> bAddresses = new ArrayList<>();
	for (BrokerAddress bAddress : brokerAddresses) {
		bAddresses.add(bAddress.toString());
	}
	String[] foo = new String[bAddresses.size()];

	Map<String, Object> adminConfigs = new HashMap<>();
	adminConfigs.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, bAddresses.toArray(foo)[0]);
	adminClient = AdminClient.create(adminConfigs);
}
 

import org.apache.kafka.clients.admin.AdminClientConfig; //导入依赖的package包/类
public KafkaTopicProvisioner(KafkaBinderConfigurationProperties kafkaBinderConfigurationProperties,
							KafkaProperties kafkaProperties) {
	Assert.isTrue(kafkaProperties != null, "KafkaProperties cannot be null");
	Map<String, Object> adminClientProperties = kafkaProperties.buildAdminProperties();
	String kafkaConnectionString = kafkaBinderConfigurationProperties.getKafkaConnectionString();

	if (ObjectUtils.isEmpty(adminClientProperties.get(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG))
			|| !kafkaConnectionString.equals(kafkaBinderConfigurationProperties.getDefaultKafkaConnectionString())) {
		adminClientProperties.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaConnectionString);
	}
	this.configurationProperties = kafkaBinderConfigurationProperties;
	this.adminClient = AdminClient.create(adminClientProperties);
}
 

import org.apache.kafka.clients.admin.AdminClientConfig; //导入依赖的package包/类
public Map<String, Object> getKsqlAdminClientConfigProps() {
  Set<String> adminClientConfigProperties = AdminClientConfig.configNames();
  Map<String, Object> adminClientConfigs = new HashMap<>();
  for (Map.Entry<String, Object> entry: ksqlStreamConfigProps.entrySet()) {
    if (adminClientConfigProperties.contains(entry.getKey())) {
      adminClientConfigs.put(entry.getKey(), entry.getValue());
    }
  }
  return adminClientConfigs;
}
 

import org.apache.kafka.clients.admin.AdminClientConfig; //导入依赖的package包/类
public AdminClient admin() {
    if (uri == null) throw new Error("uri must not be null");

    Map<String, Object> config = Maps.newHashMap();
    config.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, uri);
    return AdminClient.create(config);
}
 

import org.apache.kafka.clients.admin.AdminClientConfig; //导入依赖的package包/类
@Test(enabled = false)
public void testListTopics() throws Exception {
    Map<String, Object> adminClientConfig = new HashMap<>();
    adminClientConfig.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, "api.local.getbux.com:32400,api.local.getbux.com:32401,api.local.getbux.com:32402");
    adminClientConfig.put(AdminClientConfig.CLIENT_ID_CONFIG, "test-adminClient");
    AdminClient adminClient = AdminClient.create(adminClientConfig);

    // ensure all the topics are created
    ListTopicsResult listTopicsResult = adminClient.listTopics();
    Map<String, TopicListing> availableTopics = listTopicsResult.namesToListings().get();

    assertNotNull(availableTopics);
}
 

import org.apache.kafka.clients.admin.AdminClientConfig; //导入依赖的package包/类
private String bootstrapServers() {
    Object servers = adminConfig.get(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG);
    return servers != null ? servers.toString() : "<unknown>";
}