KafkaAdminFactory { public AdminClient create(final ClusterConfig clusterConfig, final String clientId) { final Map<String, Object> config = configUtil.applyCommonSettings(clusterConfig, clientId); return KafkaAdminClient.create(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 testCreateNonSslAdminClient() throws ExecutionException, InterruptedException { final ClusterConfig clusterConfig = ClusterConfig.newBuilder() .withBrokerHosts(sharedKafkaTestResource.getKafkaConnectString()) .build(); final KafkaAdminFactory kafkaAdminFactory = new KafkaAdminFactory( new KafkaClientConfigUtil("not/used", "MyPrefix") ); try (final AdminClient adminClient = kafkaAdminFactory.create(clusterConfig, "MyClientId")) { final DescribeClusterResult results = adminClient.describeCluster(); assertNotNull("Should have a non-null result", results); final Collection<Node> nodes = results.nodes().get(); assertNotNull("Should have non-null node result", nodes); assertFalse("Should have non-empty node", nodes.isEmpty()); } }