KafkaOperations implements AutoCloseable { public ConsumerGroupOffsets getConsumerGroupOffsets(final String consumerGroupId) { try { final ConsumerGroupOffsets.Builder builder = ConsumerGroupOffsets.newBuilder() .withConsumerId(consumerGroupId); final ListConsumerGroupOffsetsResult results = adminClient.listConsumerGroupOffsets(consumerGroupId); final Map<org.apache.kafka.common.TopicPartition, OffsetAndMetadata> partitionsToOffsets = results .partitionsToOffsetAndMetadata() .get(); for (final Map.Entry<org.apache.kafka.common.TopicPartition, OffsetAndMetadata> entry : partitionsToOffsets.entrySet()) { builder.withOffset( entry.getKey().topic(), entry.getKey().partition(), entry.getValue().offset() ); } return builder.build(); } 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 testGetConsumerGroupOffsets() { final String topicName = "AnotherTestTopic-" + System.currentTimeMillis(); sharedKafkaTestResource .getKafkaTestUtils() .createTopic(topicName, 2, (short) 1); sharedKafkaTestResource .getKafkaTestUtils() .produceRecords(10, topicName, 0); sharedKafkaTestResource .getKafkaTestUtils() .produceRecords(10, topicName, 1); final String consumerId1 = "ConsumerA-" + System.currentTimeMillis(); final String consumerPrefix = "TestConsumer"; final String finalConsumerId = consumerPrefix + "-" + consumerId1; try (final KafkaConsumer consumer = consumeFromTopic(topicName, consumerId1, consumerPrefix)) { final ConsumerGroupOffsets consumerGroupOffsets = kafkaOperations.getConsumerGroupOffsets(finalConsumerId); assertNotNull(consumerGroupOffsets); assertTrue(consumerGroupOffsets.getTopicNames().contains(topicName)); assertEquals(1, consumerGroupOffsets.getTopicNames().size()); ConsumerGroupTopicOffsets topicOffsets = consumerGroupOffsets.getOffsetsForTopic(topicName); assertEquals(topicName, topicOffsets.getTopic()); assertEquals(finalConsumerId, consumerGroupOffsets.getConsumerId()); assertEquals(2, topicOffsets.getOffsets().size()); assertEquals(10, topicOffsets.getOffsetForPartition(0)); assertEquals(10, topicOffsets.getOffsetForPartition(1)); final PartitionOffset offsetsPartition0 = topicOffsets.getOffsets().get(0); assertNotNull(offsetsPartition0); assertEquals(0, offsetsPartition0.getPartition()); assertEquals(10, offsetsPartition0.getOffset()); final PartitionOffset offsetsPartition1 = topicOffsets.getOffsets().get(1); assertNotNull(offsetsPartition1); assertEquals(1, offsetsPartition1.getPartition()); assertEquals(10, offsetsPartition1.getOffset()); } }