KafkaOperationsFactory { public KafkaOperations create(final Cluster cluster, final long userId) { final String clientId = getClientId(userId); final ClusterConfig clusterConfig = ClusterConfig.newBuilder(cluster, secretManager).build(); final AdminClient adminClient = kafkaAdminFactory.create(clusterConfig, clientId); final KafkaConsumer<String, String> kafkaConsumer = kafkaAdminFactory.createConsumer(clusterConfig, clientId); return new KafkaOperations(adminClient, kafkaConsumer); } KafkaOperationsFactory(final SecretManager secretManager, final KafkaAdminFactory kafkaAdminFactory); KafkaOperations create(final Cluster cluster, final long userId); Map<String, Object> getConsumerConfig(final Cluster cluster, final long userId);  }
@Test public void smokeTestNonSslOperationsClient() { final SecretManager secretManager = new SecretManager("notused"); final KafkaClientConfigUtil configUtil = new KafkaClientConfigUtil("Not-Used", "Test-Prefix"); final KafkaAdminFactory kafkaAdminFactory = new KafkaAdminFactory(configUtil); final KafkaOperationsFactory operationsFactory = new KafkaOperationsFactory(secretManager, kafkaAdminFactory); final Cluster cluster = new Cluster(); cluster.setBrokerHosts(sharedKafkaTestResource.getKafkaConnectString()); cluster.setSslEnabled(false); try (final KafkaOperations kafkaOperations = operationsFactory.create(cluster, 1L)) { final NodeList nodeList = kafkaOperations.getClusterNodes(); assertNotNull("Should have a non-null result", nodeList); assertFalse("Should have non-empty node", nodeList.getNodes().isEmpty()); } }