@Test
public void testGetTailOffsets() {

    // 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(12, topicName, 1);
    // Ask for list of offsets for all partitions
    final TailOffsets allTailOffsets = kafkaOperations.getTailOffsets(topicName);
    // We should have one
    assertNotNull(allTailOffsets);
    // Validate bits
    assertEquals(topicName, allTailOffsets.getTopic());
    assertEquals(2, allTailOffsets.getPartitions().size());
    assertEquals(2, allTailOffsets.getOffsets().size());
    assertEquals(10, allTailOffsets.getOffsetForPartition(0));
    assertEquals(12, allTailOffsets.getOffsetForPartition(1));
    final PartitionOffset offsetsPartition0 = allTailOffsets.getOffsets().get(0);
    assertNotNull(offsetsPartition0);
    assertEquals(0, offsetsPartition0.getPartition());
    assertEquals(10, offsetsPartition0.getOffset());
    final PartitionOffset offsetsPartition1 = allTailOffsets.getOffsets().get(1);
    assertNotNull(offsetsPartition1);
    assertEquals(1, offsetsPartition1.getPartition());
    assertEquals(12, offsetsPartition1.getOffset());
    // Now ask for tail offsets for just one partition
    final TailOffsets partialTailOffsets = kafkaOperations.getTailOffsets(topicName, Collections.singletonList(1));
    assertNotNull(partialTailOffsets);
    // Validate bits
    assertEquals(topicName, partialTailOffsets.getTopic());
    assertEquals(1, partialTailOffsets.getPartitions().size());
    assertEquals(1, partialTailOffsets.getOffsets().size());
    assertEquals(12, partialTailOffsets.getOffsetForPartition(1));
    final PartitionOffset partialOffsetsPartition1 = partialTailOffsets.getOffsets().get(0);
    assertNotNull(partialOffsetsPartition1);
    assertEquals(1, partialOffsetsPartition1.getPartition());
    assertEquals(12, partialOffsetsPartition1.getOffset());
}