KafkaOperations implements AutoCloseable { public NodeList getClusterNodes() { final List<NodeDetails> nodeDetails = new ArrayList<>(); try { final Collection<Node> nodes = adminClient.describeCluster().nodes().get(); for (final Node node : nodes) { nodeDetails.add(new NodeDetails(node.id(), node.host(), node.port(), node.rack())); } return new NodeList(nodeDetails); } catch (final ExecutionException e) { throw handleExecutionException(e); } catch (final InterruptedException e) { throw new RuntimeException(e.getMessage(), e); } } KafkaOperations(final AdminClient adminClient, final KafkaConsumer<String, String> consumerClient); TopicList getAvailableTopics(); NodeList getClusterNodes(); Map<String, TopicDetails> getTopicDetails(final Collection<String> topics); TopicDetails getTopicDetails(final String topic); TopicConfig getTopicConfig(final String topic); BrokerConfig getBrokerConfig(final String brokerId); boolean createTopic(final CreateTopic createTopic); TopicConfig alterTopicConfig(final String topic, final Map<String, String> configItems); boolean removeTopic(final String topic); List<ConsumerGroupIdentifier> listConsumers(); boolean removeConsumerGroup(final String id); ConsumerGroupDetails getConsumerGroupDetails(final String consumerGroupId); List<ConsumerGroupDetails> getConsumerGroupDetails(final List<String> consumerGroupIds); ConsumerGroupOffsets getConsumerGroupOffsets(final String consumerGroupId); ConsumerGroupOffsetsWithTailPositions getConsumerGroupOffsetsWithTailOffsets(final String consumerGroupId); TailOffsets getTailOffsets(final String topic); TailOffsets getTailOffsets(final String topic, final Collection<Integer> partitions); void close();  }
@Test public void testGetClusterNodes() { final String[] brokerHostBits = sharedKafkaTestResource .getKafkaConnectString() .replaceAll("PLAINTEXT: .split(":"); final String brokerHost = brokerHostBits[0]; final int brokerPort = Integer.valueOf(brokerHostBits[1]); final NodeList nodeList = kafkaOperations.getClusterNodes(); logger.info("{}", nodeList); assertEquals("Should have single node", 1, nodeList.getNodes().size()); final NodeDetails node = nodeList.getNodes().get(0); validateNode(node, 1, brokerHost, brokerPort); }