@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();
    // Create two topics
    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());
    // Validate leader, isr, replica
    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());
    // Validate leader, isr, replica
    validateNode(partition1.getLeader(), 1, brokerHost, brokerPort);
    validateNode(partition1.getIsr().get(0), 1, brokerHost, brokerPort);
    validateNode(partition1.getReplicas().get(0), 1, brokerHost, brokerPort);
}