GH-1475: Add KafkaSendCallback

Resolves #1475

Add `onFailure` with a narrowed `Throwable` to make it easier to access
the failed producer record.

* Make `KafkaFailureCallback` a `FunctionalInterface` and fix javadocs.

Author: garyrussell

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

@@ -0,0 +1,46 @@
+/*
+ * Copyright 2020 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.
+ * You may obtain a copy of the License at
+ *
+ *      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,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.kafka.core;
+
+import org.springframework.util.concurrent.FailureCallback;
+
+/**
+ * An enhanced {@link FailureCallback} for reporting
+ * {@link KafkaProducerException}s.
+ *
+ * @param <K> the key type.
+ * @param <V> the value type.
+ *
+ * @author Gary Russell
+ * @since 2.5
+ *
+ */
+@FunctionalInterface
+public interface KafkaFailureCallback<K, V> extends FailureCallback {
+
+	@Override
+	default void onFailure(Throwable ex) {
+		onFailure((KafkaProducerException) ex);
+	}
+
+	/**
+	 * Called when the send fails.
+	 * @param ex the exception.
+	 */
+	void onFailure(KafkaProducerException ex);
+
+}

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2019 the original author or authors.
+ * Copyright 2016-2020 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.
@@ -31,13 +31,37 @@ public class KafkaProducerException extends KafkaException {
 
 	private final ProducerRecord<?, ?> producerRecord;
 
+	/**
+	 * Construct an instance with the provided properties.
+	 * @param failedProducerRecord the producer record.
+	 * @param message the message.
+	 * @param cause the cause.
+	 */
 	public KafkaProducerException(ProducerRecord<?, ?> failedProducerRecord, String message, Throwable cause) {
 		super(message, cause);
 		this.producerRecord = failedProducerRecord;
 	}
 
+	/**
+	 * Return the failed producer record.
+	 * @return the record.
+	 * @deprecated in favor of {@link #getFailedProducerRecord()}
+	 */
+	@Deprecated
 	public ProducerRecord<?, ?> getProducerRecord() {
 		return this.producerRecord;
 	}
 
+	/**
+	 * Return the failed producer record.
+	 * @param <K> the key type.
+	 * @param <V> the value type.
+	 * @return the record.
+	 * @since 2.5
+	 */
+	@SuppressWarnings("unchecked")
+	public <K, V> ProducerRecord<K, V> getFailedProducerRecord() {
+		return (ProducerRecord<K, V>) producerRecord;
+	}
+
 }

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

@@ -0,0 +1,35 @@
+/*
+ * Copyright 2020 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.
+ * You may obtain a copy of the License at
+ *
+ *      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,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.kafka.core;
+
+import org.springframework.kafka.support.SendResult;
+import org.springframework.util.concurrent.ListenableFutureCallback;
+
+/**
+ * An enhanced {@link ListenableFutureCallback} for reporting
+ * {@link KafkaProducerException}s.
+ *
+ * @param <K> the key type.
+ * @param <V> the value type.
+ *
+ * @author Gary Russell
+ * @since 2.5
+ *
+ */
+public interface KafkaSendCallback<K, V> extends ListenableFutureCallback<SendResult<K, V>>, KafkaFailureCallback<K, V> {
+
+}

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

@@ -19,6 +19,9 @@
 import static org.assertj.core.api.Assertions.allOf;
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.assertThatExceptionOfType;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.BDDMockito.given;
+import static org.mockito.BDDMockito.willAnswer;
 import static org.mockito.Mockito.mock;
 import static org.springframework.kafka.test.assertj.KafkaConditions.key;
 import static org.springframework.kafka.test.assertj.KafkaConditions.keyValue;
@@ -41,6 +44,7 @@
 
 import org.apache.kafka.clients.consumer.Consumer;
 import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.clients.producer.Callback;
 import org.apache.kafka.clients.producer.Producer;
 import org.apache.kafka.clients.producer.ProducerConfig;
 import org.apache.kafka.clients.producer.ProducerRecord;
@@ -75,6 +79,7 @@
 import org.springframework.messaging.support.MessageBuilder;
 import org.springframework.util.concurrent.ListenableFuture;
 import org.springframework.util.concurrent.ListenableFutureCallback;
+import org.springframework.util.concurrent.SettableListenableFuture;
 
 /**
  * @author Gary Russell
@@ -317,7 +322,7 @@ public void onSuccess(ProducerRecord<Integer, String> record, RecordMetadata rec
 	@Test
 	void testWithCallback() throws Exception {
 		Map<String, Object> senderProps = KafkaTestUtils.producerProps(embeddedKafka);
-		ProducerFactory<Integer, String> pf = new DefaultKafkaProducerFactory<>(senderProps);
+		DefaultKafkaProducerFactory<Integer, String> pf = new DefaultKafkaProducerFactory<>(senderProps);
 		KafkaTemplate<Integer, String> template = new KafkaTemplate<>(pf, true);
 		template.setDefaultTopic(INT_KEY_TOPIC);
 		ListenableFuture<SendResult<Integer, String>> future = template.sendDefault("foo");
@@ -339,7 +344,66 @@ public void onFailure(Throwable ex) {
 		});
 		assertThat(KafkaTestUtils.getSingleRecord(consumer, INT_KEY_TOPIC)).has(value("foo"));
 		assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue();
-		pf.createProducer().close();
+		pf.destroy();
+	}
+
+	@SuppressWarnings("unchecked")
+	@Test
+	void testWithCallbackFailure() throws Exception {
+		Producer<Integer, String> producer = mock(Producer.class);
+		willAnswer(inv -> {
+			Callback callback = inv.getArgument(1);
+			callback.onCompletion(null, new RuntimeException("test"));
+			return new SettableListenableFuture<RecordMetadata>();
+		}).given(producer).send(any(), any());
+		ProducerFactory<Integer, String> pf = mock(ProducerFactory.class);
+		given(pf.createProducer()).willReturn(producer);
+		KafkaTemplate<Integer, String> template = new KafkaTemplate<>(pf);
+		ListenableFuture<SendResult<Integer, String>> future = template.send("foo", 1, "bar");
+		final CountDownLatch latch = new CountDownLatch(1);
+		final AtomicReference<SendResult<Integer, String>> theResult = new AtomicReference<>();
+		AtomicReference<String> value = new AtomicReference<>();
+		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();
+				value.set(failed.value());
+				latch.countDown();
+			}
+
+		});
+		assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue();
+		assertThat(value.get()).isEqualTo("bar");
+	}
+
+	@SuppressWarnings("unchecked")
+	@Test
+	void testWithCallbackFailureFunctional() throws Exception {
+		Producer<Integer, String> producer = mock(Producer.class);
+		willAnswer(inv -> {
+			Callback callback = inv.getArgument(1);
+			callback.onCompletion(null, new RuntimeException("test"));
+			return new SettableListenableFuture<RecordMetadata>();
+		}).given(producer).send(any(), any());
+		ProducerFactory<Integer, String> pf = mock(ProducerFactory.class);
+		given(pf.createProducer()).willReturn(producer);
+		KafkaTemplate<Integer, String> template = new KafkaTemplate<>(pf);
+		ListenableFuture<SendResult<Integer, String>> future = template.send("foo", 1, "bar");
+		final CountDownLatch latch = new CountDownLatch(1);
+		final AtomicReference<SendResult<Integer, String>> theResult = new AtomicReference<>();
+		AtomicReference<String> value = new AtomicReference<>();
+		future.addCallback(result -> { }, (KafkaFailureCallback<Integer, String>) ex -> {
+			ProducerRecord<Integer, String> failed = ex.getFailedProducerRecord();
+			value.set(failed.value());
+			latch.countDown();
+		});
+		assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue();
+		assertThat(value.get()).isEqualTo("bar");
 	}
 
 	@Test