RabbitMQ prefetch and message consumer order

[cdunnkiplot]

Hi,

I am trying to use MT and RabbitMQ to implement Partial Ordering.

I have a RabbitMQ consistent hash exchange that distributes messages to several queues. Each queue has a single exclusive consumer that processes the messages. Each exclusive consumer then uses a MassTransit Partitioner to further increase throughput within the process. The hash exchange and MT Partitioner both use the same Partition Key to ensure messages for the same Key end up in the same queue and MT partition.

I almost had a good working solution until I noticed the prefetch count was using the default, which in my case resolved to 40, and according to your answer here, anything but a prefetch count of 1 means that message consume order is determined by the Task Scheduler, which is nondeterministic.

I can set the prefetch count to 1 but this massively reduces the throughput of my app. I guess I could create more RabbitMQ queues and deploy more instances of my app but this will complicate my deployment topology and consume more resources in my deployment environment, both of which aren't ideal for me.

I ran some tests using the default prefetch count and in all cases I found that the first prefetch batch would execute messages out of order, but thereafter it seemed to settle and consistently execute in order. I take this with a pinch of salt however as a production environment and resource contention could change this behaviour and so I can't rely on this.

Is there a way to make the message consumer order deterministic when a batch of messages are buffered into memory when pushed from RabbitMQ? if not currently, could it be possible? and if so, could you please point me to the related code?

I feel so close to something really great that could be a real contender to Kafka, but with all the features and benefits of MassTransit. If this can't be achieved then I might have to admit defeat and look into Kafka instead, but that comes with its own different set of problems and complexity.

Any help and guidance would be much appreciated!

Thanks

[phatboyg]

You can use MassTransit with Kafka and it does in-order delivery per PartitionKey.

If you're using Batch<T> consumers in MassTransit, you can set a GroupBy on the batch options and it will deliver them in order within the batch assuming your prefetch is similarly configured.

[cdunnkiplot]

Wow thanks for the quick reply.

Due to this issue I have started to look at Kafka and I've been looking into MassTransit as the client, but I'm a little reluctant because of how great MT + RabbitMQ are together with the simplicity of how the RabbitMQ topology is automatically created and errors can be handled, retried and deferred etc. The MT + Kafka error handling seems to be limited in comparison and from what I can tell, anymore more than in-line retries seem to require quite a complex solution. Of course please correct me if I'm wrong :) other dotnet messaging clients seem to be in a similar position when it comes to handling errors (at first glance anyway).

Regarding Batch Consumers, is it still the case that there isn't any support for advanced concepts like acknowledging individual messages, faulting individual messages, etc?

Given a Batch Consumer that receives several messages in a batch, is there a way for it to find and execute a consumer for each message? in my use case, each message executes a complex process that involves a lot of I/O and integrations with 3rd party API's, so having the full power of the middleware and error handling etc would be beneficial. Here is some pseudo code to illustrate what I mean:

class BatchConsumer : IConsumer<Batch<Message>>
{
    public async Task Consume(ConsumeContext<Batch<Message>> context)
    {
        var consumer = masstransit.GetConsumer<Message>();

        // I'm assuming the order in this collection is the same as it was in the queue?
        for(int i = 0; i < context.Message.Length; i++)
        {
            // a consumer within a consumer, which now reminds me of that xzibit yo dawg meme......
            await consumer.Consume(context);
        }
    }
}

One last thing, looking at the Consumer Batch Options, does the GroupBy effectively act as a PartitionKey? so if I configure the ConcurrencyLimit to 10, I get 10 partitions? where each groupby/partition will only ever execute one batch at a time? or would I need to use a separate Partitioner for that? there's obviously a lot of advanced features with MT which I'm not familiar with!

Thanks again for the support.

[phatboyg]

You'd really just have to try various options and see what works for you. I don't generally try to recreate scenarios based on these discussions. More direct support options are available.