@PollableBean doesn't support Kotlin lambda

[asinbow]

Describe the issue @PollableBean doesn't support Kotlin lambda while other spring-cloud-function features support it.

To Reproduce ✅ @PollableBean in the following code works as intended.

@SpringBootApplication
class StreamApplication {
	@Bean
	@PollableBean
	fun produceNames(): Supplier<Flux<String>> = Supplier {
		Flux.just(
			Human("human#foo"),
			Human("human#bar"),
			Human("human#baz"),
		)
	}

	@Bean
	fun logName(): (String) -> Unit = { name ->
		println("Received humans: $name")
	}
}

fun main(args: Array<String>) {
	runApplication<StreamApplication>(*args)
}

❌ But if using Kotlin lambda(kotlin.jvm.functions.Function0), it doesn't work.

@SpringBootApplication
class StreamApplication {
	@Bean
	@PollableBean // it doesn't work with Kotlin lambda
	fun produceNames(): () -> Flux<String> = {
		Flux.just(
			Human("human#foo"),
			Human("human#bar"),
			Human("human#baz"),
		)
	}
}

Version of the framework

• spring-cloud-stream: 3.2.4
• spring-cloud: 2021.0.3

Additional context After some debugging, I found it was caused by this check which returned null for Kotlin lambda: https://github.com/spring-cloud/spring-cloud-stream/blob/779ebfcf887faf386e4c610537362f69826b56c8/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/FunctionConfiguration.java#L382-L384

Just guess whether it should use functionWrapper.isSupplier() from: https://github.com/spring-cloud/spring-cloud-stream/blob/779ebfcf887faf386e4c610537362f69826b56c8/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/FunctionConfiguration.java#L235

Possible fix

• Option 1 pass in functionWrapper and check isSupplier.

return (functionWrapper != null ? functionWrapper.isSupplier() : factoryMethod.getReturnType().isAssignableFrom(Supplier.class))
    ? AnnotationUtils.findAnnotation(factoryMethod, PollableBean.class)
    : null;

• Option 2 check it in the caller https://github.com/spring-cloud/spring-cloud-stream/blob/779ebfcf887faf386e4c610537362f69826b56c8/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/FunctionConfiguration.java#L233

private PollableBean extractPollableAnnotation(...) {
  // don't check isSupplier here
  return AnnotationUtils.findAnnotation(factoryMethod, PollableBean.class)
}
PollableBean pollable = isSupplier ? extractPollableAnnotation(functionProperties, context, proxyFactory) : null;