@Test
public void testCreateNonSslConsumer() {

    // Create Cluster config
    final ClusterConfig clusterConfig = ClusterConfig.newBuilder()
        .withBrokerHosts(sharedKafkaTestResource.getKafkaConnectString())
        .build();
    // Create a topic
    final String topicName = "MyRandomTopic";
    sharedKafkaTestResource.getKafkaTestUtils().createTopic(topicName, 1, (short) 1);
    final KafkaAdminFactory kafkaAdminFactory = new KafkaAdminFactory(new KafkaClientConfigUtil("NotUsed", "Prefix"));
    // Create instance
    try (final KafkaConsumer<String, String> consumerClient = kafkaAdminFactory.createConsumer(clusterConfig, "MyClientId")) {
        // Call method to validate things work as expected
        final Map<String, List<PartitionInfo>> results = consumerClient.listTopics();
        assertNotNull(results);
        assertTrue(results.containsKey(topicName), "Should have our topic.");
    }
}