class KafkaOperations implements AutoCloseable {

    public Map<String, TopicDetails> getTopicDetails(final Collection<String> topics) {
        final Map<String, TopicDetails> results = new HashMap<>();

        try {
            final Map<String, TopicDescription> topicDescriptionMap = adminClient.describeTopics(topics).all().get();

            // Now parse results
            for (final Map.Entry<String, TopicDescription> entry: topicDescriptionMap.entrySet()) {
                final String topic = entry.getKey();
                final TopicDescription topicDescription = entry.getValue();

                final List<PartitionDetails> partitionDetails = new ArrayList<>();

                for (final TopicPartitionInfo partitionInfo: topicDescription.partitions()) {
                    final List<NodeDetails> isrNodes = new ArrayList<>();
                    final List<NodeDetails> replicaNodes = new ArrayList<>();

                    // Translate Leader
                    final Node partitionLeader = partitionInfo.leader();
                    final NodeDetails leaderNode = new NodeDetails(
                        partitionLeader.id(), partitionLeader.host(), partitionLeader.port(), partitionLeader.rack()
                    );

                    // Translate ISR nodes
                    for (final Node node: partitionInfo.isr()) {
                        isrNodes.add(
                            new NodeDetails(node.id(), node.host(), node.port(), node.rack())
                        );
                    }

                    // Translate Replicas nodes
                    for (final Node node: partitionInfo.replicas()) {
                        replicaNodes.add(
                            new NodeDetails(node.id(), node.host(), node.port(), node.rack())
                        );
                    }

                    // Create the details
                    final PartitionDetails partitionDetail = new PartitionDetails(
                        topicDescription.name(),
                        partitionInfo.partition(),
                        leaderNode,
                        replicaNodes,
                        isrNodes
                    );

                    // Add to the list.
                    partitionDetails.add(partitionDetail);

                    // Create new TopicDetails.
                    final TopicDetails topicDetails = new TopicDetails(
                        topicDescription.name(),
                        topicDescription.isInternal(),
                        partitionDetails
                    );
                    results.put(topic, topicDetails);
                }
            }
            // Return it
            return results;
        } catch (final ExecutionException e) {
            throw handleExecutionException(e);
        } catch (final InterruptedException exception) {
            // TODO Handle this
            throw new RuntimeException(exception.getMessage(), exception);
        }
    }

    public  KafkaOperations(final AdminClient adminClient, final KafkaConsumer<String, String> consumerClient);

}