GH-3344: Treat kotlin.Unit return as null in MMIH (#3346)

* GH-3344: Treat kotlin.Unit return as null in MMIH

Fixes https://github.com/spring-projects/spring-integration/issues/3344

When function lambda doesn't return anything (e.g. a `void` method call is the last one),
Kotlin produces a `kotlin.Unit` instance as a return value which is not null and produced
as a reply message payload.

* Fix `MessagingMethodInvokerHelper` to treat a `kotlin.Unit` as `null` for reply
making Kotlin lambdas working the same way as Java lambdas when we don't return anything
from from there

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

* * Introduce `ClassUtils.isKotlinUnit(Class)` API;
use it in the `MessagingMethodInvokerHelper` instead of
`.getName().equals()`

* * Fix since on new `isKotlinUnit()` API

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

@@ -17,12 +17,9 @@
 package org.springframework.integration.dsl
 
 import assertk.assertThat
-import assertk.assertions.isEqualTo
-import assertk.assertions.isGreaterThanOrEqualTo
-import assertk.assertions.isInstanceOf
-import assertk.assertions.isNotNull
-import assertk.assertions.isTrue
-import assertk.assertions.size
+import assertk.assertions.*
+import org.apache.commons.logging.Log
+import org.apache.commons.logging.LogFactory
 import org.junit.jupiter.api.Test
 import org.springframework.beans.factory.BeanFactory
 import org.springframework.beans.factory.annotation.Autowired
@@ -51,6 +48,7 @@ import org.springframework.test.context.junit.jupiter.SpringJUnitConfig
 import reactor.core.publisher.Flux
 import reactor.test.StepVerifier
 import java.util.*
+import java.util.concurrent.atomic.AtomicReference
 import java.util.function.Function
 
 /**
@@ -210,6 +208,23 @@ class KotlinDslTests {
 		assertThat(payload).isInstanceOf(List::class.java).size().isGreaterThanOrEqualTo(1)
 	}
 
+	@Test
+	fun `no reply from handle`() {
+		val payloadReference = AtomicReference<String>()
+		val integrationFlow =
+				integrationFlow("handlerInputChanenl") {
+					handle<String> { payload, _ ->  payloadReference.set(payload) }
+				}
+
+		val registration = this.integrationFlowContext.registration(integrationFlow).register()
+
+		registration.inputChannel.send(GenericMessage("test"))
+
+		assertThat(payloadReference.get()).isEqualTo("test")
+
+		registration.destroy()
+	}
+
 	@Configuration
 	@EnableIntegration
 	class Config {