Spring Cloud Stream 3 RabbitMQ consumer not working

Question

I'm able to make Spring+Rabbit work with the non-functional way (prior to 2.0?), but I'm trying to use with the functional pattern as the previous one is deprecated.

I've been following this doc: https://docs.spring.io/spring-cloud-stream/docs/3.1.0/reference/html/spring-cloud-stream.html#_binding_and_binding_names

The queue (consumer) is not being created in Rabbit with the new method. I can see the connection being created but without any consumer.

I have the following in my application.properties:

spring.cloud.stream.function.bindings.approved-in-0=approved
spring.cloud.stream.bindings.approved.destination=myTopic.exchange
spring.cloud.stream.bindings.approved.group=myGroup.approved
spring.cloud.stream.bindings.approved.consumer.back-off-initial-interval=2000
spring.cloud.stream.rabbit.bindings.approved.consumer.queueNameGroupOnly=true
spring.cloud.stream.rabbit.bindings.approved.consumer.bindingRoutingKey=myRoutingKey

which is replacing:

spring.cloud.stream.bindings.approved.destination=myTopic.exchange
spring.cloud.stream.bindings.approved.group=myGroup.approved
spring.cloud.stream.bindings.approved.consumer.back-off-initial-interval=2000
spring.cloud.stream.rabbit.bindings.approved.consumer.queueNameGroupOnly=true
spring.cloud.stream.rabbit.bindings.approved.consumer.bindingRoutingKey=myRoutingKey

And the new class

@Slf4j
@Service
public class ApprovedReceiver {

    @Bean
    public Consumer<String> approved() {
        // I also saw that it's recommended to not use Consumer, but use Function instead
        // https://docs.spring.io/spring-cloud-stream/docs/3.1.0/reference/html/spring-cloud-stream.html#_consumer_reactive
        return value -> log.info("value: {}", value);
    }

}

which is replacing

// BindableApprovedChannel.class
@Configuration
public interface BindableApprovedChannel {
    @Input("approved")
    SubscribableChannel getApproved();
}

// ApprovedReceiver.class
@Service
@EnableBinding(BindableApprovedChannel.class)
public class ApprovedReceiver {

    @StreamListener("approved")
    public void handleMessage(String payload) {
        log.info("value: {}", payload);
    }
}

Thanks!

Gary Russell Gary Russell · Accepted Answer · 2021-08-20T16:12:13

If you have multiple beans of type Function, Supplier or Consumer (which could be declared by third party libraries), the framework does not know which one to bind to.

Try setting the spring.cloud.function.definition property to approved.

https://docs.spring.io/spring-cloud-stream/docs/3.1.3/reference/html/spring-cloud-stream.html#spring_cloud_function

In the event you only have single bean of type java.util.function.[Supplier/Function/Consumer], you can skip the spring.cloud.function.definition property, since such functional bean will be auto-discovered. However, it is considered best practice to use such property to avoid any confusion. Some time this auto-discovery can get in the way, since single bean of type java.util.function.[Supplier/Function/Consumer] could be there for purposes other then handling messages, yet being single it is auto-discovered and auto-bound. For these rare scenarios you can disable auto-discovery by providing spring.cloud.stream.function.autodetect property with value set to false.

Spring Cloud Stream 3 RabbitMQ consumer not working

2 Answers