KafkaOperations implements AutoCloseable { public TailOffsets getTailOffsets(final String topic) { final TopicDetails topicDetails = getTopicDetails(topic); final List<Integer> partitions = topicDetails.getPartitions() .stream() .map(PartitionDetails::getPartition) .collect(Collectors.toList()); return getTailOffsets(topic, partitions); } 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 testGetTailOffsets() { final String topicName = "AnotherTestTopic-" + System.currentTimeMillis(); sharedKafkaTestResource .getKafkaTestUtils() .createTopic(topicName, 2, (short) 1); sharedKafkaTestResource .getKafkaTestUtils() .produceRecords(10, topicName, 0); sharedKafkaTestResource .getKafkaTestUtils() .produceRecords(12, topicName, 1); final TailOffsets allTailOffsets = kafkaOperations.getTailOffsets(topicName); assertNotNull(allTailOffsets); 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()); final TailOffsets partialTailOffsets = kafkaOperations.getTailOffsets(topicName, Collections.singletonList(1)); assertNotNull(partialTailOffsets); 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()); }