GH-2218 add ability to pass AdminClient

- Useful when you want to pass an AdminClient that has already
been initialised with specific properties to connect to the cluster
e.g. SSL properties.

Author: gurpiarbassi

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 static OffsetAndMetadata getCurrentOffset(String brokerAddresses, String
 		}
 	}
 
+	/**
+	 * 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.

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 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 void testMultiMinRecords(EmbeddedKafkaBroker broker) throws Exception {
 		consumer.close();
 	}
 
+	@Test
+	public void testGetCurrentOffsetWithAdminClient(EmbeddedKafkaBroker broker) throws Exception {
+		Map<String, Object> adminClientProps = Map.of(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, broker.getBrokersAsString());
+		Map<String, Object> producerProps = KafkaTestUtils.producerProps(broker);
+		try (AdminClient adminClient = AdminClient.create(adminClientProps); KafkaProducer<Integer, String> 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);
+		}
+
+	}
+
 }