@Test
public void testGetConsumerGroupOffsets() {

    // First need to create a topic.
    final String topicName = "AnotherTestTopic-" + System.currentTimeMillis();
    // Create topic
    sharedKafkaTestResource
        .getKafkaTestUtils()
        .createTopic(topicName, 2, (short) 1);
    // Publish data into the topic
    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;
    // Create consumer, consume from topic, keep alive.
    try (final KafkaConsumer consumer = consumeFromTopic(topicName, consumerId1, consumerPrefix)) {
        // Ask for list of offsets.
        final ConsumerGroupOffsets consumerGroupOffsets = kafkaOperations.getConsumerGroupOffsets(finalConsumerId);
        // We should have one
        assertNotNull(consumerGroupOffsets);
        // Validate bits
        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());
    }
}