Wrong functionType is determined for the FunctionInvocationWrapper

[artembilan]

We have a bean like this:

@Bean
Scheduler reactiveScheduler() {
	return Schedulers.elastic();
}

where a target instance is a Supplier, according Schedulers.CachedScheduler extension. And this one is registered in the FunctionCatalog as a Supplier.
But at the same time it looks like functionType for the FunctionInvocationWrapper is determined from the bean definition type.
And in this case we have a plain:

public interface Scheduler extends Disposable {

so, that Disposable is set into the functionType, which leads to an exception like this when we perform function lookup:

Caused by: java.lang.IllegalArgumentException: Must be one of Supplier, Function, Consumer or FunctionRegistration. Was interface reactor.core.Disposable
    at org.springframework.util.Assert.isTrue(Assert.java:118)
    at org.springframework.cloud.function.context.catalog.FunctionTypeUtils.assertSupportedTypes(FunctionTypeUtils.java:390)
    at org.springframework.cloud.function.context.catalog.FunctionTypeUtils.getInputType(FunctionTypeUtils.java:176)
    at org.springframework.cloud.stream.function.FunctionConfiguration$FunctionChannelBindingInitializer.bindSimpleFunctions(FunctionConfiguration.java:377)
    at org.springframework.cloud.stream.function.FunctionConfiguration$FunctionChannelBindingInitializer.bindOrComposeSimpleFunctions(FunctionConfiguration.java:353)
    at org.springframework.cloud.stream.function.FunctionConfiguration$FunctionChannelBindingInitializer.lambda$afterPropertiesSet$0(FunctionConfiguration.java:261)
    at java.util.Spliterators$ArraySpliterator.forEachRemaining(Spliterators.java:948)
    at java.util.stream.ReferencePipeline$Head.forEach(ReferencePipeline.java:580)
    at org.springframework.cloud.stream.function.FunctionConfiguration$FunctionChannelBindingInitializer.afterPropertiesSet(FunctionConfiguration.java:251)
    at org.springframework.beans.factory.support.AbstractAutowireCapableBeanFactory.invokeInitMethods(AbstractAutowireCapableBeanFactory.java:1863)
    at org.springframework.beans.factory.support.AbstractAutowireCapableBeanFactory.initializeBean(AbstractAutowireCapableBeanFactory.java:1800)
    ... 64 common frames omitted

For me it sounds like we really need to check a bean definition type (not the target object) before registering into the FunctionCatalog.

Also would be great to improve that exception message to let us know which bean is guilty.

[elefeint]

Related bug in Spring Cloud Stream, for posterity: spring-cloud/spring-cloud-stream#1794

[elefeint]

I've been reading the documentation updates in Spring Cloud Stream. The project is moving away from being a light Spring Integration input/output router, and instead towards enabling pure microservices/functions, which got me to wonder if we are no longer using it philosophically correctly.

Instead of trying to figure out how to exclude the binder infrastructure bean that happens to look like a functional interface from being discovered, it may be preferable to explicitly mark it as "those are not the functional beans you are looking for".

Would Spring Cloud Function or Stream consider adding an annotation like @NonFunctional that would explicitly exclude a bean from being considered for the function registry?

[olegz]

@elefeint @artembilan

Elena

There are several issues at play here, so I decided to do a little writeup that I hope to turn into a blog. But first let me give you the short one:

The core issue is that in s-c-function we determine function type based on the signatures of the method, class etc., otherwise we’re giving you an option to configure all of it yourself via FunctionRegistration class. In your case the signature of the factory method in question is not a function type (not Supplier, Function or Consumer), however the instance returned by this factory method and the resulting bean is a function (Supplier) and that is the problem which we did not account for in s-c-function. So regardless of everything else, that is the root of the problem and it needs to be addressed.

Now the long one. . .

Interesting perception - "being a light Spring Integration input/output router. . .”, but I have to disagree, since it never has been about Spring Integration(SI) and/or router. In fact this statement shows part of the problem where SI (framework of choice to support some of the internal requirements of spring-cloud-stream(SCSt), was somehow perceived to be the core of spring-cloud-stream in such way that many perceive SCSt to be an extension to SI. It is not. It has always been about pure microservices. Just read the first sentence of our front page "Spring Cloud Stream is a framework for building highly scalable event-driven microservices connected with shared messaging systems."
If you abstract yourself far enough from knowing the internals of SCSt, you quickly realise that it is really a binding and activation framework. It binds a piece of code provided by the user to broker destinations supported by the provided binder and activates such code according to binder implementation (e.g., message arrival etc). That is pretty much it.

To function or not to function

Let's look at the following two code snippets

Annotation-based:

@SpringBootApplication
@EnableBinding(Processor.class)
public class SampleApplication  {
    @StreamListener(Processor.INPUT)
    @SendTo(Processor.OUTPUT)
    public String uppercase(String value) {
        return value.toUpperCase();
    }
}

Function based:

@SpringBootApplication
public class SampleApplication  {
    @Bean
    public Function<String, String> uppercase() {
        return value -> value.toUpperCase();
    }
}

Both are valid and fully functioning SCSt applications. Both are doing the same thing and both will produce the same result except that in the annotation-based example user has to be aware of SCSt abstractions (i.e., messaging, channels, bindings etc.) while the code has nothing to do with any of these abstractions. That raises a question why? Spring has always been about “you worry about functional requirements and we take care of non-functional (boilerplate)”. So, in he context of SCSt as a framework and its core goals of "binding and activation" we quickly realised that these abstractions are boilerplate and should not be leaked into the user’s code especially in the form of annotations as they contribute to binary dependency of such code on SCSt for no valid reason.
Also, given that the basis for most new frameworks within spring portfolio is spring-boot, think about the spring-boot's core message - dependency (e.g., JAR) includes auto-configuration which is effectively an opinion on how we (spring) believe things should be, while giving you a way to opt out. So, in this context why do you need to provide so many instructions especially thru annotations (EnableBinding, Processor, StreamListener etc) where one can easily extract/derive the same information (in the context of SCSt ) by simply following some convention. For example, Function bean in the context of SCSt is a Processor. We know that Processor has only one input destination and one output and we know their names so why do we need to explicitly state the known and the obvious? And so on. . . Also keep in mind that while deriving all of that we are still preserving the use of consumer/producer properties and all other configuration options. They still apply here allowing you to configure and reconfigure the same things as you would with the StreamListener.

So with that what I am also saying is that we are starting on our slow journey of moving away from annotation-based programming model and into a more spring-boot aligned opinionated model of clearly documented and intuitive convention with limited out-of-the-box configuration required from the user.

So finally to your case where something in your configuration (outside of your control) fits the category of function, but that is/was not your intention. Valid, interesting and we definitely need to provide a path to opt-out and that is what we’re working on, but I hope I was able to convince you that while you are correct in requesting such option, annotation would not be the right approach here, so I am considering a property.

Please feel free to provide any feedback

Cheers
Oleg

[elefeint]

That's a good blog post as is!
I don't need much convincing that less boilerplate is good -- I was looking at Spring Cloud GCP stream binder sample, and half of it consists of annotations that will go away in the new world.

[olegz]

So the fix is in and I believe it actually addresses the issue. Basically with this fix during the default discovery process (process where function definition is not provided) we no longer treat instance of Function, Supplier or Consumer as functional if its declared type is not actually Function, Supplier or Consumer (see example below):

@EnableAutoConfiguration
public static class SCF_GH_409ConfigurationAsSupplier {

	@Bean
	public Serializable blah() {
		return new Foo();
	}
		
        private static class Foo implements Supplier<Object>, Serializable {
		. . .
	}
}

So for the case in SCSt where it was firs discovered it will no longer be discovered in the first place.

Also, the above solution is valid IMHO for another reason. If we can't properly discover the actual type of the function (it's input/output argument types) we can't perform any type conversion effectively pushing a problem down the stack instead of addressing it.

[elefeint]

Excellent! I backed out the spring.cloud.stream.function.definition workaround, and our build is healthy.

Thank you for your help!