KafkaOperations implements AutoCloseable { public ConsumerGroupDetails getConsumerGroupDetails(final String consumerGroupId) { final List<ConsumerGroupDetails> consumerGroupDetails = getConsumerGroupDetails( Collections.singletonList(consumerGroupId) ); if (consumerGroupDetails.size() == 1) { return consumerGroupDetails.get(0); } throw new RuntimeException("Unable to locate consumer group id " + consumerGroupId); } 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 testGetConsumerGroupDetails() { final String topicName = "AnotherTestTopic-" + System.currentTimeMillis(); sharedKafkaTestResource .getKafkaTestUtils() .createTopic(topicName, 1, (short) 1); sharedKafkaTestResource .getKafkaTestUtils() .produceRecords(10, topicName, 0); final String consumerId1 = "ConsumerA-" + System.currentTimeMillis(); final String consumerPrefix = "TestConsumer"; final String finalConsumerGroupId = consumerPrefix + "-" + consumerId1; final String finalConsumerId = finalConsumerGroupId + "-" + Thread.currentThread().getId(); try (final KafkaConsumer consumer = consumeFromTopic(topicName, consumerId1, consumerPrefix)) { final ConsumerGroupDetails consumerGroupDetails = kafkaOperations.getConsumerGroupDetails(finalConsumerGroupId); assertNotNull(consumerGroupDetails); assertEquals(finalConsumerGroupId, consumerGroupDetails.getGroupId()); assertFalse(consumerGroupDetails.isSimple()); assertEquals("range", consumerGroupDetails.getPartitionAssignor()); assertEquals("Stable", consumerGroupDetails.getState()); assertEquals(1, consumerGroupDetails.getMembers().size()); final ConsumerGroupDetails.Member memberDetails = consumerGroupDetails.getMembers().get(0); assertNotNull(memberDetails); assertEquals("/127.0.0.1", memberDetails.getHost()); assertEquals(finalConsumerId, memberDetails.getClientId()); assertTrue(memberDetails.getMemberId().startsWith(finalConsumerId)); } }