Add KafkaTestUtils methods using Duration for timeout (#2467)

* Add KafkaTestUtils methods using Duration for timeout

To adjust codebase to modern JVM (and Kafka API itself - e.g. poll(Duration)).
Old methods with "long timeout" are deprecated.

* Remove deprecated KafkaTestUtils methods using "long timeout"

Replaced by method with Duration for timeout.

Author: szpak

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

@@ -136,30 +136,30 @@ public static Map<String, Object> producerProps(String brokers) {
 	 * @param <V> the value type.
 	 * @return the record.
 	 * @throws IllegalStateException if exactly one record is not received.
-	 * @see #getSingleRecord(Consumer, String, long)
+	 * @see #getSingleRecord(Consumer, String, Duration)
 	 */
 	public static <K, V> ConsumerRecord<K, V> getSingleRecord(Consumer<K, V> consumer, String topic) {
-		return getSingleRecord(consumer, topic, 60000); // NOSONAR magic #
+		return getSingleRecord(consumer, topic, Duration.ofSeconds(60)); // NOSONAR magic #
 	}
 
 	/**
 	 * Poll the consumer, expecting a single record for the specified topic.
 	 * @param consumer the consumer.
 	 * @param topic the topic.
-	 * @param timeout max time in milliseconds to wait for records; forwarded to {@link Consumer#poll(long)}.
+	 * @param timeout max duration to wait for records; forwarded to {@link Consumer#poll(Duration)}.
 	 * @param <K> the key type.
 	 * @param <V> the value type.
 	 * @return the record.
 	 * @throws IllegalStateException if exactly one record is not received.
-	 * @since 2.0
+	 * @since 2.9.3
 	 */
-	public static <K, V> ConsumerRecord<K, V> getSingleRecord(Consumer<K, V> consumer, String topic, long timeout) {
-		long expire = System.currentTimeMillis() + timeout;
+	public static <K, V> ConsumerRecord<K, V> getSingleRecord(Consumer<K, V> consumer, String topic, Duration timeout) {
+		long expire = System.currentTimeMillis() + timeout.toMillis();
 		ConsumerRecords<K, V> received;
 		Iterator<ConsumerRecord<K, V>> iterator;
-		long remaining = timeout;
+		long remaining = timeout.toMillis();
 		do {
-			received = getRecords(consumer, remaining);
+			received = getRecords(consumer, Duration.ofMillis(remaining));
 			iterator = received.records(topic).iterator();
 			Map<TopicPartition, Long> reset = new HashMap<>();
 			received.forEach(rec -> {
@@ -197,12 +197,12 @@ public static <K, V> ConsumerRecord<K, V> getSingleRecord(Consumer<K, V> consume
 	 * @param commit commit offset after polling or not.
 	 * @param timeout the timeout.
 	 * @return the record or null if no record received.
-	 * @since 2.3
+	 * @since 2.9.3
 	 */
 	@Nullable
 	@SuppressWarnings({ "rawtypes", "unchecked" })
 	public static ConsumerRecord<?, ?> getOneRecord(String brokerAddresses, String group, String topic, int partition,
-			boolean seekToLast, boolean commit, long timeout) {
+			boolean seekToLast, boolean commit, Duration timeout) {
 
 		Map<String, Object> consumerConfig = consumerProps(brokerAddresses, group, "false");
 		consumerConfig.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 1);
@@ -215,7 +215,7 @@ public static <K, V> ConsumerRecord<K, V> getSingleRecord(Consumer<K, V> consume
 					consumer.seek(topicPart, consumer.position(topicPart) - 1);
 				}
 			}
-			ConsumerRecords<?, ?> records = consumer.poll(Duration.ofMillis(timeout));
+			ConsumerRecords<?, ?> records = consumer.poll(timeout);
 			ConsumerRecord<?, ?> record = records.count() == 1 ? records.iterator().next() : null;
 			if (record != null && commit) {
 				consumer.commitSync();
@@ -298,44 +298,41 @@ public static Map<TopicPartition, Long> getEndOffsets(Consumer<?, ?> consumer, S
 	 * @param <K> the key type.
 	 * @param <V> the value type.
 	 * @return the records.
-	 * @see #getRecords(Consumer, long)
+	 * @see #getRecords(Consumer, Duration)
 	 */
 	public static <K, V> ConsumerRecords<K, V> getRecords(Consumer<K, V> consumer) {
-		return getRecords(consumer, 60000); // NOSONAR magic #
+		return getRecords(consumer, Duration.ofSeconds(60)); // NOSONAR magic #
 	}
 
 	/**
 	 * Poll the consumer for records.
 	 * @param consumer the consumer.
-	 * @param timeout max time in milliseconds to wait for records; forwarded to
-	 * {@link Consumer#poll(long)}.
+	 * @param timeout max time in milliseconds to wait for records; forwarded to {@link Consumer#poll(Duration)}.
 	 * @param <K> the key type.
 	 * @param <V> the value type.
 	 * @return the records.
 	 * @throws IllegalStateException if the poll returns null (since 2.3.4).
-	 * @since 2.0
+	 * @since 2.9.3
 	 */
-	public static <K, V> ConsumerRecords<K, V> getRecords(Consumer<K, V> consumer, long timeout) {
+	public static <K, V> ConsumerRecords<K, V> getRecords(Consumer<K, V> consumer, Duration timeout) {
 		return getRecords(consumer, timeout, -1);
 	}
 
 	/**
 	 * Poll the consumer for records.
 	 * @param consumer the consumer.
-	 * @param timeout max time in milliseconds to wait for records; forwarded to
-	 * {@link Consumer#poll(long)}.
+	 * @param timeout max time in milliseconds to wait for records; forwarded to {@link Consumer#poll(Duration)}.
 	 * @param <K> the key type.
 	 * @param <V> the value type.
-	 * @param minRecords wait until the timeout or at least this number of receords are
-	 * received.
+	 * @param minRecords wait until the timeout or at least this number of records are received.
 	 * @return the records.
 	 * @throws IllegalStateException if the poll returns null.
-	 * @since 2.4.2
+	 * @since 2.9.3
 	 */
-	public static <K, V> ConsumerRecords<K, V> getRecords(Consumer<K, V> consumer, long timeout, int minRecords) {
+	public static <K, V> ConsumerRecords<K, V> getRecords(Consumer<K, V> consumer, Duration timeout, int minRecords) {
 		logger.debug("Polling...");
 		Map<TopicPartition, List<ConsumerRecord<K, V>>> records = new HashMap<>();
-		long remaining = timeout;
+		long remaining = timeout.toMillis();
 		int count = 0;
 		do {
 			long t1 = System.currentTimeMillis();
@@ -344,8 +341,7 @@ public static <K, V> ConsumerRecords<K, V> getRecords(Consumer<K, V> consumer, l
 					+ received.partitions().stream()
 					.flatMap(p -> received.records(p).stream())
 					// map to same format as send metadata toString()
-					.map(r -> r.topic() + "-" + r.partition() + "@" + r.offset())
-					.collect(Collectors.toList()));
+					.map(r -> r.topic() + "-" + r.partition() + "@" + r.offset()).toList());
 			if (received == null) {
 				throw new IllegalStateException("null received from consumer.poll()");
 			}

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

@@ -19,6 +19,7 @@
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.assertThatExceptionOfType;
 
+import java.time.Duration;
 import java.util.Map;
 
 import org.apache.kafka.clients.admin.AdminClient;
@@ -72,7 +73,7 @@ void testGetSingleWithMoreThatOneTopicRecordNotThereYet(EmbeddedKafkaBroker brok
 		broker.consumeFromEmbeddedTopics(consumer, "singleTopic4", "singleTopic5");
 		long t1 = System.currentTimeMillis();
 		assertThatExceptionOfType(IllegalStateException.class).isThrownBy(() ->
-			KafkaTestUtils.getSingleRecord(consumer, "singleTopic5", 2000L));
+			KafkaTestUtils.getSingleRecord(consumer, "singleTopic5", Duration.ofSeconds(2)));
 		assertThat(System.currentTimeMillis() - t1).isGreaterThanOrEqualTo(2000L);
 		producer.send(new ProducerRecord<>("singleTopic5", 1, "foo"));
 		producer.close();
@@ -92,14 +93,14 @@ public void testGetOneRecord(EmbeddedKafkaBroker broker) throws Exception {
 		producer.send(new ProducerRecord<>("singleTopic3", 0, 1, "foo"));
 		producer.close();
 		ConsumerRecord<?, ?> oneRecord = KafkaTestUtils.getOneRecord(broker.getBrokersAsString(), "getOne",
-				"singleTopic3", 0, false, true, 10_000L);
+				"singleTopic3", 0, false, true, Duration.ofSeconds(10));
 		assertThat(oneRecord.value()).isEqualTo("foo");
 		assertThat(KafkaTestUtils.getCurrentOffset(broker.getBrokersAsString(), "getOne", "singleTopic3", 0))
 				.isNotNull()
 				.extracting(omd -> omd.offset())
 				.isEqualTo(1L);
 		oneRecord = KafkaTestUtils.getOneRecord(broker.getBrokersAsString(), "getOne",
-				"singleTopic3", 0, true, true, 10_000L);
+				"singleTopic3", 0, true, true, Duration.ofSeconds(10));
 		assertThat(oneRecord.value()).isEqualTo("foo");
 		assertThat(KafkaTestUtils.getCurrentOffset(broker.getBrokersAsString(), "getOne", "singleTopic3", 0))
 				.isNotNull()
@@ -124,7 +125,7 @@ public void testMultiMinRecords(EmbeddedKafkaBroker broker) throws Exception {
 				Thread.currentThread().interrupt();
 			}
 		}).start();
-		ConsumerRecords<Integer, String> records = KafkaTestUtils.getRecords(consumer, 10_000L, 2);
+		ConsumerRecords<Integer, String> records = KafkaTestUtils.getRecords(consumer, Duration.ofSeconds(10), 2);
 		assertThat(records.count()).isEqualTo(2);
 		producer.close();
 		consumer.close();
@@ -138,7 +139,7 @@ public void testGetCurrentOffsetWithAdminClient(EmbeddedKafkaBroker broker) thro
 			producer.send(new ProducerRecord<>("singleTopic3", 0, 1, "foo"));
 
 			KafkaTestUtils.getOneRecord(broker.getBrokersAsString(), "testGetCurrentOffsetWithAdminClient",
-					"singleTopic3", 0, false, true, 10_000L);
+					"singleTopic3", 0, false, true, Duration.ofSeconds(10));
 			assertThat(KafkaTestUtils.getCurrentOffset(adminClient, "testGetCurrentOffsetWithAdminClient", "singleTopic3", 0))
 					.isNotNull()
 					.extracting(omd -> omd.offset())