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(); 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<>(); final Node partitionLeader = partitionInfo.leader(); final NodeDetails leaderNode = new NodeDetails( partitionLeader.id(), partitionLeader.host(), partitionLeader.port(), partitionLeader.rack() ); for (final Node node: partitionInfo.isr()) { isrNodes.add( new NodeDetails(node.id(), node.host(), node.port(), node.rack()) ); } for (final Node node: partitionInfo.replicas()) { replicaNodes.add( new NodeDetails(node.id(), node.host(), node.port(), node.rack()) ); } final PartitionDetails partitionDetail = new PartitionDetails( topicDescription.name(), partitionInfo.partition(), leaderNode, replicaNodes, isrNodes ); partitionDetails.add(partitionDetail); final TopicDetails topicDetails = new TopicDetails( topicDescription.name(), topicDescription.isInternal(), partitionDetails ); results.put(topic, topicDetails); } } return results; } catch (final ExecutionException e) { throw handleExecutionException(e); } catch (final InterruptedException exception) { throw new RuntimeException(exception.getMessage(), exception); } } KafkaOperations(final AdminClient adminClient, final KafkaConsumer<String, String> consumerClient); TopicList getAvailableTopics(); NodeList getClusterNodes(); Map<String, TopicDetails> getTopicDetails(final Collection<String> topics); TopicDetails getTopicDetails(final String topic); TopicConfig getTopicConfig(final String topic); BrokerConfig getBrokerConfig(final String brokerId); boolean createTopic(final CreateTopic createTopic); TopicConfig alterTopicConfig(final String topic, final Map<String, String> configItems); boolean removeTopic(final String topic); List<ConsumerGroupIdentifier> listConsumers(); boolean removeConsumerGroup(final String id); ConsumerGroupDetails getConsumerGroupDetails(final String consumerGroupId); List<ConsumerGroupDetails> getConsumerGroupDetails(final List<String> consumerGroupIds); ConsumerGroupOffsets getConsumerGroupOffsets(final String consumerGroupId); ConsumerGroupOffsetsWithTailPositions getConsumerGroupOffsetsWithTailOffsets(final String consumerGroupId); TailOffsets getTailOffsets(final String topic); TailOffsets getTailOffsets(final String topic, final Collection<Integer> partitions); void close();  }
@Test public void testGetTopicDetails() { final String[] brokerHostBits = sharedKafkaTestResource .getKafkaConnectString() .replaceAll("PLAINTEXT: .split(":"); final String brokerHost = brokerHostBits[0]; final int brokerPort = Integer.valueOf(brokerHostBits[1]); final String topic1 = "MyTopic1" + System.currentTimeMillis(); final String topic2 = "MyTopic2" + System.currentTimeMillis(); sharedKafkaTestResource .getKafkaTestUtils() .createTopic(topic1, 2, (short) 1); sharedKafkaTestResource .getKafkaTestUtils() .createTopic(topic2, 1, (short) 1); final TopicDetails topicDetails = kafkaOperations.getTopicDetails(topic1); assertEquals("Has correct name", topic1, topicDetails.getName()); assertEquals("Is not internal", false, topicDetails.isInternal()); assertEquals("Has 2 partitions", 2, topicDetails.getPartitions().size()); final PartitionDetails partition0 = topicDetails.getPartitions().get(0); assertEquals("Has correct topic name", topic1, partition0.getTopic()); assertEquals("Has correct partition", 0, partition0.getPartition()); assertEquals("Has one replica", 1, partition0.getReplicas().size()); assertEquals("Has one isr", 1, partition0.getIsr().size()); assertNotNull("Has a leader", partition0.getLeader()); validateNode(partition0.getLeader(), 1, brokerHost, brokerPort); validateNode(partition0.getIsr().get(0), 1, brokerHost, brokerPort); validateNode(partition0.getReplicas().get(0), 1, brokerHost, brokerPort); final PartitionDetails partition1 = topicDetails.getPartitions().get(1); assertEquals("Has correct topic name", topic1, partition1.getTopic()); assertEquals("Has correct partition", 1, partition1.getPartition()); assertEquals("Has one replica", 1, partition1.getReplicas().size()); assertEquals("Has one isr", 1, partition1.getIsr().size()); assertNotNull("Has a leader", partition1.getLeader()); validateNode(partition1.getLeader(), 1, brokerHost, brokerPort); validateNode(partition1.getIsr().get(0), 1, brokerHost, brokerPort); validateNode(partition1.getReplicas().get(0), 1, brokerHost, brokerPort); }