Using Kotlin with Kafka Branching functions causes type resolution to fail

[GFriedrich]

Describe the bug When using Kotlin functions for @Component beans causes the type resolution of the bean to fail.

Sample Use the example from https://github.com/spring-cloud/spring-cloud-stream-samples/tree/main/kafka-streams-samples/kafka-streams-branching and rewrite it to use Kotlin with a @Component function bean.

Running the example causes the app to fail with an exception like:

Caused by: java.lang.UnsupportedOperationException: Attempted to reify generic array type, whose generic component type could not be reified to some Class<?>. Handling for this case is not implemented
	at net.jodah.typetools.TypeResolver.reify(TypeResolver.java:449)
	at net.jodah.typetools.TypeResolver.reify(TypeResolver.java:430)
	at net.jodah.typetools.TypeResolver.reify(TypeResolver.java:402)
	at net.jodah.typetools.TypeResolver.reify(TypeResolver.java:205)
	at org.springframework.cloud.function.context.catalog.FunctionTypeUtils.discoverFunctionTypeFromClass(FunctionTypeUtils.java:195)
	at org.springframework.cloud.function.context.catalog.FunctionTypeUtils.discoverFunctionType(FunctionTypeUtils.java:366)

Not sure how to resolve this though as this is the result of the library used for type resolution to use some unimplemented code path. So likely one has to implement some sort of exceptional case for the Kafka branching solution here like it was already done for the Java code path which is handled in the same discoverFunctionTypeFromClass method just a few lines below.

Used Spring Cloud function library version: 4.0.3

[olegz]

I don't know what does rewrite it to use Kotlin means. You are assuming everyone would end up with identical code. Rewriting anything could be done in many different ways Please provide a solid way to reproduce it (samople that you can push to github)

[GFriedrich]

Instead of using this Java bean in the example:

@Bean
@SuppressWarnings("unchecked")
public Function<KStream<Object, String>, KStream<?, WordCount>[]> process() {
  ...
}

you would need to use a @Component definition using Kotlin function types like:

@Component
class KafkaStreamBean : (KStream<Object, String>) -> Array<KStream<*, WordCount>> {
  ...
}