class KafkaAdminClient implements Closeable {

    public void deleteTopic(String topic) {
        if (StringUtils.isBlank(topic))
            throw new IllegalArgumentException("topic cannot be null, empty or blank");

        LOG.debug("Deleting topic [{}]", topic);

        try {
            getNewAdminClient().deleteTopics(Collections.singleton(topic)).all().get(operationTimeout, TimeUnit.MILLISECONDS);

            long start = System.currentTimeMillis();
            boolean operationCompleted = false;

            do {
                LOG.debug("Sleeping for {} ms for delete topic operation to complete for topic [{}]", operationSleep, topic);
                try {
                    Thread.sleep(operationSleep);
                } catch (InterruptedException e) {
                    throw new AdminOperationException("Interrupted waiting for topic " + topic + " to be deleted", e);
                }
                operationCompleted = !topicExists(topic);
            } while (!operationCompleted && !operationTimedOut(start));

            if (!operationCompleted)
                throw new AdminOperationException("Timeout waiting for topic " + topic + " to be deleted");

        } catch (InterruptedException | ExecutionException | TimeoutException e) {
            Throwable throwable = e.getCause();
            if (throwable instanceof UnknownTopicOrPartitionException) {
                LOG.warn("Topic [{}] to be deleted was not found", topic, e);
            } else {
                throw new AdminOperationException("Unable to delete topic: " + topic, e);
            }
        }
    }

    public  KafkaAdminClient(Properties properties);

}