### Question:
KafkaAdminClient implements Closeable { ZkUtils getZkUtils() { if (zkUtils == null) { Tuple2<ZkClient, ZkConnection> tuple; try { ZKConfig zkConfig = new ZKConfig(new VerifiableProperties(properties)); tuple = ZkUtils.createZkClientAndConnection(zkConfig.zkConnect(), zkConfig.zkSessionTimeoutMs(), zkConfig.zkConnectionTimeoutMs()); } catch (ZkException | ZooKeeperClientException e) { throw new AdminOperationException("Unable to create admin connection", e); } boolean isSecure = Boolean.valueOf(properties.getProperty(ZOOKEEPER_SECURE, DEFAULT_ZOOKEEPER_SECURE)); zkUtils = new ZkUtils(tuple._1(), tuple._2(), isSecure); } return zkUtils; } KafkaAdminClient(Properties properties); Set<String> getTopics(); Set<TopicAndPartition> getPartitions(); Set<TopicAndPartition> getPartitions(String topic); Authorizer getAuthorizer(); Map<Resource, Set<Acl>> getAcls(); Map<Resource, Set<Acl>> getAcls(KafkaPrincipal principal); Set<Acl> getAcls(Resource resource); void addAcls(Set<Acl> acls, Resource resource); void removeAcls(Set<Acl> acls, Resource resource); void createTopic(String topic, int partitions, int replicationFactor); void createTopic(String topic, int partitions, int replicationFactor, Properties topicConfig); void deleteTopic(String topic); Properties getTopicConfig(String topic); void updateTopicConfig(String topic, Properties properties); int getTopicReplicationFactor(String topic); int getTopicPartitions(String topic); void addTopicPartitions(String topic, int partitions); AdminClient.ConsumerGroupSummary getConsumerGroupSummary(String consumerGroup); Collection<AdminClient.ConsumerSummary> getConsumerGroupSummaries(String consumerGroup); Map<TopicPartition, String> getConsumerGroupAssignments(String consumerGroup); @Override void close(); static final String ZOOKEEPER_SECURE; static final String DEFAULT_ZOOKEEPER_SECURE; static final String OPERATION_TIMEOUT_MS; static final String DEFAULT_OPERATION_TIMEOUT_MS; static final String OPERATION_SLEEP_MS; static final String DEFAULT_OPERATION_SLEEP_MS; }### Answer:
@Test public void constructorProperties_alternateConfig() { properties.setProperty(KafkaAdminClient.OPERATION_TIMEOUT_MS, "123"); properties.setProperty(KafkaAdminClient.OPERATION_SLEEP_MS, "321"); properties.setProperty(KafkaAdminClient.ZOOKEEPER_SECURE, "true"); try (KafkaAdminClient adminClient = new KafkaAdminClient(properties)) { assertThat(adminClient.operationTimeout, is(123L)); assertThat(adminClient.operationSleep, is(321L)); assertThat(adminClient.getZkUtils().isSecure(), is(true)); } }