KafkaOperations implements AutoCloseable { public List<ConsumerGroupIdentifier> listConsumers() { final ListConsumerGroupsResult results = adminClient.listConsumerGroups(); final List<ConsumerGroupIdentifier> consumerIds = new ArrayList<>(); try { results .all() .get() .forEach((result) -> { consumerIds.add( new ConsumerGroupIdentifier(result.groupId(), result.isSimpleConsumerGroup()) ); }); consumerIds.sort(Comparator.comparing(ConsumerGroupIdentifier::getId)); return Collections.unmodifiableList(consumerIds); } 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 testListConsumers() { final String topicName = "AnotherTestTopic-" + System.currentTimeMillis(); sharedKafkaTestResource .getKafkaTestUtils() .createTopic(topicName, 1, (short) 1); sharedKafkaTestResource .getKafkaTestUtils() .produceRecords(10, topicName, 0); final String consumerId1 = "ConsumerA"; final String consumerId2 = "ConsumerB"; final String consumerPrefix = "TestConsumer"; consumeFromTopicAndClose(topicName, consumerId1, consumerPrefix); consumeFromTopicAndClose(topicName, consumerId2, consumerPrefix); final List<ConsumerGroupIdentifier> consumerIds = kafkaOperations.listConsumers(); assertTrue("Should have at least 2 consumers listed", consumerIds.size() >= 2); boolean foundGroup1 = false; boolean foundGroup2 = false; for (final ConsumerGroupIdentifier foundConsumerGroupIds : consumerIds) { if (foundConsumerGroupIds.getId().equals(consumerPrefix + "-" + consumerId1)) { foundGroup1 = true; } else if (foundConsumerGroupIds.getId().equals(consumerPrefix + "-" + consumerId2)) { foundGroup2 = true; } } assertTrue("Found consumer group 1", foundGroup1); assertTrue("Found consumer group 2", foundGroup2); }