Add KafkaTestUtils methods using Duration for timeout (2.9) (#2468)

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

* Fix checkstyle error breaking the build

* Remove annotation elements not compatible with JDK 8

Author: szpak

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

@@ -136,10 +136,10 @@ 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 #
 	}
 
 	/**
@@ -152,14 +152,31 @@ public static <K, V> ConsumerRecord<K, V> getSingleRecord(Consumer<K, V> consume
 	 * @return the record.
 	 * @throws IllegalStateException if exactly one record is not received.
 	 * @since 2.0
+	 * @deprecated in favor of {@link #getSingleRecord(Consumer, String, Duration)}
 	 */
+	@Deprecated
 	public static <K, V> ConsumerRecord<K, V> getSingleRecord(Consumer<K, V> consumer, String topic, long timeout) {
-		long expire = System.currentTimeMillis() + timeout;
+		return getSingleRecord(consumer, topic, Duration.ofMillis(timeout));
+	}
+
+	/**
+	 * Poll the consumer, expecting a single record for the specified topic.
+	 * @param consumer the consumer.
+	 * @param topic the topic.
+	 * @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.9.3
+	 */
+	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 -> {
@@ -198,11 +215,31 @@ public static <K, V> ConsumerRecord<K, V> getSingleRecord(Consumer<K, V> consume
 	 * @param timeout the timeout.
 	 * @return the record or null if no record received.
 	 * @since 2.3
+	 * @deprecated in favor of {@link #getOneRecord(String, String, String, int, boolean, boolean, Duration)}
+	 */
+	@Nullable
+	@Deprecated
+	public static ConsumerRecord<?, ?> getOneRecord(String brokerAddresses, String group, String topic, int partition,
+													boolean seekToLast, boolean commit, long timeout) {
+		return getOneRecord(brokerAddresses, group, topic, partition, seekToLast, commit, Duration.ofMillis(timeout));
+	}
+
+	/**
+	 * Get a single record for the group from the topic/partition. Optionally, seeking to the current last record.
+	 * @param brokerAddresses the broker address(es).
+	 * @param group the group.
+	 * @param topic the topic.
+	 * @param partition the partition.
+	 * @param seekToLast true to fetch an existing last record, if present.
+	 * @param commit commit offset after polling or not.
+	 * @param timeout the timeout.
+	 * @return the record or null if no record received.
+	 * @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 +252,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 +335,74 @@ 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(long)}.
 	 * @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
+	 * @deprecated in favor of {@link #getRecords(Consumer, Duration)}
 	 */
+	@Deprecated
 	public static <K, V> ConsumerRecords<K, V> getRecords(Consumer<K, V> consumer, long 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.
+	 * @return the records.
+	 * @throws IllegalStateException if the poll returns null (since 2.3.4).
+	 * @since 2.9.3
+	 */
+	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 <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
+	 * @deprecated in favor of {#{@link #getRecords(Consumer, Duration, int)}}
 	 */
+	@Deprecated
 	public static <K, V> ConsumerRecords<K, V> getRecords(Consumer<K, V> consumer, long timeout, int minRecords) {
+		return getRecords(consumer, Duration.ofMillis(timeout), minRecords);
+	}
+
+	/**
+	 * Poll the consumer for records.
+	 * @param consumer the consumer.
+	 * @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 records are received.
+	 * @return the records.
+	 * @throws IllegalStateException if the poll returns null.
+	 * @since 2.9.3
+	 */
+	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();

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())

spring-kafka/src/test/java/org/springframework/kafka/listener/ErrorHandlingCoverageTests.java

@@ -5,7 +5,7 @@
  * you may not use this file except in compliance with the License.
  * You may obtain a copy of the License at
  *
- *      http://www.apache.org/licenses/LICENSE-2.0
+ *      https://www.apache.org/licenses/LICENSE-2.0
  *
  * Unless required by applicable law or agreed to in writing, software
  * distributed under the License is distributed on an "AS IS" BASIS,