class 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();
            // This will be thrown if the topic already has the specified number of partitions, ignore
            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);
    }

    public  KafkaAdminClient(Properties properties);

}