KafkaAdminFactory { public KafkaConsumer<String, String> createConsumer(final ClusterConfig clusterConfig, final String clientId) { final Map<String, Object> config = getConsumerConfig(clusterConfig, clientId); return new KafkaConsumer<>(config); } KafkaAdminFactory(final KafkaClientConfigUtil configUtil); AdminClient create(final ClusterConfig clusterConfig, final String clientId); KafkaConsumer<String, String> createConsumer(final ClusterConfig clusterConfig, final String clientId); Map<String, Object> getConsumerConfig(final ClusterConfig clusterConfig, final String clientId);  }
@Test public void testCreateNonSslConsumer() { final ClusterConfig clusterConfig = ClusterConfig.newBuilder() .withBrokerHosts(sharedKafkaTestResource.getKafkaConnectString()) .build(); final String topicName = "MyRandomTopic"; sharedKafkaTestResource.getKafkaTestUtils().createTopic(topicName, 1, (short) 1); final KafkaAdminFactory kafkaAdminFactory = new KafkaAdminFactory(new KafkaClientConfigUtil("NotUsed", "Prefix")); try (final KafkaConsumer<String, String> consumerClient = kafkaAdminFactory.createConsumer(clusterConfig, "MyClientId")) { final Map<String, List<PartitionInfo>> results = consumerClient.listTopics(); assertNotNull(results); assertTrue(results.containsKey(topicName), "Should have our topic."); } }