GH-1437: Check for immediate failure on send

Resolves #1437

The future returned by `Producer.send()` may have been immediately completed
with an exception; check if `future.isDone()` and call `get()` so that any
such exception is propagated to the caller.

**I will perform backports after review/merge.**

* Fix mock tests.

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

@@ -19,18 +19,22 @@
 import java.time.Duration;
 import java.util.List;
 import java.util.Map;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Future;
 
 import org.apache.commons.logging.LogFactory;
 import org.apache.kafka.clients.consumer.OffsetAndMetadata;
 import org.apache.kafka.clients.producer.Callback;
 import org.apache.kafka.clients.producer.Producer;
 import org.apache.kafka.clients.producer.ProducerRecord;
+import org.apache.kafka.clients.producer.RecordMetadata;
 import org.apache.kafka.common.Metric;
 import org.apache.kafka.common.MetricName;
 import org.apache.kafka.common.PartitionInfo;
 import org.apache.kafka.common.TopicPartition;
 
 import org.springframework.core.log.LogAccessor;
+import org.springframework.kafka.KafkaException;
 import org.springframework.kafka.support.KafkaHeaders;
 import org.springframework.kafka.support.KafkaUtils;
 import org.springframework.kafka.support.LoggingProducerListener;
@@ -401,7 +405,21 @@ protected ListenableFuture<SendResult<K, V>> doSend(final ProducerRecord<K, V> p
 		final Producer<K, V> producer = getTheProducer();
 		this.logger.trace(() -> "Sending: " + producerRecord);
 		final SettableListenableFuture<SendResult<K, V>> future = new SettableListenableFuture<>();
-		producer.send(producerRecord, buildCallback(producerRecord, producer, future));
+		Future<RecordMetadata> sendFuture =
+				producer.send(producerRecord, buildCallback(producerRecord, producer, future));
+		// May be an immediate failure
+		if (sendFuture.isDone()) {
+			try {
+				sendFuture.get();
+			}
+			catch (InterruptedException e) {
+				Thread.currentThread().interrupt();
+				throw new KafkaException("Interrupted", e);
+			}
+			catch (ExecutionException e) {
+				throw new KafkaException("Send failed", e.getCause()); // NOSONAR, stack trace
+			}
+		}
 		if (this.autoFlush) {
 			flush();
 		}

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2018-2019 the original author or authors.
+ * Copyright 2018-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.
@@ -19,6 +19,7 @@
 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.BDDMockito.willThrow;
 import static org.mockito.Mockito.inOrder;
@@ -45,6 +46,7 @@
 import org.springframework.kafka.transaction.KafkaTransactionManager;
 import org.springframework.transaction.CannotCreateTransactionException;
 import org.springframework.transaction.support.TransactionTemplate;
+import org.springframework.util.concurrent.SettableListenableFuture;
 
 /**
  * @author Gary Russell
@@ -57,6 +59,7 @@ public class DefaultKafkaProducerFactoryTests {
 	@Test
 	void testProducerClosedAfterBadTransition() throws Exception {
 		final Producer producer = mock(Producer.class);
+		given(producer.send(any(), any())).willReturn(new SettableListenableFuture<>());
 		DefaultKafkaProducerFactory pf = new DefaultKafkaProducerFactory(new HashMap<>()) {
 
 			@Override

spring-kafka/src/test/java/org/springframework/kafka/core/KafkaTemplateTests.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.
@@ -18,6 +18,7 @@
 
 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.Mockito.mock;
 import static org.springframework.kafka.test.assertj.KafkaConditions.key;
 import static org.springframework.kafka.test.assertj.KafkaConditions.keyValue;
@@ -39,11 +40,13 @@
 import org.apache.kafka.clients.consumer.ConsumerConfig;
 import org.apache.kafka.clients.consumer.ConsumerRecord;
 import org.apache.kafka.clients.producer.Producer;
+import org.apache.kafka.clients.producer.ProducerConfig;
 import org.apache.kafka.clients.producer.ProducerRecord;
 import org.apache.kafka.clients.producer.RecordMetadata;
 import org.apache.kafka.common.Metric;
 import org.apache.kafka.common.MetricName;
 import org.apache.kafka.common.PartitionInfo;
+import org.apache.kafka.common.errors.TimeoutException;
 import org.apache.kafka.common.header.Header;
 import org.apache.kafka.common.serialization.StringDeserializer;
 import org.apache.kafka.common.serialization.StringSerializer;
@@ -52,6 +55,7 @@
 import org.junit.jupiter.api.BeforeAll;
 import org.junit.jupiter.api.Test;
 
+import org.springframework.kafka.KafkaException;
 import org.springframework.kafka.support.Acknowledgment;
 import org.springframework.kafka.support.CompositeProducerListener;
 import org.springframework.kafka.support.DefaultKafkaHeaderMapper;
@@ -358,4 +362,17 @@ public void testTemplateDisambiguation() {
 		pf.destroy();
 	}
 
+	@Test
+	void testFutureFailureOnSend() {
+		Map<String, Object> senderProps = KafkaTestUtils.producerProps(embeddedKafka);
+		senderProps.put(ProducerConfig.MAX_BLOCK_MS_CONFIG, 10);
+		DefaultKafkaProducerFactory<Integer, String> pf = new DefaultKafkaProducerFactory<>(senderProps);
+		KafkaTemplate<Integer, String> template = new KafkaTemplate<>(pf, true);
+
+		assertThatExceptionOfType(KafkaException.class).isThrownBy(() ->
+			template.send("missing.topic", "foo"))
+				.withCauseExactlyInstanceOf(TimeoutException.class);
+		pf.destroy();
+	}
+
 }

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

@@ -77,6 +77,7 @@
 import org.springframework.transaction.annotation.Transactional;
 import org.springframework.transaction.support.AbstractPlatformTransactionManager;
 import org.springframework.transaction.support.TransactionTemplate;
+import org.springframework.util.concurrent.SettableListenableFuture;
 
 /**
  * @author Gary Russell
@@ -306,8 +307,10 @@ public void testTransactionSynchronizationExceptionOnCommit() {
 	public void testDeadLetterPublisherWhileTransactionActive() {
 		@SuppressWarnings("unchecked")
 		Producer<Object, Object> producer1 = mock(Producer.class);
+		given(producer1.send(any(), any())).willReturn(new SettableListenableFuture<>());
 		@SuppressWarnings("unchecked")
 		Producer<Object, Object> producer2 = mock(Producer.class);
+		given(producer2.send(any(), any())).willReturn(new SettableListenableFuture<>());
 		producer1.initTransactions();
 
 		@SuppressWarnings("unchecked")
@@ -329,6 +332,7 @@ public void testDeadLetterPublisherWhileTransactionActive() {
 				});
 
 		verify(producer1).beginTransaction();
+
 		verify(producer1).commitTransaction();
 		verify(producer1).close(any());
 		verify(producer2, never()).beginTransaction();
@@ -477,8 +481,10 @@ public void testAbort() {
 	public void testExecuteInTransactionNewInnerTx() {
 		@SuppressWarnings("unchecked")
 		Producer<Object, Object> producer1 = mock(Producer.class);
+		given(producer1.send(any(), any())).willReturn(new SettableListenableFuture<>());
 		@SuppressWarnings("unchecked")
 		Producer<Object, Object> producer2 = mock(Producer.class);
+		given(producer2.send(any(), any())).willReturn(new SettableListenableFuture<>());
 		producer1.initTransactions();
 		AtomicBoolean first = new AtomicBoolean(true);
 
@@ -530,26 +536,30 @@ protected Producer<Object, Object> createTransactionalProducerForPartition(Strin
 	@EnableTransactionManagement
 	public static class DeclarativeConfig {
 
-		@SuppressWarnings("rawtypes")
+		@SuppressWarnings({ "rawtypes", "unchecked" })
 		@Bean
-		public ProducerFactory pf() {
-			ProducerFactory pf = mock(ProducerFactory.class);
-			given(pf.transactionCapable()).willReturn(true);
-			given(pf.createProducer(isNull())).willReturn(producer1());
-			given(pf.createProducer(anyString())).willReturn(producer2());
-			return pf;
+		public Producer producer1() {
+			Producer mock = mock(Producer.class);
+			given(mock.send(any(), any())).willReturn(new SettableListenableFuture<>());
+			return mock;
 		}
 
-		@SuppressWarnings("rawtypes")
+		@SuppressWarnings({ "rawtypes", "unchecked" })
 		@Bean
-		public Producer producer1() {
-			return mock(Producer.class);
+		public Producer producer2() {
+			Producer mock = mock(Producer.class);
+			given(mock.send(any(), any())).willReturn(new SettableListenableFuture<>());
+			return mock;
 		}
 
 		@SuppressWarnings("rawtypes")
 		@Bean
-		public Producer producer2() {
-			return mock(Producer.class);
+		public ProducerFactory pf() {
+			ProducerFactory pf = mock(ProducerFactory.class);
+			given(pf.transactionCapable()).willReturn(true);
+			given(pf.createProducer(isNull())).willReturn(producer1());
+			given(pf.createProducer(anyString())).willReturn(producer2());
+			return pf;
 		}
 
 		@SuppressWarnings({ "rawtypes", "unchecked" })

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

@@ -97,6 +97,7 @@
 import org.springframework.transaction.support.AbstractPlatformTransactionManager;
 import org.springframework.transaction.support.DefaultTransactionStatus;
 import org.springframework.util.backoff.FixedBackOff;
+import org.springframework.util.concurrent.SettableListenableFuture;
 
 /**
  * @author Gary Russell
@@ -176,6 +177,7 @@ private void testConsumeAndProduceTransactionGuts(boolean chained, boolean handl
 		ConsumerFactory cf = mock(ConsumerFactory.class);
 		willReturn(consumer).given(cf).createConsumer("group", "", null, KafkaTestUtils.defaultPropertyOverrides());
 		Producer producer = mock(Producer.class);
+		given(producer.send(any(), any())).willReturn(new SettableListenableFuture<>());
 		final CountDownLatch closeLatch = new CountDownLatch(2);
 		willAnswer(i -> {
 			closeLatch.countDown();
@@ -413,6 +415,7 @@ public void testConsumeAndProduceTransactionExternalTM() throws Exception {
 		ConsumerFactory cf = mock(ConsumerFactory.class);
 		willReturn(consumer).given(cf).createConsumer("group", "", null, KafkaTestUtils.defaultPropertyOverrides());
 		Producer producer = mock(Producer.class);
+		given(producer.send(any(), any())).willReturn(new SettableListenableFuture<>());
 
 		final CountDownLatch closeLatch = new CountDownLatch(1);
 

spring-kafka/src/test/java/org/springframework/kafka/requestreply/ReplyingKafkaTemplateTests.java

@@ -88,6 +88,7 @@
 import org.springframework.messaging.support.MessageBuilder;
 import org.springframework.test.annotation.DirtiesContext;
 import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
+import org.springframework.util.concurrent.SettableListenableFuture;
 
 /**
  * @author Gary Russell
@@ -401,7 +402,7 @@ public void testAggregateOrphansNotStored() throws Exception {
 		willAnswer(invocation -> {
 			ProducerRecord rec = invocation.getArgument(0);
 			correlation.set(rec.headers().lastHeader(KafkaHeaders.CORRELATION_ID).value());
-			return null;
+			return new SettableListenableFuture<>();
 		}).given(producer).send(any(), any());
 		AggregatingReplyingKafkaTemplate template = new AggregatingReplyingKafkaTemplate(pf, container,
 				(list, timeout) -> true);