Problem in deserialization when defining consumers with generics types

[ferblaca]

Describe the issue
When implementing a consumer class with generic types that implement java.util.Consumer, we have observed that depending on how the consumer bean is defined, the deserialization of messages works correctly or not.

We have created this repository that describes this situation, defining the Consumer beans "correctly" or "incorrectly".

We understand that it should work both ways. Additionally, it is a good practice and recommended by Spring to define the implementation as the bean output rather than the interface.

To Reproduce
Steps to reproduce the behavior:

1. Define a Class that implements Consumer with types variables

public class GenericConsumer<T> implements Consumer<T> {
...
}

2. Define a bean of this class with this signature:

    @Bean
    public GenericConsumer<ProductCompleteDTO> productConsumer_KO() {
        return new GenericConsumer<>("productConsumer_KO");
    }

3. Send a message.

Version of the framework
Spring-Boot 3.4.2
Spring Cloud 2024.0.0

Observed outcome
The message is not deserialized and arrives in byte[].
For batch mode the message payload is null in the list of messages.

Expected behavior
The message is deserialized correctly by the converter.

Additional context
The same behavior occurs for consumer in batch-mode.
Works OK when the bean is defined in this manner:

    @Bean
    public Consumer<ProductCompleteDTO> productConsumer_OK() {
        return new GenericConsumer<>("productConsumer_OK");
    }

[cbrachem]

We have encountered the same issue very recently when trying to upgrade the Spring Boot version in one of our applications. This way of using a subclass (or a sub-interface) of Consumer still works in Spring Cloud Stream 4.1.5. I believe the change was introduced when switching the Spring Cloud Function dependency to 4.2.0. It appears that Spring Cloud Function recently changed the way type resolution works. During debugging, I could observe that the problem already happens when the consumer is registered with the SimpleFunctionRegistry during application startup: the resolved input type is Object.
Now, when a message arrives and Spring Cloud Stream wants to invoke a function, it looks it up in the SimpleFunctionRegistry and the incoming message should be deserialised to the actual input type in FunctionInvocationWrapper#convertInputIfNecessary, which doesn't convert the input, because, well, a byte array is an Object.

The actual change in behaviour is in FunctionTypeUtils#getInputType. The call to GenericTypeResolver.resolveType here returns a TypeVariable which makes getInputType return Object.
spring-projects/spring-framework#33887 seems to be related.

I am not sure what's the best way to go tackle this issue or if the way we're using Spring Cloud Stream is even considered correct, but it has been working well for us so far. (We use a sub-interface of Consumer<T> to set internal auth info for our cloud stream consumers via AOP.)

[olegz]

Can you please try it with Spring Cloud 2024.0.1. There were regressions with some of the refactoring we did

[ferblaca]

Hi @olegz,

Unfortunately, the same problem persists after updating to Spring Cloud 2024.0.1. To reproduce it, simply start the application found in the repository: