Binding superstream to MQTT Topic Exchange

[ricardoduncan]

Community Support Policy

• I have read RabbitMQ's Community Support Policy

RabbitMQ version used

4.0.2

How is RabbitMQ deployed?

Community Docker image

Steps to reproduce the behavior in question

Is it possible to have an MQTT producer send messages to a RabbitMQ MQTT topic exchange (using QoS 0 or 1) and then consumers receive these events over super-streams (connected to streams making use of partitioning with partition keys). I am using .Net and it seems that the super stream and partition keys are setup by the producer. However in my case the producer is an MQTT client and not a stream client. Is this topology possible with RabbitMQ?

[michaelklishin]

You can use a stream with publishers MQTT (as if it was a queue) but MQTT clients cannot set up the partitioning that a stream protocol client can.

Even then, you can use N streams with MQTT publishers.

[ricardoduncan]

Thanks, so although one can connect N streams to MQTT publishers, this will achieve a "fan out" design whereby a copy of each message is placed on each of the N streams? If so, it would not achieve our goal of wanting to partition a single super stream across many streams using a partition key (with publishing to the MQTT topic exchange/ super stream being done by an MQTT client).

[ricardoduncan]

As an alternative approach, can this be setup manually using configuration in the broker, as opposed to setting it up via the producer?

[ansd]

although one can connect N streams to MQTT publishers, this will achieve a "fan out" design whereby a copy of each message is placed on each of the N streams?

No, see #12623 (comment)

[ansd]

As an alternative approach, can this be setup manually using configuration in the broker, as opposed to setting it up via the producer?

Yes, see option 2 in #12623 (comment)

[ansd]

You have the following two options to achieve your use case:

1. Create N streams. You achieve partitioning by binding each stream with a different binding key to the topic exchange. For example, stream 1 could bind to the topic exchange matching all topics starting with topic level a. Stream 2 could bind to the topic exchange matching all topics starting with topic level b.
2. Enable the sharding plugin to have the exchange type x-modulus-hash being registered in RabbitMQ. Create a fixed number of streams and bind them to an instance of the x-modulus-hash exchange (the binding keys do not matter). Create an exchange-to-exchange binding from the x-modulus-hash exchange to the topic exchange with a binding key matching all topics (#). This will perform the hashing and modulus operations on the topic names within RabbitMQ.

If it's possible to define the partitioning scheme a priori (for example you know that roughly a similar amount of messages are sent to topics with prefix a and topics with prefix b), I would go with option 1 since it's simpler.

[ricardoduncan]

Thanks for the advice, I tested the second approach as we cannot determine the partitioning scheme a priori and it worked! I noticed one of the streams bound to the x-modulus hash exchange did not receive any messages though.

This design requires that each stream consumer is hard-coded to consume from a particular stream. Is there any way to define a pool of consumers bound to the group of streams and the system connects a single consumer to each of the streams. In this way our consumers don't require extra configuration to specify which stream to consume from. This makes sense as the partitioning scheme is not defined a priori - our consumers just care about processing a single partitioned stream of messages but they should not care about which stream the messages are consumed from.
We could code this if there is an api to indicate if a stream already has a consumer attached, then our consumer can just attach to the next free stream in the group.

[ansd]

Thanks for the advice, I tested the second approach as we cannot determine the partitioning scheme a priori and it worked!

Glad that it worked!

I noticed one of the streams bound to the x-modulus hash exchange did not receive any messages though.

This can happen depending on the distribution of topic names.
For example, if you bind three streams, but send with always the same four topic names, it may happen that all four topic names end up in the same two streams. If you increase the number of different topic names your clients publish with, all three streams should receive messages.

Is there any way to define a pool of consumers bound to the group of streams and the system connects a single consumer to each of the streams.

Yes, this is described in

• https://www.rabbitmq.com/blog/2022/07/13/rabbitmq-3-11-feature-preview-super-streams
• https://www.rabbitmq.com/blog/2022/07/05/rabbitmq-3-11-feature-preview-single-active-consumer-for-streams

[michaelklishin]

If x-modulo-hash is not enough, rabbitmq_consistent_hash exchange is a more powerful alternative that requires every queue or stream to be bound only once to the source exchange.

It is unlikely that you are going to need it with MQTT publishers but still. x-random-exchange is another, less evenly distributing options that was developed first.

[ricardoduncan]

Thanks all for the responses!
Since our use consists of MQTT publishers we are unable to define partitions for super streams as we will not be using the stream client. on the producer side, but the MQTT client. So it seems like having consumers consume from specific queues, which receive messages that are distributed via a hashing algorithm on the topic names, is the most feasible. While super streams in this context won't work.
I notice some additional latency (roughly 100ms) was added by the introduction of exchange to exchange binding (second exchange type being the x-modulus-hash exchange (I'm running 4 consumers on 4 separate threads on a 12 core machine with 8g memory). Is this mostly due to the hashing being applied?

[ansd]

While super streams in this context won't work.

A "super stream" is just N streams bound to an exchange. So, the topology you created by binding N streams to an x-modulus-hash exchange instance is your own variant of "super stream" if you want to name it this way.

Is this mostly due to the hashing being applied?

Hashing on the broker side is not for free, but hashing a routing key should definitely not take 100ms.