GH-2357: Switch to CompletableFuture

Resolves #2357

Spring Framework is planning to deprecate `ListenableFuture` in 6.0.

Add methods to the `KafkaOperations` (`KafkaTemplate`) that return
`CompletableFuture` instead; the `ListenableFuture` methods have now
been removed in 3.0.

* Remove KafkaOperations2; doc polishing.

* Fix import.

Author: garyrussell

spring-kafka-docs/src/main/asciidoc/changes-since-1.0.adoc

@@ -33,13 +33,22 @@ You can now configure which inbound headers should be mapped.
 Also available in version 2.8.8 or later.
 See <<headers>> for more information.
 
+[[x29-template-changes]]
+==== `KafkaTemplate` Changes
+
+In 3.0, the futures returned by this class will be `CompletableFuture` s instead of `ListenableFuture` s.
+See <<kafka-template>> for assistance in transitioning when using this release.
+
 [[x29-rkt-changes]]
 ==== `ReplyingKafkaTemplate` Changes
 
 The template now provides a method to wait for assignment on the reply container, to avoid a race when sending a request before the reply container is initialized.
 Also available in version 2.8.8 or later.
 See <<replying-template>>.
 
+In 3.0, the futures returned by this class will be `CompletableFuture` s instead of `ListenableFuture` s.
+See <<replying-template>> and <<exchanging-messages>> for assistance in transitioning when using this release.
+
 === What's New in 2.8 Since 2.7
 
 This section covers the changes made from version 2.7 to version 2.8.

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

@@ -170,25 +170,25 @@ The following listing shows the relevant methods from `KafkaTemplate`:
 ====
 [source, java]
 ----
-ListenableFuture<SendResult<K, V>> sendDefault(V data);
+CompletableFuture<SendResult<K, V>> sendDefault(V data);
 
-ListenableFuture<SendResult<K, V>> sendDefault(K key, V data);
+CompletableFuture<SendResult<K, V>> sendDefault(K key, V data);
 
-ListenableFuture<SendResult<K, V>> sendDefault(Integer partition, K key, V data);
+CompletableFuture<SendResult<K, V>> sendDefault(Integer partition, K key, V data);
 
-ListenableFuture<SendResult<K, V>> sendDefault(Integer partition, Long timestamp, K key, V data);
+CompletableFuture<SendResult<K, V>> sendDefault(Integer partition, Long timestamp, K key, V data);
 
-ListenableFuture<SendResult<K, V>> send(String topic, V data);
+CompletableFuture<SendResult<K, V>> send(String topic, V data);
 
-ListenableFuture<SendResult<K, V>> send(String topic, K key, V data);
+CompletableFuture<SendResult<K, V>> send(String topic, K key, V data);
 
-ListenableFuture<SendResult<K, V>> send(String topic, Integer partition, K key, V data);
+CompletableFuture<SendResult<K, V>> send(String topic, Integer partition, K key, V data);
 
-ListenableFuture<SendResult<K, V>> send(String topic, Integer partition, Long timestamp, K key, V data);
+CompletableFuture<SendResult<K, V>> send(String topic, Integer partition, Long timestamp, K key, V data);
 
-ListenableFuture<SendResult<K, V>> send(ProducerRecord<K, V> record);
+CompletableFuture<SendResult<K, V>> send(ProducerRecord<K, V> record);
 
-ListenableFuture<SendResult<K, V>> send(Message<?> message);
+CompletableFuture<SendResult<K, V>> send(Message<?> message);
 
 Map<MetricName, ? extends Metric> metrics();
 
@@ -210,6 +210,9 @@ interface ProducerCallback<K, V, T> {
 
 See the https://docs.spring.io/spring-kafka/api/org/springframework/kafka/core/KafkaTemplate.html[Javadoc] for more detail.
 
+IMPORTANT: In version 3.0, the methods that previously returned `ListenableFuture` have been changed to return `CompletableFuture`.
+To facilitate the migration, the 2.9 version added a method `.usingCompletableFuture()` which provided the same methods with `CompletableFuture` return types; this method is no longer available.
+
 The `sendDefault` API requires that a default topic has been provided to the template.
 
 The API takes in a `timestamp` as a parameter and stores this timestamp in the record.
@@ -302,75 +305,27 @@ By default, the template is configured with a `LoggingProducerListener`, which l
 
 For convenience, default method implementations are provided in case you want to implement only one of the methods.
 
-Notice that the send methods return a `ListenableFuture<SendResult>`.
+Notice that the send methods return a `CompletableFuture<SendResult>`.
 You can register a callback with the listener to receive the result of the send asynchronously.
 The following example shows how to do so:
 
 ====
 [source, java]
 ----
-ListenableFuture<SendResult<Integer, String>> future = template.send("myTopic", "something");
-future.addCallback(new ListenableFutureCallback<SendResult<Integer, String>>() {
-
-    @Override
-    public void onSuccess(SendResult<Integer, String> result) {
-        ...
-    }
-
-    @Override
-    public void onFailure(Throwable ex) {
-        ...
-    }
-
+CompletableFuture<SendResult<Integer, String>> future = template.send("myTopic", "something");
+future.whenComplete((result, ex) -> {
+    ...
 });
 ----
 ====
 
 `SendResult` has two properties, a `ProducerRecord` and `RecordMetadata`.
 See the Kafka API documentation for information about those objects.
 
-The `Throwable` in `onFailure` can be cast to a `KafkaProducerException`; its `failedProducerRecord` property contains the failed record.
-
-Starting with version 2.5, you can use a `KafkaSendCallback` instead of a `ListenableFutureCallback`, making it easier to extract the failed `ProducerRecord`, avoiding the need to cast the `Throwable`:
-
-====
-[source, java]
-----
-ListenableFuture<SendResult<Integer, String>> future = template.send("topic", 1, "thing");
-future.addCallback(new KafkaSendCallback<Integer, String>() {
-
-    @Override
-    public void onSuccess(SendResult<Integer, String> result) {
-        ...
-    }
-
-    @Override
-    public void onFailure(KafkaProducerException ex) {
-        ProducerRecord<Integer, String> failed = ex.getFailedProducerRecord();
-        ...
-    }
-
-});
-----
-====
-
-You can also use a pair of lambdas:
-
-====
-[source, java]
-----
-ListenableFuture<SendResult<Integer, String>> future = template.send("topic", 1, "thing");
-future.addCallback(result -> {
-        ...
-    }, (KafkaFailureCallback<Integer, String>) ex -> {
-            ProducerRecord<Integer, String> failed = ex.getFailedProducerRecord();
-            ...
-    });
-----
-====
+The `Throwable` can be cast to a `KafkaProducerException`; its `failedProducerRecord` property contains the failed record.
 
 If you wish to block the sending thread to await the result, you can invoke the future's `get()` method; using the method with a timeout is recommended.
-You may wish to invoke `flush()` before waiting or, for convenience, the template has a constructor with an `autoFlush` parameter that causes the template to `flush()` on each send.
+If you have set a `linger.ms`, you may wish to invoke `flush()` before waiting or, for convenience, the template has a constructor with an `autoFlush` parameter that causes the template to `flush()` on each send.
 Flushing is only needed if you have set the `linger.ms` producer property and want to immediately send a partial batch.
 
 ====== Examples
@@ -384,19 +339,14 @@ This section shows examples of sending messages to Kafka:
 public void sendToKafka(final MyOutputData data) {
     final ProducerRecord<String, String> record = createRecord(data);
 
-    ListenableFuture<SendResult<Integer, String>> future = template.send(record);
-    future.addCallback(new KafkaSendCallback<Integer, String>() {
-
-        @Override
-        public void onSuccess(SendResult<Integer, String> result) {
+    CompletableFuture<SendResult<Integer, String>> future = template.send(record);
+    future.whenComplete((result, ex) -> {
+        if (ex == null) {
             handleSuccess(data);
         }
-
-        @Override
-        public void onFailure(KafkaProducerException ex) {
+        else {
             handleFailure(data, record, ex);
         }
-
     });
 }
 ----
@@ -549,10 +499,12 @@ RequestReplyFuture<K, V, R> sendAndReceive(ProducerRecord<K, V> record,
 
 (Also see <<exchanging-messages>>).
 
-The result is a `ListenableFuture` that is asynchronously populated with the result (or an exception, for a timeout).
+The result is a `CompletableFuture` that is asynchronously populated with the result (or an exception, for a timeout).
 The result also has a `sendFuture` property, which is the result of calling `KafkaTemplate.send()`.
 You can use this future to determine the result of the send operation.
 
+IMPORTANT: In version 3.0, the futures returned by these methods (and their `sendFuture` properties) have been changed to  `CompletableFuture` s instead of `ListenableFuture` s.
+
 If the first method is used, or the `replyTimeout` argument is `null`, the template's `defaultReplyTimeout` property is used (5 seconds by default).
 
 Starting with version 2.8.8, the template has a new method `waitForAssignment`.
@@ -791,6 +743,8 @@ RequestReplyMessageFuture<K, V> sendAndReceive(Message<?> message);
 
 These will use the template's default `replyTimeout`, there are also overloaded versions that can take a timeout in the method call.
 
+IMPORTANT: In version 3.0, the futures returned by these methods (and their `sendFuture` properties) have been changed to  `CompletableFuture` s instead of `ListenableFuture` s.
+
 Use the first method if the consumer's `Deserializer` or the template's `MessageConverter` can convert the payload without any additional information, either via configuration or type metadata in the reply message.
 
 Use the second method if you need to provide type information for the return type, to assist the message converter.
@@ -2236,6 +2190,7 @@ public ConcurrentKafkaListenerContainerFactory<Integer, String> kafkaListenerCon
 ====
 
 When you use `@SendTo`, you must configure the `ConcurrentKafkaListenerContainerFactory` with a `KafkaTemplate` in its `replyTemplate` property to perform the send.
+Spring Boot will automatically wire in its auto configured template (or any if a single instance is present).
 
 NOTE: Unless you use <<replying-template,request/reply semantics>> only the simple `send(topic, value)` method is used, so you may wish to create a subclass to generate the partition or key.
 The following example shows how to do so:
@@ -2248,7 +2203,7 @@ public KafkaTemplate<String, String> myReplyingTemplate() {
     return new KafkaTemplate<Integer, String>(producerFactory()) {
 
         @Override
-        public ListenableFuture<SendResult<String, String>> send(String topic, String data) {
+        public CompletableFuture<SendResult<String, String>> send(String topic, String data) {
             return super.send(topic, partitionForData(data), keyForData(data), data);
         }
 

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

@@ -34,3 +34,15 @@ See <<retry-config>> for more information.
 
 Events related to consumer authentication and authorization failures are now published by the container.
 See <<events>> for more information.
+
+[[x30-template-changes]]
+==== `KafkaTemplate` Changes
+
+The futures returned by this class are now `CompletableFuture` s instead of `ListenableFuture` s.
+See <<kafka-template>>.
+
+[[x30-rkt-changes]]
+==== `ReplyingKafkaTemplate` Changes
+
+The futures returned by this class are now `CompletableFuture` s instead of `ListenableFuture` s.
+See <<replying-template>> and <<exchanging-messages>>.

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

@@ -20,6 +20,7 @@
 import java.util.Collection;
 import java.util.List;
 import java.util.Map;
+import java.util.concurrent.CompletableFuture;
 
 import org.apache.kafka.clients.consumer.ConsumerGroupMetadata;
 import org.apache.kafka.clients.consumer.ConsumerRecord;
@@ -36,10 +37,9 @@
 import org.springframework.kafka.support.TopicPartitionOffset;
 import org.springframework.lang.Nullable;
 import org.springframework.messaging.Message;
-import org.springframework.util.concurrent.ListenableFuture;
 
 /**
- * The basic Kafka operations contract returning {@link ListenableFuture}s.
+ * The basic Kafka operations contract returning {@link CompletableFuture}s.
  *
  * @param <K> the key type.
  * @param <V> the value type.
@@ -68,15 +68,15 @@ public interface KafkaOperations<K, V> {
 	 * @param data The data.
 	 * @return a Future for the {@link SendResult}.
 	 */
-	ListenableFuture<SendResult<K, V>> sendDefault(V data);
+	CompletableFuture<SendResult<K, V>> sendDefault(V data);
 
 	/**
 	 * Send the data to the default topic with the provided key and no partition.
 	 * @param key the key.
 	 * @param data The data.
 	 * @return a Future for the {@link SendResult}.
 	 */
-	ListenableFuture<SendResult<K, V>> sendDefault(K key, V data);
+	CompletableFuture<SendResult<K, V>> sendDefault(K key, V data);
 
 	/**
 	 * Send the data to the default topic with the provided key and partition.
@@ -85,7 +85,7 @@ public interface KafkaOperations<K, V> {
 	 * @param data the data.
 	 * @return a Future for the {@link SendResult}.
 	 */
-	ListenableFuture<SendResult<K, V>> sendDefault(Integer partition, K key, V data);
+	CompletableFuture<SendResult<K, V>> sendDefault(Integer partition, K key, V data);
 
 	/**
 	 * Send the data to the default topic with the provided key and partition.
@@ -96,15 +96,15 @@ public interface KafkaOperations<K, V> {
 	 * @return a Future for the {@link SendResult}.
 	 * @since 1.3
 	 */
-	ListenableFuture<SendResult<K, V>> sendDefault(Integer partition, Long timestamp, K key, V data);
+	CompletableFuture<SendResult<K, V>> sendDefault(Integer partition, Long timestamp, K key, V data);
 
 	/**
 	 * Send the data to the provided topic with no key or partition.
 	 * @param topic the topic.
 	 * @param data The data.
 	 * @return a Future for the {@link SendResult}.
 	 */
-	ListenableFuture<SendResult<K, V>> send(String topic, V data);
+	CompletableFuture<SendResult<K, V>> send(String topic, V data);
 
 	/**
 	 * Send the data to the provided topic with the provided key and no partition.
@@ -113,7 +113,7 @@ public interface KafkaOperations<K, V> {
 	 * @param data The data.
 	 * @return a Future for the {@link SendResult}.
 	 */
-	ListenableFuture<SendResult<K, V>> send(String topic, K key, V data);
+	CompletableFuture<SendResult<K, V>> send(String topic, K key, V data);
 
 	/**
 	 * Send the data to the provided topic with the provided key and partition.
@@ -123,7 +123,7 @@ public interface KafkaOperations<K, V> {
 	 * @param data the data.
 	 * @return a Future for the {@link SendResult}.
 	 */
-	ListenableFuture<SendResult<K, V>> send(String topic, Integer partition, K key, V data);
+	CompletableFuture<SendResult<K, V>> send(String topic, Integer partition, K key, V data);
 
 	/**
 	 * Send the data to the provided topic with the provided key and partition.
@@ -135,15 +135,15 @@ public interface KafkaOperations<K, V> {
 	 * @return a Future for the {@link SendResult}.
 	 * @since 1.3
 	 */
-	ListenableFuture<SendResult<K, V>> send(String topic, Integer partition, Long timestamp, K key, V data);
+	CompletableFuture<SendResult<K, V>> send(String topic, Integer partition, Long timestamp, K key, V data);
 
 	/**
 	 * Send the provided {@link ProducerRecord}.
 	 * @param record the record.
 	 * @return a Future for the {@link SendResult}.
 	 * @since 1.3
 	 */
-	ListenableFuture<SendResult<K, V>> send(ProducerRecord<K, V> record);
+	CompletableFuture<SendResult<K, V>> send(ProducerRecord<K, V> record);
 
 	/**
 	 * Send a message with routing information in message headers. The message payload
@@ -154,7 +154,7 @@ public interface KafkaOperations<K, V> {
 	 * @see org.springframework.kafka.support.KafkaHeaders#PARTITION
 	 * @see org.springframework.kafka.support.KafkaHeaders#KEY
 	 */
-	ListenableFuture<SendResult<K, V>> send(Message<?> message);
+	CompletableFuture<SendResult<K, V>> send(Message<?> message);
 
 	/**
 	 * See {@link Producer#partitionsFor(String)}.