diff --git a/spring-kafka-test/src/main/java/org/springframework/kafka/test/utils/KafkaTestUtils.java b/spring-kafka-test/src/main/java/org/springframework/kafka/test/utils/KafkaTestUtils.java index 23cfce2f..ef01a0b0 100644 --- a/spring-kafka-test/src/main/java/org/springframework/kafka/test/utils/KafkaTestUtils.java +++ b/spring-kafka-test/src/main/java/org/springframework/kafka/test/utils/KafkaTestUtils.java @@ -1,5 +1,5 @@ /* - * Copyright 2016-2021 the original author or authors. + * Copyright 2016-2022 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -244,6 +244,23 @@ public final class KafkaTestUtils { } } + /** + * Get the current offset and metadata for the provided group/topic/partition. + * @param adminClient the AdminClient instance. + * @param group the group. + * @param topic the topic. + * @param partition the partition. + * @return the offset and metadata. + * @throws Exception if an exception occurs. + * @since 3.0 + */ + public static OffsetAndMetadata getCurrentOffset(AdminClient adminClient, String group, String topic, int partition) + throws Exception { // NOSONAR + + return adminClient.listConsumerGroupOffsets(group).partitionsToOffsetAndMetadata().get() // NOSONAR false positive + .get(new TopicPartition(topic, partition)); + } + /** * Return the end offsets of the requested topic/partitions * @param consumer the consumer. diff --git a/spring-kafka-test/src/test/java/org/springframework/kafka/test/utils/KafkaTestUtilsTests.java b/spring-kafka-test/src/test/java/org/springframework/kafka/test/utils/KafkaTestUtilsTests.java index adc9c6bf..5784eb15 100644 --- a/spring-kafka-test/src/test/java/org/springframework/kafka/test/utils/KafkaTestUtilsTests.java +++ b/spring-kafka-test/src/test/java/org/springframework/kafka/test/utils/KafkaTestUtilsTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2019-2020 the original author or authors. + * Copyright 2019-2022 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -21,6 +21,8 @@ import static org.assertj.core.api.Assertions.assertThatExceptionOfType; import java.util.Map; +import org.apache.kafka.clients.admin.AdminClient; +import org.apache.kafka.clients.consumer.ConsumerConfig; import org.apache.kafka.clients.consumer.ConsumerRecord; import org.apache.kafka.clients.consumer.ConsumerRecords; import org.apache.kafka.clients.consumer.KafkaConsumer; @@ -128,4 +130,21 @@ public class KafkaTestUtilsTests { consumer.close(); } + @Test + public void testGetCurrentOffsetWithAdminClient(EmbeddedKafkaBroker broker) throws Exception { + Map adminClientProps = Map.of(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, broker.getBrokersAsString()); + Map producerProps = KafkaTestUtils.producerProps(broker); + try (AdminClient adminClient = AdminClient.create(adminClientProps); KafkaProducer producer = new KafkaProducer<>(producerProps)) { + producer.send(new ProducerRecord<>("singleTopic3", 0, 1, "foo")); + + KafkaTestUtils.getOneRecord(broker.getBrokersAsString(), "testGetCurrentOffsetWithAdminClient", + "singleTopic3", 0, false, true, 10_000L); + assertThat(KafkaTestUtils.getCurrentOffset(adminClient, "testGetCurrentOffsetWithAdminClient", "singleTopic3", 0)) + .isNotNull() + .extracting(omd -> omd.offset()) + .isEqualTo(1L); + } + + } + }