RabbitMQ Streams feature support

[ArgoTown]

Do you have a plan to support Rabbit MQ new feature Streams ? Any deadline ?

[NooNameR]

I do not see .NET client for it. Before it releases - probably not, but it would be perfect candidate for MT Rider

[chlumper]

They are working on a first .NET client https://github.com/rabbitmq/rabbitmq-stream-dotnet-client
(from https://blog.rabbitmq.com/posts/2021/07/rabbitmq-streams-first-application/).

[briandrennan]

What exactly will it take to get stream support? I imagine some code needs to be written, and some documentation created... Is there a place to view a MassTransit-oriented design doc? I'd love to help contribute, as we'd like to use this feature in our environment.

[phatboyg]

Well, according to the streams support documentation, there are a bunch of new parameters that need to be specified for both queues and consumers. An approach would need to be determined of how to present streams to developers, how to keep track of offsets, etc.

[phatboyg]

I don't actually think it needs to be a rider, it's just a different way to configure the endpoint. Streams are just queues in RMQ.

[NooNameR]

gotcha, make sense then

[briandrennan]

I'd like to volunteer to take a stab at what this might look like, and hopefully help move it along in implementation. Obviously, figuring out an interface from MassTransit when declaring consumers in an IRabbitMqConfigurator is a worthwhile starting point. Based on my current experience with MassTransit, I suspect that would look something like this:

bus.UsingRabbitMq((busRegistration, rabbit) =>
{
    rabbit.ReceiveEndpoint("my-receive-endpoint", e =>
    {
        // Sets `x-queue-type: stream`
        e.SetStreamQueue();

        // These are each optional, and can be configured by either a broker policy
        // or client code.

        // Sets the `x-max-length-bytes` queue argument.
        e.SetStreamMaxLengthBytes(1_000_000);
        // Sets the `x-max-age` queue argument. `StreamAgeUnits` is an enum that ensures
        // only valid strings are passed to the queue (Y, M, D, h, m, s). Example here is one month.
        e.SetStreamMaxAge(1, StreamAgeUnits.Month);
        // Sets the `x-stream-max-segment-size-bytes` queue argument.
        e.SetStreamMaxSegmentSizeBytes(1_000_0000;
    });
});

I guess I'd look to you @phatboyg for some guidance on where best to get started if I wanted to make a contribution?

[phatboyg]

Something more like:

e.Stream(s =>
{
    s.MaxLengthBytes = 1_000_000;
    s.MaxAge = ...;
    etc.
});

Though I don't really know all the details of how a stream is customized. The consumer interface seems different as well, so I dunno.

[briandrennan]

Okay. So if I understand your feedback properly, the general preference is:

1. A method for declaring a queue of a different type
2. Properties for applying queue configuration arguments

I'm less familiar with the other arguments in RabbitMQ, but the sense I've got from their documentation is that these arguments are rather new, and specific to streams. Assuming that's true, do you think it makes sense to have another callback for configuring those stream-specific options (similar to how transport configuration works)? That seems most consistent with the patterns in use today. Perhaps a RabbitMqStreamOptions object would be appropriate to pass to the SetStreamQueue() method (likely with a different name)?

From what I've read thus far, consumers can use the current queue interface with streams, but will get less throughput than if they use the updated interfaces. That's more an implementation detail in providing the MassTransit interface in that scenario though (I think).

I suspect that users will want to have the ability to customize the error handling endpoints when they customize the receive endpoint, i.e., if the consumer uses a stream, they'll want the error handler to use one as well (by default, with ability to override that behavior).

I know adding support in MassTransit will be contingent on upstream support in the RabbitMQ client. Once that becomes available, what's the correct process for making a contribution? Is it a Github issue? As noted previously, my team and I would really like support for this, and I'm certainly willing to help get it implemented (and ensure stability through use). I'd also like to help contribute docs to the project, if there's any guidance you think the project should give to new consumers of the project.

One thing I'm not completely clear on is the relationship between exchanges and streams. The primary benefit of streams (as I see it) is that you can minimize the fanout overhead, since streams allow each client to have their own cursor over the stream. I don't think there's a major change in the sense of how exchange binding works, but there's likely some practical guidance that users would need. The whole point of streams is that you can declare a stream, bind it to an exchange (undoubtedly in fanout mode), and have multiple clients point to the same stream. One could alternatively just publish directly to the stream queue, which will essentially serve as the exchange for all subscribers, but I think that goes against general recommended guidance... The general guidance is to write to an exchange, and allow it to route as it sees fit. It also creates opportunity for some interesting (and potentially challenging) upgrade paths.

Thanks for the engagement on this one. It's really appreciated! I'd really love to help shape this into a great user experience. 😄

[Hawxy]

Would also love to see this feature as replay would come in handy. I noticed the .NET client has reached RC stage as well.

[paulps]

We would also love to see this feature as lossless message retry is very much needed for us.

[briandrennan]

So I've been using MassTransit and RabbitMQ a bit in production - not probably to as serious an extent as others - but I would think MassTransit dumping failed messages to an .+_error$ queue would be sufficient to avoid loss of messages. The obvious challenge there is that you can have antecedent messages that fail while future messages that depend on some previous message "succeed" but as I recall, MassTransit has a convention for consumers to stop consuming messages if a health check starts failing. Would you mind replying about challenges in this space, how you're working around the limitations (if any), and what you think an ideal solution would look like?

[aeb-dev]

@briandrennan

Is there any progress on this issue? If you do not have time can you share your progress so that other people can continue?

[briandrennan]

@aeb-dev I'm sorry, but no, I haven't made any progress on it. I thought I had time for this a while back, but decided that waiting for the official RabbitMQ-Stream SDK to become available was probably worthwhile. That reached a proper v1.0 at some point (I can't recall when), and so building an integration with it probably makes sense at this point.

My thought about implementing this would be to do so as a plugin that would eventually get adopted in the MassTransit project directly, but I'm not sure if that's the preferred approach for the project. There's very likely a short path to getting this done that is just exposing a new SetStreamQueue() argument to declare that you want a stream, but I think a more useful integration would take advantage of the capabilities of the streaming protocol and official SDK. As a user of MassTransit, I'd prefer to see that as a separate package, as the main package is very stable. My personal preference to avoid changes in the main project also drives my thinking that this should probably start as a plugin, but I'm just an end-user of the project and not in a position to make such decisions.

[moezRebai]

Any news regarding this subject ???

[francisfdo-ggl]

its a nice feature to have. keen to see it happaning soon

[Gsantomaggio]

Hello all,
I am the main contributor of https://github.com/rabbitmq/rabbitmq-stream-dotnet-client. let me know If you have questions or how I can help with the integration.

[dominikjeske]

Any plans for this?

[phatboyg]

It isn't important to me and no customers are asking for it. Only the random post on here every few months. If a customer were to request it, it would be prioritized, but that isn't/hasn't been the case.

[davorko]

Ok, lets push the posts from every few months, to weeks, days. We need RabbitMQ Streams support.

[phatboyg]

If you really want streams support, you can submit a PR. Otherwise, customer priorities come first, and I currently have no paying customers who want streams support for RabbitMQ.

[michaelklishin]

@davorko I'm a RabbitMQ core team member and I completely agree with what @phatboyg says above. Open source means you have access to the source code and an opportunity to contribute, not that you have a moral right to "push" maintainers to do what you want.

[dominikjeske]

I agree that we can push only when we pay - sometimes we are forgetting that OpenSource is not growing from air. For now I really hope for some customer willing to pay for this ;)

[Gsantomaggio]

It is worth adding this link: Stream Core vs Stream Plugin, which explains the difference between Core and Plugin.

[shaunhavelaarx]

It is worth adding this link: Stream Core vs Stream Plugin, which explains the difference between Core and Plugin.

Do you have any plans to become a contributor to the project? As @phatboyg mentioned above, he does not have any paying customers who want this atm so it seems like this would be a community-driven feature