### Question:
KafkaAdminClient implements Closeable { public void updateTopicConfig(String topic, Properties properties) { if (StringUtils.isBlank(topic)) throw new IllegalArgumentException("topic cannot be null, empty or blank"); if (properties == null) throw new IllegalArgumentException("properties cannot be null"); LOG.debug("Updating topic config for topic [{}] with config [{}]", topic, properties); try { List<ConfigEntry> configEntries = new ArrayList<>(); for (String property : properties.stringPropertyNames()) { configEntries.add(new ConfigEntry(property, properties.getProperty(property))); } getNewAdminClient() .alterConfigs( Collections.singletonMap( new ConfigResource(ConfigResource.Type.TOPIC, topic), new Config(configEntries))) .all() .get(operationTimeout, TimeUnit.MILLISECONDS); } catch (InterruptedException | ExecutionException | TimeoutException e) { throw new AdminOperationException("Unable to update configuration for topic: " + topic, e); } } 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 updateTopicConfig() { Properties topicConfig = new Properties(); topicConfig.setProperty(LogConfig.MinInSyncReplicasProp(), "1"); client.createTopic(topic, 1, 1, topicConfig); assertThat(client.getTopicConfig(topic), is(topicConfig)); topicConfig.setProperty(LogConfig.RetentionMsProp(), "3600000"); client.updateTopicConfig(topic, topicConfig); assertThat(client.getTopicConfig(topic), is(topicConfig)); }
@Test (expected = IllegalArgumentException.class) public void updateTopicConfig_nullTopic() { client.updateTopicConfig(null, new Properties()); }
@Test (expected = IllegalArgumentException.class) public void updateTopicConfig_emptyTopic() { client.updateTopicConfig("", new Properties()); }
@Test (expected = IllegalArgumentException.class) public void updateTopicConfig_nullProperties() { client.updateTopicConfig(topic, null); }