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.

Author: garyrussell

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

@@ -18,6 +18,8 @@
 
 import java.util.List;
 import java.util.Map;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Future;
 import java.util.concurrent.TimeUnit;
 
 import org.apache.commons.logging.Log;
@@ -32,6 +34,7 @@
 import org.apache.kafka.common.PartitionInfo;
 import org.apache.kafka.common.TopicPartition;
 
+import org.springframework.kafka.KafkaException;
 import org.springframework.kafka.support.LoggingProducerListener;
 import org.springframework.kafka.support.ProducerListener;
 import org.springframework.kafka.support.SendResult;
@@ -365,7 +368,8 @@ protected ListenableFuture<SendResult<K, V>> doSend(final ProducerRecord<K, V> p
 			this.logger.trace("Sending: " + producerRecord);
 		}
 		final SettableListenableFuture<SendResult<K, V>> future = new SettableListenableFuture<>();
-		producer.send(producerRecord, new Callback() {
+		Future<RecordMetadata> sendFuture =
+				producer.send(producerRecord, new Callback() {
 
 			@Override
 			public void onCompletion(RecordMetadata metadata, Exception exception) {
@@ -397,6 +401,19 @@ public void onCompletion(RecordMetadata metadata, Exception exception) {
 			}
 
 		});
+		// 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 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.
@@ -17,6 +17,7 @@
 package org.springframework.kafka.core;
 
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.BDDMockito.given;
 import static org.mockito.BDDMockito.willAnswer;
 import static org.mockito.Matchers.any;
 import static org.mockito.Mockito.inOrder;
@@ -36,6 +37,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
@@ -48,6 +50,7 @@ public class DefaultKafkaProducerFactoryTests {
 	@Test
 	public 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-2017 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.
@@ -17,6 +17,7 @@
 package org.springframework.kafka.core;
 
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.fail;
 import static org.mockito.Mockito.mock;
 import static org.springframework.kafka.test.assertj.KafkaConditions.key;
 import static org.springframework.kafka.test.assertj.KafkaConditions.partition;
@@ -33,10 +34,12 @@
 import org.apache.kafka.clients.consumer.Consumer;
 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.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;
@@ -46,6 +49,7 @@
 import org.junit.ClassRule;
 import org.junit.Test;
 
+import org.springframework.kafka.KafkaException;
 import org.springframework.kafka.support.Acknowledgment;
 import org.springframework.kafka.support.DefaultKafkaHeaderMapper;
 import org.springframework.kafka.support.KafkaHeaders;
@@ -309,4 +313,21 @@ public void testTemplateDisambiguation() throws Exception {
 		pf.destroy();
 	}
 
+	@Test
+	public 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);
+
+		try {
+			template.send("missing.topic", "foo");
+			fail("Expected exception");
+		}
+		catch (KafkaException e) {
+			assertThat(e.getCause()).isInstanceOf(TimeoutException.class);
+		}
+		pf.destroy();
+	}
+
 }

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

@@ -74,6 +74,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;
 
 import kafka.server.KafkaConfig;
 
@@ -195,7 +196,7 @@ public void testDeclarative() {
 		ProducerFactory producerFactory = ctx.getBean(ProducerFactory.class);
 		verify(producerFactory, times(2)).createProducer();
 		Producer producer1 = ctx.getBean("producer1", Producer.class);
-		Producer producer2 = ctx.getBean("producer1", Producer.class);
+		Producer producer2 = ctx.getBean("producer2", Producer.class);
 		InOrder inOrder = inOrder(producer1, producer2);
 		inOrder.verify(producer1).beginTransaction();
 		inOrder.verify(producer1).send(eq(new ProducerRecord("foo", "bar")), any(Callback.class));
@@ -344,8 +345,10 @@ public Producer<String, String> createProducer() {
 	public void testExcecuteInTransactionNewInnerTx() {
 		@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);
 
@@ -396,25 +399,29 @@ Producer<Object, Object> createTransactionalProducerForPartition() {
 	@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()).willReturn(producer1(), 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 producer1();
+		public ProducerFactory pf() {
+			ProducerFactory pf = mock(ProducerFactory.class);
+			given(pf.transactionCapable()).willReturn(true);
+			given(pf.createProducer()).willReturn(producer1(), producer2());
+			return pf;
 		}
 
 		@SuppressWarnings({ "rawtypes", "unchecked" })

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

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