GH-2470 Verify Kotlin lambda working with PollableBean

Resolves #2470

core/spring-cloud-stream/pom.xml

@@ -9,7 +9,11 @@
 	<name>spring-cloud-stream</name>
 	<description>Messaging Microservices with Spring Integration</description>
 
-	<parent>
+    <properties>
+        <kotlin.version>1.7.20-Beta</kotlin.version>
+    </properties>
+
+    <parent>
 		<groupId>org.springframework.cloud</groupId>
 		<artifactId>spring-cloud-stream-core</artifactId>
 		<version>3.2.5-SNAPSHOT</version>
@@ -81,7 +85,18 @@
 			<artifactId>spring-kafka</artifactId>
 			<scope>test</scope>
 		</dependency>
-	</dependencies>
+        <dependency>
+            <groupId>org.jetbrains.kotlin</groupId>
+            <artifactId>kotlin-stdlib-jdk8</artifactId>
+            <version>${kotlin.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>org.jetbrains.kotlin</groupId>
+            <artifactId>kotlin-test</artifactId>
+            <version>${kotlin.version}</version>
+            <scope>test</scope>
+        </dependency>
+    </dependencies>
 
 	<build>
 		<plugins>
@@ -102,7 +117,57 @@
 					</execution>
 				</executions>
 			</plugin>
-		</plugins>
+            <plugin>
+                <groupId>org.jetbrains.kotlin</groupId>
+                <artifactId>kotlin-maven-plugin</artifactId>
+                <version>${kotlin.version}</version>
+                <executions>
+                    <execution>
+                        <id>compile</id>
+                        <phase>compile</phase>
+                        <goals>
+                            <goal>compile</goal>
+                        </goals>
+                        <configuration>
+                            <sourceDirs>
+                                <source>src/main/java</source>
+                                <source>src/test/kotlin</source>
+                            </sourceDirs>
+                        </configuration>
+                    </execution>
+                    <execution>
+                        <id>test-compile</id>
+                        <phase>test-compile</phase>
+                        <goals>
+                            <goal>test-compile</goal>
+                        </goals>
+                    </execution>
+                </executions>
+                <configuration>
+                    <jvmTarget>1.8</jvmTarget>
+                </configuration>
+            </plugin>
+            <plugin>
+                <groupId>org.apache.maven.plugins</groupId>
+                <artifactId>maven-compiler-plugin</artifactId>
+                <executions>
+                    <execution>
+                        <id>compile</id>
+                        <phase>compile</phase>
+                        <goals>
+                            <goal>compile</goal>
+                        </goals>
+                    </execution>
+                    <execution>
+                        <id>testCompile</id>
+                        <phase>test-compile</phase>
+                        <goals>
+                            <goal>testCompile</goal>
+                        </goals>
+                    </execution>
+                </executions>
+            </plugin>
+        </plugins>
 	</build>
 
 </project>

core/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/kotlin/KotlinConfigurationTests.java

@@ -0,0 +1,48 @@
+/*
+ * Copyright 2022-2022 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.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.cloud.stream.kotlin;
+
+import org.junit.jupiter.api.Test;
+
+import org.springframework.boot.WebApplicationType;
+import org.springframework.boot.builder.SpringApplicationBuilder;
+import org.springframework.cloud.stream.binder.test.OutputDestination;
+import org.springframework.cloud.stream.binder.test.TestChannelBinderConfiguration;
+import org.springframework.context.ConfigurableApplicationContext;
+import org.springframework.messaging.Message;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+public class KotlinConfigurationTests {
+
+	@Test
+	void testKotlinSupplierPollableBean() {
+		try (ConfigurableApplicationContext context = new SpringApplicationBuilder(
+			TestChannelBinderConfiguration.getCompleteConfiguration(KotlinTestConfiguration.class))
+				.web(WebApplicationType.NONE)
+				.run("--spring.jmx.enabled=false", "--spring.cloud.function.definition=produceNames")) {
+
+			OutputDestination output = context.getBean(OutputDestination.class);
+			Message<byte[]> result = output.receive(1000, "produceNames-out-0");
+			assertThat(result.getPayload()).isEqualTo("Ricky".getBytes());
+			result = output.receive(1000, "produceNames-out-0");
+			assertThat(result.getPayload()).isEqualTo("Julien".getBytes());
+			result = output.receive(1000, "produceNames-out-0");
+			assertThat(result.getPayload()).isEqualTo("Bubbles".getBytes());
+		}
+	}
+}

core/spring-cloud-stream/src/test/kotlin/org/springframework/cloud/stream/kotlin/KotlinTestConfiguration.kt

@@ -0,0 +1,36 @@
+/*
+ * Copyright 2022-2022 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.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.cloud.stream.kotlin
+
+import org.springframework.boot.autoconfigure.EnableAutoConfiguration
+import org.springframework.cloud.function.context.PollableBean
+import org.springframework.context.annotation.Configuration
+import reactor.core.publisher.Flux
+
+@Configuration
+@EnableAutoConfiguration
+open class KotlinTestConfiguration {
+
+	@PollableBean // it doesn't work with Kotlin lambda
+	open fun produceNames(): () -> Flux<String> = {
+		Flux.just(
+			"Ricky",
+			"Julien",
+			"Bubbles"
+		)
+	}
+
+}