GH-1728: ReplyingKT Hook for User Errors

Resolves #1728

Add a hook allowing detection of server errors and complete the future
exceptionally.

**cherry-pick to 2.6.x (minus whatsnew.adoc)**

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

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2018-2020 the original author or authors.
+ * Copyright 2018-2021 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.
@@ -104,6 +104,8 @@ public class ReplyingKafkaTemplate<K, V, R> extends KafkaTemplate<K, V> implemen
 
 	private String replyPartitionHeaderName = KafkaHeaders.REPLY_PARTITION;
 
+	private Function<ConsumerRecord<?, ?>, Exception> replyErrorChecker = rec -> null;
+
 	private volatile boolean running;
 
 	private volatile boolean schedulerInitialized;
@@ -263,6 +265,16 @@ public void setReplyPartitionHeaderName(String replyPartitionHeaderName) {
 		this.replyPartitionHeaderName = replyPartitionHeaderName;
 	}
 
+	/**
+	 * Set a function to examine replies for an error returned by the server.
+	 * @param replyErrorChecker the error checker function.
+	 * @since 2.6.7
+	 */
+	public void setReplyErrorChecker(Function<ConsumerRecord<?, ?>, Exception> replyErrorChecker) {
+		Assert.notNull(replyErrorChecker, "'replyErrorChecker' cannot be null");
+		this.replyErrorChecker = replyErrorChecker;
+	}
+
 	@Override
 	public void afterPropertiesSet() {
 		if (!this.schedulerSet && !this.schedulerInitialized) {
@@ -408,12 +420,10 @@ public void onMessage(List<ConsumerRecord<K, R>> data) {
 				}
 				else {
 					boolean ok = true;
-					if (record.value() == null) {
-						DeserializationException de = checkDeserialization(record, this.logger);
-						if (de != null) {
-							ok = false;
-							future.setException(de);
-						}
+					Exception exception = checkForErrors(record);
+					if (exception != null) {
+						ok = false;
+						future.setException(exception);
 					}
 					if (ok) {
 						this.logger.debug(() -> "Received: " + record + WITH_CORRELATION_ID + correlationKey);
@@ -424,6 +434,24 @@ public void onMessage(List<ConsumerRecord<K, R>> data) {
 		});
 	}
 
+	/**
+	 * Check for errors in a reply. The default implementation checks for {@link DeserializationException}s
+	 * and invokes the {@link #setReplyErrorChecker(Function) replyErrorChecker} function.
+	 * @param record the record.
+	 * @return the exception, or null if none.
+	 * @since 2.6.7
+	 */
+	@Nullable
+	protected Exception checkForErrors(ConsumerRecord<K, R> record) {
+		if (record.value() == null || record.key() == null) {
+			DeserializationException de = checkDeserialization(record, this.logger);
+			if (de != null) {
+				return de;
+			}
+		}
+		return this.replyErrorChecker.apply(record);
+	}
+
 	/**
 	 * Return a {@link DeserializationException} if either the key or value failed
 	 * deserialization; null otherwise. If you need to determine whether it was the key or

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2018-2020 the original author or authors.
+ * Copyright 2018-2021 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.
@@ -110,7 +110,8 @@
 		ReplyingKafkaTemplateTests.H_REPLY, ReplyingKafkaTemplateTests.H_REQUEST,
 		ReplyingKafkaTemplateTests.I_REPLY, ReplyingKafkaTemplateTests.I_REQUEST,
 		ReplyingKafkaTemplateTests.J_REPLY, ReplyingKafkaTemplateTests.J_REQUEST,
-		ReplyingKafkaTemplateTests.K_REPLY, ReplyingKafkaTemplateTests.K_REQUEST })
+		ReplyingKafkaTemplateTests.K_REPLY, ReplyingKafkaTemplateTests.K_REQUEST,
+		ReplyingKafkaTemplateTests.L_REPLY, ReplyingKafkaTemplateTests.L_REQUEST })
 public class ReplyingKafkaTemplateTests {
 
 	public static final String A_REPLY = "aReply";
@@ -157,6 +158,10 @@ public class ReplyingKafkaTemplateTests {
 
 	public static final String K_REQUEST = "kRequest";
 
+	public static final String L_REPLY = "lReply";
+
+	public static final String L_REQUEST = "lRequest";
+
 	@Autowired
 	private EmbeddedKafkaBroker embeddedKafka;
 
@@ -203,6 +208,32 @@ public void testGood() throws Exception {
 		}
 	}
 
+	@Test
+	void userDefinedException() throws Exception {
+		ReplyingKafkaTemplate<Integer, String, String> template = createTemplate(L_REPLY);
+		try {
+			template.setDefaultReplyTimeout(Duration.ofSeconds(30));
+			template.setReplyErrorChecker(record -> {
+				org.apache.kafka.common.header.Header error = record.headers().lastHeader("serverSentAnError");
+				if (error != null) {
+					return new IllegalStateException(new String(error.value()));
+				}
+				else {
+					return null;
+				}
+			});
+			ProducerRecord<Integer, String> record = new ProducerRecord<>(L_REQUEST, null, null, null, "foo", null);
+			assertThatExceptionOfType(ExecutionException.class)
+					.isThrownBy(() -> template.sendAndReceive(record).get(10, TimeUnit.SECONDS))
+					.withCauseExactlyInstanceOf(IllegalStateException.class)
+					.withMessageContaining("user error");
+		}
+		finally {
+			template.stop();
+			template.destroy();
+		}
+	}
+
 	@Test
 	void testConsumerRecord() throws Exception {
 		ReplyingKafkaTemplate<Integer, String, String> template = createTemplate(K_REPLY);
@@ -728,6 +759,14 @@ public String handleK(ConsumerRecord<String, String> in) {
 			return in.value().toUpperCase();
 		}
 
+		@KafkaListener(id = L_REQUEST, topics = L_REQUEST)
+		@SendTo  // default REPLY_TOPIC header
+		public Message<String> handleL(String in) throws InterruptedException {
+			return MessageBuilder.withPayload(in.toUpperCase())
+					.setHeader("serverSentAnError", "user error")
+					.build();
+		}
+
 	}
 
 	@KafkaListener(topics = C_REQUEST, groupId = C_REQUEST)

src/reference/asciidoc/kafka.adoc

@@ -637,6 +637,46 @@ Note that we can use Boot's auto-configured container factory to create the repl
 If a non-trivial deserializer is being used for replies, consider using an <<error-handling-deserializer,`ErrorHandlingDeserializer`>> that delegates to your configured deserializer.
 When so configured, the `RequestReplyFuture` will be completed exceptionally and you can catch the `ExecutionException`, with the `DeserializationException` in its `cause` property.
 
+Starting with version 2.6.7, in addition to detecting `DeserializationException` s, the template will call the `replyErrorChecker` function, if provided.
+If it returns an exception, the future will be completed exceptionally.
+
+Here is an example:
+
+====
+[source, java]
+----
+template.setReplyErrorChecker(record -> {
+    Header error = record.headers().lastHeader("serverSentAnError");
+    if (error != null) {
+        return new MyException(new String(error.value()));
+    }
+    else {
+        return null;
+    }
+});
+
+...
+
+RequestReplyFuture<Integer, String, String> future = template.sendAndReceive(record);
+try {
+    future.getSendFuture().get(10, TimeUnit.SECONDS); // send ok
+    ConsumerRecord<Integer, String> consumerRecord = future.get(10, TimeUnit.SECONDS);
+    ...
+}
+catch (InterruptedException e) {
+    ...
+}
+catch (ExecutionException e) {
+    if (e.getCause instanceof MyException) {
+        ...
+    }
+}
+catch (TimeoutException e) {
+    ...
+}
+----
+====
+
 The template sets a header (named `KafkaHeaders.CORRELATION_ID` by default), which must be echoed back by the server side.
 
 In this case, the following `@KafkaListener` application responds:
@@ -842,6 +882,7 @@ NOTE: If you use an <<error-handling-deserializer,`ErrorHandlingDeserializer`>>
 Instead, the record (with a `null` value) will be returned intact, with the deserialization exception(s) in headers.
 It is recommended that applications call the utility method `ReplyingKafkaTemplate.checkDeserialization()` method to determine if a deserialization exception occurred.
 See its javadocs for more information.
+The `replyErrorChecker` is also not called for this aggregating template; you should perform the checks on each element of the reply.
 
 [[receiving-messages]]
 ==== Receiving Messages