* polish DelegatingInvocableHandler and add javadoc

* polish `HandlerAdapter`
* change `InvocationResult` to record
* Optimization `MessagingMessageListenerAdapter.asyncFailure`

spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/DelegatingInvocableHandler.java

@@ -337,9 +337,15 @@ private boolean assignPayload(MethodParameter methodParameter, Class<?> payloadC
 				&& methodParameter.getParameterType().isAssignableFrom(payloadClass);
 	}
 
+	/**
+	 * Return the result of a method invocation by providing a result and payload.
+	 * @param result the result.
+	 * @param inboundPayload the payload.
+	 * @return the result of a method invocation.
+	 * @since 3.2
+	 */
 	@Nullable
 	public InvocationResult getInvocationResultFor(Object result, Object inboundPayload) {
-
 		InvocableHandlerMethod handler = findHandlerForPayload(inboundPayload.getClass());
 		if (handler != null) {
 			return new InvocationResult(result, this.handlerSendTo.get(handler),

spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/HandlerAdapter.java

@@ -101,7 +101,6 @@ public Object getBean() {
 
 	@Nullable
 	public InvocationResult getInvocationResultFor(Object result, @Nullable Object inboundPayload) {
-
 		if (this.delegatingHandler != null && inboundPayload != null) {
 			return this.delegatingHandler.getInvocationResultFor(result, inboundPayload);
 		}

spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/InvocationResult.java

@@ -21,39 +21,14 @@
 
 /**
  * The result of a method invocation.
+ * @param result the result.
+ * @param messageReturnType the message return type.
+ * @param sendTo the expression about sends topic.
  *
  * @author Gary Russell
  * @since 2.2
  */
-public final class InvocationResult {
-
-	@Nullable
-	private final Object result;
-
-	@Nullable
-	private final Expression sendTo;
-
-	private final boolean messageReturnType;
-
-	public InvocationResult(@Nullable Object result, @Nullable Expression sendTo, boolean messageReturnType) {
-		this.result = result;
-		this.sendTo = sendTo;
-		this.messageReturnType = messageReturnType;
-	}
-
-	@Nullable
-	public Object getResult() {
-		return this.result;
-	}
-
-	@Nullable
-	public Expression getSendTo() {
-		return this.sendTo;
-	}
-
-	public boolean isMessageReturnType() {
-		return this.messageReturnType;
-	}
+public record InvocationResult(@Nullable Object result, @Nullable Expression sendTo, boolean messageReturnType) {
 
 	@Override
 	public String toString() {

spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/MessagingMessageListenerAdapter.java

@@ -465,10 +465,10 @@ protected void handleResult(Object resultArg, Object request, @Nullable Acknowle
 				"a KafkaTemplate is required to support replies");
 
 		Object result = resultArg instanceof InvocationResult invocationResult ?
-				invocationResult.getResult() :
+				invocationResult.result() :
 				resultArg;
 		boolean messageReturnType = resultArg instanceof InvocationResult invocationResult ?
-				invocationResult.isMessageReturnType() :
+				invocationResult.messageReturnType() :
 				this.messageReturnType;
 
 		if (result instanceof CompletableFuture<?> completable) {
@@ -506,7 +506,7 @@ else if (monoPresent && result instanceof Mono<?> mono) {
 	private String evaluateReplyTopic(Object request, Object source, Object result) {
 		String replyTo = null;
 		if (result instanceof InvocationResult invResult) {
-			replyTo = evaluateTopic(request, source, result, invResult.getSendTo());
+			replyTo = evaluateTopic(request, source, result, invResult.sendTo());
 		}
 		else if (this.replyTopicExpression != null) {
 			replyTo = evaluateTopic(request, source, result, this.replyTopicExpression);
@@ -656,16 +656,16 @@ protected void acknowledge(@Nullable Acknowledgment acknowledgment) {
 
 	protected void asyncFailure(Object request, @Nullable Acknowledgment acknowledgment, Consumer<?, ?> consumer,
 			Throwable t, Message<?> source) {
+
 		try {
 			handleException(request, acknowledgment, consumer, source,
 					new ListenerExecutionFailedException(createMessagingErrorMessage(
 							"Async Fail", source.getPayload()), t));
-			return;
 		}
-		catch (Exception ex) {
+		catch (Throwable ex) {
+			this.logger.error(t, () -> "Future, Mono, or suspend function was completed with an exception for " + source);
+			acknowledge(acknowledgment);
 		}
-		this.logger.error(t, () -> "Future, Mono, or suspend function was completed with an exception for " + source);
-		acknowledge(acknowledgment);
 	}
 
 	protected void handleException(Object records, @Nullable Acknowledgment acknowledgment, Consumer<?, ?> consumer,