### Question:
KafkaAdminClient implements Closeable { public void addTopicPartitions(String topic, int partitions) { if (StringUtils.isBlank(topic)) throw new IllegalArgumentException("topic cannot be null, empty or blank"); if (partitions <= 1) throw new IllegalArgumentException("partitions cannot be <= 1"); LOG.debug("Adding topic partitions for topic [{}] with partitions [{}]", topic, partitions); try { getNewAdminClient().createPartitions(Collections.singletonMap(topic, NewPartitions.increaseTo(partitions))) .all().get(operationTimeout, TimeUnit.MILLISECONDS); } catch (InterruptedException | ExecutionException | TimeoutException e) { Throwable throwable = e.getCause(); if (!(throwable instanceof InvalidPartitionsException)) { throw new AdminOperationException("Unable to add partitions to topic: " + topic, e); } } long start = System.currentTimeMillis(); boolean operationCompleted; do { LOG.debug("Sleeping for {} ms for add topic partition operation to complete for topic [{}]", operationSleep, topic); try { Thread.sleep(operationSleep); } catch (InterruptedException e) { throw new AdminOperationException("Interrupted waiting for partitions to be added to topic: " + topic, e); } operationCompleted = topicHasPartitions(topic, partitions); } while (!operationCompleted && !operationTimedOut(start)); if (!operationCompleted) throw new AdminOperationException("Timeout waiting for partitions to be added to topic: " + topic); } 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 addTopicPartitions() { client.createTopic(topic, 1, 1); assertThat(client.getTopicPartitions(topic), is(1)); client.addTopicPartitions(topic, 2); assertThat(client.getTopicPartitions(topic), is(2)); }
@Test (expected = IllegalArgumentException.class) public void addTopicPartitions_nullTopic() { client.addTopicPartitions(null, 1); }
@Test (expected = IllegalArgumentException.class) public void addTopicPartitions_emptyTopic() { client.addTopicPartitions("", 1); }
@Test (expected = IllegalArgumentException.class) public void addTopicPartitions_partitionsLessThanZero() { client.addTopicPartitions(topic, -1); }
@Test (expected = IllegalArgumentException.class) public void addTopicPartitions_partitionsIsZero() { client.addTopicPartitions(topic, 0); }
@Test (expected = IllegalArgumentException.class) public void addTopicPartitions_partitionsIsOne() { client.addTopicPartitions(topic, 1); }