Stream reactive #2038 commit introduced breaking changes

[hshimalaya]

https://github.com/spring-cloud/spring-cloud-sleuth/blob/2d72833e7d6f0e97333918da9a762c773c8a22d8/spring-cloud-sleuth-instrumentation/src/main/java/org/springframework/cloud/sleuth/instrument/messaging/TraceFunctionAroundWrapper.java#L212

Issue : Our use case was to have an app "sleuth enabled" and bind a stream function in this app with an output type as a "Tuple of publishers" instead of Publisher. This used to work fine till spring-cloud-sleuth-instrumentation:3.0.4. After changes were made in revision "Stream reactive #2038" onwards, the app fails during startup with below class cast exception (while using instrumentation:3.1.2):

Exception in thread "main" org.springframework.beans.factory.BeanCreationException: Error creating bean with name 'functionInitializer' defined in class path resource [org/springframework/cloud/stream/function/FunctionConfiguration.class]: Invocation of init method failed; nested exception is java.lang.ClassCastException: class reactor.util.function.Tuple2 cannot be cast to class org.reactivestreams.Publisher (reactor.util.function.Tuple2 and org.reactivestreams.Publisher are in unnamed module of loader 'app')Exception in thread "main" org.springframework.beans.factory.BeanCreationException: Error creating bean with name 'functionInitializer' defined in class path resource [org/springframework/cloud/stream/function/FunctionConfiguration.class]: Invocation of init method failed; nested exception is java.lang.ClassCastException: class reactor.util.function.Tuple2 cannot be cast to class org.reactivestreams.Publisher (reactor.util.function.Tuple2 and org.reactivestreams.Publisher are in unnamed module of loader 'app') at org.springframework.beans.factory.support.AbstractAutowireCapableBeanFactory.initializeBean(AbstractAutowireCapableBeanFactory.java:1804) at org.springframework.beans.factory.support.AbstractAutowireCapableBeanFactory.doCreateBean(AbstractAutowireCapableBeanFactory.java:620) at org.springframework.beans.factory.support.AbstractAutowireCapableBeanFactory.createBean(AbstractAutowireCapableBeanFactory.java:542) at org.springframework.beans.factory.support.AbstractBeanFactory.lambda$doGetBean$0(AbstractBeanFactory.java:335) at org.springframework.beans.factory.support.DefaultSingletonBeanRegistry.getSingleton(DefaultSingletonBeanRegistry.java:234) at org.springframework.beans.factory.support.AbstractBeanFactory.doGetBean(AbstractBeanFactory.java:333) at org.springframework.beans.factory.support.AbstractBeanFactory.getBean(AbstractBeanFactory.java:208) at org.springframework.beans.factory.support.DefaultListableBeanFactory.preInstantiateSingletons(DefaultListableBeanFactory.java:953) at org.springframework.context.support.AbstractApplicationContext.finishBeanFactoryInitialization(AbstractApplicationContext.java:918) at org.springframework.context.support.AbstractApplicationContext.refresh(AbstractApplicationContext.java:583) at org.springframework.boot.web.reactive.context.ReactiveWebServerApplicationContext.refresh(ReactiveWebServerApplicationContext.java:64) at org.springframework.boot.SpringApplication.refresh(SpringApplication.java:745) at org.springframework.boot.SpringApplication.refreshContext(SpringApplication.java:420) at org.springframework.boot.SpringApplication.run(SpringApplication.java:307) at org.springframework.boot.SpringApplication.run(SpringApplication.java:1317) at org.springframework.boot.SpringApplication.run(SpringApplication.java:1306) at com.jpmorgan.gwm.advisorconnectsuite.service.profile.stream.source.ConnectApplication.main(ConnectApplication.java:19)Caused by: java.lang.ClassCastException: class reactor.util.function.Tuple2 cannot be cast to class org.reactivestreams.Publisher (reactor.util.function.Tuple2 and org.reactivestreams.Publisher are in unnamed module of loader 'app') at org.springframework.cloud.sleuth.instrument.messaging.TraceFunctionAroundWrapper.reactorFluxStream(TraceFunctionAroundWrapper.java:212) at org.springframework.cloud.sleuth.instrument.messaging.TraceFunctionAroundWrapper.reactorStream(TraceFunctionAroundWrapper.java:137) at org.springframework.cloud.sleuth.instrument.messaging.TraceFunctionAroundWrapper.doApply(TraceFunctionAroundWrapper.java:102) at org.springframework.cloud.function.context.catalog.FunctionAroundWrapper.apply(FunctionAroundWrapper.java:47) at org.springframework.cloud.function.context.catalog.SimpleFunctionRegistry$1.doApply(SimpleFunctionRegistry.java:257) at org.springframework.cloud.function.context.catalog.SimpleFunctionRegistry$FunctionInvocationWrapper.apply(SimpleFunctionRegistry.java:551) at org.springframework.cloud.stream.function.PartitionAwareFunctionWrapper.apply(PartitionAwareFunctionWrapper.java:84) at org.springframework.cloud.stream.function.FunctionConfiguration$FunctionToDestinationBinder.bindFunctionToDestinations(FunctionConfiguration.java:551) at org.springframework.cloud.stream.function.FunctionConfiguration$FunctionToDestinationBinder.afterPropertiesSet(FunctionConfiguration.java:457) at org.springframework.beans.factory.support.AbstractAutowireCapableBeanFactory.invokeInitMethods(AbstractAutowireCapableBeanFactory.java:1863) at org.springframework.beans.factory.support.AbstractAutowireCapableBeanFactory.initializeBean(AbstractAutowireCapableBeanFactory.java:1800) ... 16 more

The exception occurred at code line mentioned at top of this description. Requesting help on this. Please let me know if anything else needed.

[marcingrzejszczak]

Do you have an example that replicates this problem?

[hshimalaya]

Hey Marcin,

We have created custom stream function which has one inbound and multiple outbound topics (the correct outbound is selected at runtime based on information received from inbound using predicates)

We have created custom framework to register this stream function using Spring BeanDefinitionrRegistry with function input type as Flux<Message> and output as Tuple2<Flux,Flux> or Tuple3<Flux,Flux,Flux> and so on, (Tuple arity based on the number of outbounds we wanted in this one stream function using as many number of predicates)

This Tuple is now being cast to a Publisher and errors out. The expected output type was an Object earlier. Now it is a publisher.

Any help would be greatly appreciated. Please let me know if you need more info on this.

Regards, Himalaya

[marcingrzejszczak]

I understand but it would be great to see an example that we can analyse

[ShubhamKale94]

We're facing same issue, tried searching historical issues. https://github.com/spring-cloud/spring-cloud-sleuth/issues/2080 is closed, but user had raised exact same concern.

Publisher seems to be compatible now, but Iterable or Tuple still throw ClassCastException

[marcingrzejszczak]

Can you please create a sample that replicates this issue? Of course as a workaround please disable sleuth for reactive stream

[spring-cloud-issues]

If you would like us to look at this issue, please provide the requested information. If the information is not provided within the next 7 days this issue will be closed.

[spring-cloud-issues]

Closing due to lack of requested feedback. If you would like us to look at this issue, please provide the requested information and we will re-open the issue.