Spring/spring-kafka
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

GH-2218 add ability to pass AdminClient

Browse Source 
- 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.
This commit is contained in:
Gurps Bassi
2022-04-07 12:22:38 +01:00
committed by Gary Russell
parent 26de6d292e
commit e7db6d47cc
2 changed files with 38 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files

19
spring-kafka-test/src/main/java/org/springframework/kafka/test/utils/KafkaTestUtils.java
View File

@@ -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.

21
spring-kafka-test/src/test/java/org/springframework/kafka/test/utils/KafkaTestUtilsTests.java
View File

@@ -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<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);
}
}
}