From 6401697f18976ece0d43a6fea46b721ec68a551e Mon Sep 17 00:00:00 2001 From: Soby Chacko Date: Wed, 17 Mar 2021 14:41:19 -0400 Subject: [PATCH] Array reification error with KStream[] in Kotlin Exclude Kafka Streams functions with KStream[] return type as this causes some array reification errors in Kotlin. See this issue for more details: https://github.com/spring-cloud/spring-cloud-stream-binder-kafka/issues/1044 Resolves https://github.com/spring-cloud/spring-cloud-function/issues/669 Resolves #670 --- .../cloud/function/context/FunctionRegistration.java | 3 +++ .../context/catalog/BeanFactoryAwareFunctionRegistry.java | 6 ++++-- .../cloud/function/context/catalog/FunctionTypeUtils.java | 7 +++++++ 3 files changed, 14 insertions(+), 2 deletions(-) diff --git a/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/FunctionRegistration.java b/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/FunctionRegistration.java index 5b7772ab6..536ce9ac2 100644 --- a/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/FunctionRegistration.java +++ b/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/FunctionRegistration.java @@ -125,6 +125,9 @@ public class FunctionRegistration implements BeanNameAware { public FunctionRegistration type(FunctionType type) { Type t = FunctionTypeUtils.discoverFunctionTypeFromClass(this.target.getClass()); + if (t == null) { // only valid for Kafka Stream KStream[] return type. + return null; + } FunctionType discoveredFunctionType = FunctionType.of(t); Class inputType = TypeResolver.resolveRawClass(discoveredFunctionType.getInputType(), null); Class outputType = TypeResolver.resolveRawClass(discoveredFunctionType.getOutputType(), null); diff --git a/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/catalog/BeanFactoryAwareFunctionRegistry.java b/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/catalog/BeanFactoryAwareFunctionRegistry.java index cdae8951e..c893c96bc 100644 --- a/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/catalog/BeanFactoryAwareFunctionRegistry.java +++ b/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/catalog/BeanFactoryAwareFunctionRegistry.java @@ -141,8 +141,10 @@ public class BeanFactoryAwareFunctionRegistry extends SimpleFunctionRegistry imp if (functionRegistration == null) { functionRegistration = new FunctionRegistration(functionCandidate, functionName).type(functionType); } - - this.register(functionRegistration); + // Certain Kafka Streams functions such as KStream[] return types could be null (esp when using Kotlin). + if (functionRegistration != null) { + this.register(functionRegistration); + } } else { if (logger.isDebugEnabled()) { diff --git a/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/catalog/FunctionTypeUtils.java b/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/catalog/FunctionTypeUtils.java index 5b2ff708e..c25cfe219 100644 --- a/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/catalog/FunctionTypeUtils.java +++ b/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/catalog/FunctionTypeUtils.java @@ -149,6 +149,13 @@ public final class FunctionTypeUtils { Assert.isTrue(isFunctional(functionalClass), "Type must be one of Supplier, Function or Consumer"); if (Function.class.isAssignableFrom(functionalClass)) { + for (Type superInterface : functionalClass.getGenericInterfaces()) { + if (superInterface != null && !superInterface.equals(Object.class)) { + if (superInterface.toString().contains("KStream") && ResolvableType.forType(superInterface).getGeneric(1).isArray()) { + return null; + } + } + } return TypeResolver.reify(Function.class, (Class>) functionalClass); } else if (Consumer.class.isAssignableFrom(functionalClass)) {