GH-1861: Add KafkaTemplate.receive()

Resolves #1861

* Pull constant up to interface.

Author: garyrussell

spring-kafka-docs/src/main/asciidoc/kafka.adoc

@@ -2457,6 +2457,24 @@ As an aside; previously, containers in each group were added to a bean of type `
 These collections are now deprecated in favor of beans of type `ContainerGroup` with a bean name that is the group name, suffixed with `.group`; in the example above, there would be 2 beans `g1.group` and `g2.group`.
 The `Collection` beans will be removed in a future release.
 
+[[kafka-template-receive]]
+===== Using `KafkaTemplate` to Receive
+
+This section covers how to use `KafkaTemplate` to receive messages.
+
+Starting with version 2.8, the template has two `receive()` methods:
+
+====
+[source, jvava]
+----
+ConsumerRecord<K, V> receive(String topic, int partition, long offset);
+
+ConsumerRecord<K, V> receive(String topic, int partition, long offset, Duration pollTimeout);
+----
+====
+
+As you can see, you need to know the partition and offset of the record you need to retrieve; a new `Consumer` is created (and closed) for each operation.
+
 [[container-props]]
 ==== Listener Container Properties
 

spring-kafka-docs/src/main/asciidoc/whats-new.adoc

@@ -14,3 +14,9 @@ This version requires the 2.8.0 `kafka-clients`.
 The listener container can now be configured to accept manual offset commits out of order (usually asynchronously).
 The container will defer the commit until the missing offset is acknowledged.
 See <<ooo-commits>> for more information.
+
+[[x28-template]]
+==== `KafkaTemplate` Changes
+
+You can now receive a single record, given the topic, partition and offset.
+See <<kafka-template-receive>> for more information.

spring-kafka/src/main/java/org/springframework/kafka/core/KafkaOperations.java

@@ -16,10 +16,12 @@
 
 package org.springframework.kafka.core;
 
+import java.time.Duration;
 import java.util.List;
 import java.util.Map;
 
 import org.apache.kafka.clients.consumer.ConsumerGroupMetadata;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
 import org.apache.kafka.clients.consumer.OffsetAndMetadata;
 import org.apache.kafka.clients.producer.Producer;
 import org.apache.kafka.clients.producer.ProducerRecord;
@@ -53,6 +55,11 @@
  */
 public interface KafkaOperations<K, V> {
 
+	/**
+	 * Default timeout for {@link #receive(String, int, long)}.
+	 */
+	Duration DEFAULT_POLL_TIMEOUT = Duration.ofSeconds(5);
+
 	/**
 	 * Send the data to the default topic with no key or partition.
 	 * @param data The data.
@@ -267,6 +274,30 @@ default ProducerFactory<K, V> getProducerFactory() {
 		throw new UnsupportedOperationException("This implementation does not support this operation");
 	}
 
+	/**
+	 * Receive a single record with the default poll timeout (5 seconds).
+	 * @param topic the topic.
+	 * @param partition the partition.
+	 * @param offset the offset.
+	 * @return the record or null.
+	 * @since 2.8
+	 * @see #DEFAULT_POLL_TIMEOUT
+	 */
+	@Nullable
+	ConsumerRecord<K, V> receive(String topic, int partition, long offset);
+
+	/**
+	 * Receive a single record.
+	 * @param topic the topic.
+	 * @param partition the partition.
+	 * @param offset the offset.
+	 * @param pollTimeout the timeout.
+	 * @return the record or null.
+	 * @since 2.8
+	 */
+	@Nullable
+	ConsumerRecord<K, V> receive(String topic, int partition, long offset, Duration pollTimeout);
+
 	/**
 	 * A callback for executing arbitrary operations on the {@link Producer}.
 	 * @param <K> the key type.

spring-kafka/src/main/java/org/springframework/kafka/core/KafkaTemplate.java

@@ -17,14 +17,20 @@
 package org.springframework.kafka.core;
 
 import java.time.Duration;
+import java.util.Collections;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.Properties;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.Future;
 
 import org.apache.commons.logging.LogFactory;
+import org.apache.kafka.clients.consumer.Consumer;
+import org.apache.kafka.clients.consumer.ConsumerConfig;
 import org.apache.kafka.clients.consumer.ConsumerGroupMetadata;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.clients.consumer.ConsumerRecords;
 import org.apache.kafka.clients.consumer.OffsetAndMetadata;
 import org.apache.kafka.clients.producer.Callback;
 import org.apache.kafka.clients.producer.Producer;
@@ -112,6 +118,8 @@ public class KafkaTemplate<K, V> implements KafkaOperations<K, V>, ApplicationCo
 
 	private boolean converterSet;
 
+	private ConsumerFactory<K, V> consumerFactory;
+
 	private volatile boolean micrometerEnabled = true;
 
 	private volatile MicrometerHolder micrometerHolder;
@@ -347,6 +355,15 @@ protected ProducerFactory<K, V> getProducerFactory(String topic) {
 		return this.producerFactory;
 	}
 
+	/**
+	 * Set a consumer factory for receive operations.
+	 * @param consumerFactory the consumer factory.
+	 * @since 2.8
+	 */
+	public void setConsumerFactory(ConsumerFactory<K, V> consumerFactory) {
+		this.consumerFactory = consumerFactory;
+	}
+
 	@Override
 	public void onApplicationEvent(ContextStoppedEvent event) {
 		if (this.customProducerFactory) {
@@ -541,6 +558,31 @@ public void sendOffsetsToTransaction(Map<TopicPartition, OffsetAndMetadata> offs
 		producerForOffsets().sendOffsetsToTransaction(offsets, groupMetadata);
 	}
 
+
+	@Override
+	@Nullable
+	public ConsumerRecord<K, V> receive(String topic, int partition, long offset) {
+		return receive(topic, partition, offset, DEFAULT_POLL_TIMEOUT);
+	}
+
+	@Override
+	@Nullable
+	public ConsumerRecord<K, V> receive(String topic, int partition, long offset, Duration pollTimeout) {
+		Assert.notNull(this.consumerFactory, "A consumerFactory is required");
+		Properties props = new Properties();
+		props.setProperty(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, "1");
+		try (Consumer<K, V> consumer = this.consumerFactory.createConsumer(null, null, null, props)) {
+			TopicPartition topicPartition = new TopicPartition(topic, partition);
+			consumer.assign(Collections.singletonList(topicPartition));
+			consumer.seek(topicPartition, offset);
+			ConsumerRecords<K, V> records = consumer.poll(pollTimeout);
+			if (records.count() == 1) {
+				return records.iterator().next();
+			}
+			return null;
+		}
+	}
+
 	private Producer<K, V> producerForOffsets() {
 		Producer<K, V> producer = this.producers.get();
 		if (producer == null) {

spring-kafka/src/test/java/org/springframework/kafka/core/KafkaTemplateTests.java

@@ -171,6 +171,12 @@ void testTemplate() {
 		assertThat(partitions).isNotNull();
 		assertThat(partitions).hasSize(2);
 		assertThat(KafkaTestUtils.getPropertyValue(pf.createProducer(), "delegate")).isSameAs(wrapped.get());
+		template.setConsumerFactory(
+				new DefaultKafkaConsumerFactory<>(KafkaTestUtils.consumerProps("xx", "false", embeddedKafka)));
+		ConsumerRecord<Integer, String> receive = template.receive(INT_KEY_TOPIC, 0, received.offset());
+		assertThat(receive).has(allOf(keyValue(2, "buz"), partition(0)))
+				.extracting(rec -> rec.offset())
+				.isEqualTo(received.offset());
 		pf.destroy();
 	}