GH-3558: Kotlin DSL: propagate generics info (#3561)

Fixes #3558

Kotlin lambdas mostly used to configure endpoints in DSL manner
are not really Java lambdas, but rather anonymous classes implementing
respective Java interfaces.

While in most cases such classes carry generic info for their method impls
properly in Java, it is somehow doesn't work well for `GenericHandler`
implemented by Kotlin lambdas

* Wrap provided `GenericHandler` in the `BaseIntegrationFlowDefinition.handle()`
into a Java lambda and call `handle()` recursively to carry an expected type to
the `LambdaMessageProcessor`
* Fix `LambdaMessageProcessor` to handle `ClassUtils.isKotlinUnit()` result of
an invocation as a `null` reply

**Cherry-pick to `5.4.x` & `5.3.x`**

spring-integration-core/src/main/java/org/springframework/integration/dsl/BaseIntegrationFlowDefinition.java

@@ -1046,6 +1046,9 @@ public <P> B handle(Class<P> payloadType, GenericHandler<P> handler,
 		if (ClassUtils.isLambda(handler.getClass())) {
 			serviceActivatingHandler = new ServiceActivatingHandler(new LambdaMessageProcessor(handler, payloadType));
 		}
+		else if (payloadType != null) {
+			return handle(payloadType, handler::handle, endpointConfigurer);
+		}
 		else {
 			serviceActivatingHandler = new ServiceActivatingHandler(handler, ClassUtils.HANDLER_HANDLE_METHOD);
 		}

spring-integration-core/src/main/java/org/springframework/integration/handler/LambdaMessageProcessor.java

@@ -94,7 +94,11 @@ public Object processMessage(Message<?> message) {
 		Object[] args = buildArgs(message);
 
 		try {
-			return this.method.invoke(this.target, args);
+			Object result = this.method.invoke(this.target, args);
+			if (result != null && org.springframework.integration.util.ClassUtils.isKotlinUnit(result.getClass())) {
+				result = null;
+			}
+			return result;
 		}
 		catch (InvocationTargetException e) {
 			if (e.getTargetException() instanceof ClassCastException) {

spring-integration-core/src/test/kotlin/org/springframework/integration/dsl/KotlinDslTests.kt

@@ -1,5 +1,5 @@
 /*
- * Copyright 2020 the original author or authors.
+ * Copyright 2020-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.
@@ -211,8 +211,8 @@ class KotlinDslTests {
 	fun `no reply from handle`() {
 		val payloadReference = AtomicReference<String>()
 		val integrationFlow =
-			integrationFlow("handlerInputChanenl") {
-				handle<String> { payload, _ -> payloadReference.set(payload) }
+			integrationFlow("handlerInputChannel") {
+				handle<Message<String>> { message, _ -> payloadReference.set(message.payload) }
 			}
 
 		val registration = this.integrationFlowContext.registration(integrationFlow).register()