Unable to connect to RabbitMQ Stream

[Lixuhuilll]

In what version(s) of Spring AMQP are you seeing this issue?

3.1.3

Describe the bug

When I use version 3.1.3 of SpringBoot and import spring-boot-starter-amqp and spring-rabbit-stream, once I listen to any Stream, it will fail to start.

Expected behavior

Start and listen to Stream normally.

Sample

https://github.com/Lixuhuilll/demo-rabbit-stream.git

[Lixuhuilll]

Fill in my RabbitMQ info：

[artembilan]

I have figured out that it works like this:

    @Bean
    EnvironmentBuilderCustomizer environmentBuilderCustomizer(RabbitProperties rabbitProperties) {
        RabbitProperties.Stream rabbitStreamProperties = rabbitProperties.getStream();
        return builder -> builder
                .addressResolver(add -> new Address(rabbitStreamProperties.getHost(), rabbitStreamProperties.getPort()));
    }

The plain host/port provided by Spring Boot in the RabbitStreamConfiguration doesn't work somehow.
Plus you have a problem in the @RabbitListener method.
since you are asking for the spring.rabbitmq.listener.stream.native-listener=true, you have to expect a stream message instead:

    @RabbitListener(id = "consumer", queues = "stream.test")
    public void consumerStream0(com.rabbitmq.stream.Message msg) {

Now after running your application and publishing message to that stream I got this in logs:

2023-09-12T12:04:57.284-04:00  INFO 17068 --- [pool-5-thread-1] c.e.d.DemoRabbitStreamApplication        : consumer-0-0:Data{Hello from issue #2522}

I'll discuss with RabbitMQ team what is wrong with plain host/port provided at the moment by Spring Boot.

[michaelklishin]

We are looking into what may be going on with @artembilan but it's worth mentioning that the RabbitMQ Stream protocol is different from the messaging protocols RabbitMQ supports in several important ways, in particular that clients perform cluster topology discovery.

Please see this post, in particular if you use a load balancer in your environment.

[Lixuhuilll]

Thanks for your help. After using Artem Bilan's method, RabbitMQ Stream can now be successfully connected.
I have only tested it locally on a single machine, not in a cluster.

[artembilan]

Thanks for confirmation.
I'm transferring this issue to Spring Boot since there is nothing wrong on Spring AMQP side and we need to revise how we auto-configure an Environment for RabbitMQ Stream for those host/port.
Or add more properties to map properly to Stream topology or so.

[wilkinsona]

Having read the "Connecting to Streams" blog post, given that there's neither a load balancer nor a cluster involved here I don't understand why Boot's configuration of just the host and port does not work. Shouldn't Rabbit's default behavior of using the address as-is work in this case, @michaelklishin?

The blog post describes configuring an address resolver the returns a hard-coded host and port as a workaround for when a load balancer is in use. As such, it doesn't sound like a generally applicable change that should be made in Boot. I think that users who find themselves in that situation should define their own rabbitStreamEnvironment bean that is an instance of com.rabbitmq.stream.Environment.

[michaelklishin]

@wilkinsona my best guess would be that the node reports hostname A but the client resolves hostname B, and they are not identical. On Slack, @acogoluegnes has recommended using

.addressResolver(add -> new Address(rabbitStreamProperties.getHost(), rabbitStreamProperties.getPort()));

which effectively ignores the server-advertised values.

Without debug logs we can only guess. And I'm not sure if we log the advertised hostname in the stream client. If not, we should.

[wilkinsona]

Thanks. That's similar to the recommendation in the blog post and I am still not sure that it's generally applicable such that the change should be made in Boot. It would mean that the server-advertised values are ignored by default. If that's appropriate default behavior, I would expect the stream client's default resolver to behave in this way, using the host and port configured on the com.rabbitmq.stream.Environment rather than returning the address as-is.

[Lixuhuilll]

I actually uploaded a log of connection failures in the sample if you need logs.

Lixuhuilll/demo-rabbit-stream@02e5ac9#diff-1e6dd7994a5ef4b580ff18d0a5f120e9edbb7ef1e178f2a774024724f4c8834b

[michaelklishin]

For environments where there's just one node or an LB is used, ignoring server-advertised values is fine. But if you want direct parallel connections because, well, that's the option that offers the best throughput, you have to make sure that the server advertises hosts that the clients will be able to connect to.

Which is why I've asked @artembilan for an easy way to configure Spring projects for both scenarios, with the LB one perhaps being a more reasonable default.

[acogoluegnes]

How is the broker started (Docker container or not)? Is it running locally?

Can you enable debug log level for com.rabbitmq.stream.impl and share the output?

[artembilan]

As I said in our Slack discussion the scenario is as simple as:

1. Local RabbitMQ server with enabled Stream plugin.
2. Add a queue with type stream.
3. Configured Spring Boot for localhost:5552 pair (essentially default from RabbitMQ and from Spring Boot).
4. App fails to connect on start.

So, as a beginner experience we would expect to have it working out-of-the-box with all those defaults.

[Lixuhuilll]

How is the broker started (Docker container or not)? Is it running locally?

Can you enable debug log level for com.rabbitmq.stream.impl and share the output?

I didn't use container,broker runs directly on the local machine. And I didn't modify any of the default configurations, only installed the required plugins.

This is a log with logging.level.com.rabbitmq.stream.impl=debug:
debug-log.log