spring-projectsGH-1883: Fix replyTimeout Logic

Resolves spring-projects#1883

NPE when `null` `replyTimeout` supplied (javadoc says it's allowed and means use default).
`replyTimeout` argument ignored on message-based methods.

**cherry-pick to 2.7.x**

Author: garyrussell

spring-kafka/src/main/java/org/springframework/kafka/requestreply/ReplyingKafkaTemplate.java

@@ -333,11 +333,12 @@ public <P> RequestReplyTypedMessageFuture<K, V, P> sendAndReceive(Message<?> mes
 
 	@SuppressWarnings("unchecked")
 	@Override
-	public <P> RequestReplyTypedMessageFuture<K, V, P> sendAndReceive(Message<?> message, Duration replyTimeout,
+	public <P> RequestReplyTypedMessageFuture<K, V, P> sendAndReceive(Message<?> message,
+			@Nullable Duration replyTimeout,
 			@Nullable ParameterizedTypeReference<P> returnType) {
 
 		RequestReplyFuture<K, V, R> future = sendAndReceive((ProducerRecord<K, V>) getMessageConverter()
-				.fromMessage(message, getDefaultTopic()));
+				.fromMessage(message, getDefaultTopic()), replyTimeout);
 		RequestReplyTypedMessageFuture<K, V, P> replyFuture =
 				new RequestReplyTypedMessageFuture<>(future.getSendFuture());
 		future.addCallback(
@@ -360,8 +361,12 @@ public RequestReplyFuture<K, V, R> sendAndReceive(ProducerRecord<K, V> record) {
 	}
 
 	@Override
-	public RequestReplyFuture<K, V, R> sendAndReceive(ProducerRecord<K, V> record, Duration replyTimeout) {
+	public RequestReplyFuture<K, V, R> sendAndReceive(ProducerRecord<K, V> record, @Nullable Duration replyTimeout) {
 		Assert.state(this.running, "Template has not been start()ed"); // NOSONAR (sync)
+		Duration timeout = replyTimeout;
+		if (timeout == null) {
+			timeout = this.defaultReplyTimeout;
+		}
 		CorrelationKey correlationId = this.correlationStrategy.apply(record);
 		Assert.notNull(correlationId, "the created 'correlationId' cannot be null");
 		Headers headers = record.headers();
@@ -383,7 +388,7 @@ public RequestReplyFuture<K, V, R> sendAndReceive(ProducerRecord<K, V> record, D
 			this.futures.remove(correlationId);
 			throw new KafkaException("Send failed", e);
 		}
-		scheduleTimeout(record, correlationId, replyTimeout);
+		scheduleTimeout(record, correlationId, timeout);
 		return future;
 	}
 

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

@@ -37,6 +37,7 @@
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicReference;
 
@@ -45,9 +46,11 @@
 import org.apache.kafka.clients.consumer.ConsumerRebalanceListener;
 import org.apache.kafka.clients.consumer.ConsumerRecord;
 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.ProducerConfig;
 import org.apache.kafka.clients.producer.ProducerRecord;
+import org.apache.kafka.clients.producer.RecordMetadata;
 import org.apache.kafka.common.TopicPartition;
 import org.apache.kafka.common.header.Headers;
 import org.apache.kafka.common.header.internals.RecordHeader;
@@ -676,6 +679,67 @@ public void withCustomHeaders() throws Exception {
 		}
 	}
 
+	@SuppressWarnings({ "rawtypes", "unchecked" })
+	@Test
+	void nullDuration() throws Exception {
+		ProducerFactory pf = mock(ProducerFactory.class);
+		Producer producer = mock(Producer.class);
+		willAnswer(invocation -> {
+			Callback callback = invocation.getArgument(1);
+			SettableListenableFuture<Object> future = new SettableListenableFuture<>();
+			future.set("done");
+			callback.onCompletion(new RecordMetadata(new TopicPartition("foo", 0), 0, 0, 0, null, 0, 0), null);
+			return future;
+		}).given(producer).send(any(), any());
+		given(pf.createProducer()).willReturn(producer);
+		GenericMessageListenerContainer container = mock(GenericMessageListenerContainer.class);
+		ContainerProperties properties = new ContainerProperties("two");
+		given(container.getContainerProperties()).willReturn(properties);
+		ReplyingKafkaTemplate template = new ReplyingKafkaTemplate(pf, container);
+		template.start();
+		Message<?> msg = MessageBuilder.withPayload("foo".getBytes())
+				.setHeader(KafkaHeaders.TOPIC, "foo")
+				.build();
+		// was NPE here
+		template.sendAndReceive(new ProducerRecord("foo", 0, "bar", "baz"), null).getSendFuture().get();
+	}
+
+	@SuppressWarnings({ "rawtypes", "unchecked" })
+	@Test
+	void requestTimeoutWithMessage() {
+		ProducerFactory pf = mock(ProducerFactory.class);
+		Producer producer = mock(Producer.class);
+		willAnswer(invocation -> {
+			return new SettableListenableFuture<>();
+		}).given(producer).send(any(), any());
+		given(pf.createProducer()).willReturn(producer);
+		GenericMessageListenerContainer container = mock(GenericMessageListenerContainer.class);
+		ContainerProperties properties = new ContainerProperties("two");
+		given(container.getContainerProperties()).willReturn(properties);
+		ReplyingKafkaTemplate template = new ReplyingKafkaTemplate(pf, container);
+		template.start();
+		Message<?> msg = MessageBuilder.withPayload("foo".getBytes())
+				.setHeader(KafkaHeaders.TOPIC, "foo")
+				.build();
+		long t1 = System.currentTimeMillis();
+		RequestReplyTypedMessageFuture<String, String, Foo> future = template.sendAndReceive(msg, Duration.ofMillis(10),
+				new ParameterizedTypeReference<Foo>() {
+				});
+		try {
+			future.get(10, TimeUnit.SECONDS);
+		}
+		catch (TimeoutException ex) {
+			fail("get timed out");
+		}
+		catch (InterruptedException e) {
+			Thread.currentThread().interrupt();
+			fail("Interrupted");
+		}
+		catch (ExecutionException e) {
+			assertThat(System.currentTimeMillis() - t1).isLessThan(3000L);
+		}
+	}
+
 	@Configuration
 	@EnableKafka
 	public static class Config {