The art of simplicity

A few days ago I blogged about a situation we had where some messages send to RabbitMQ got lost. I showed a possible solution when using MassTransit. We further investigated the issue and a colleague(thanks Stijn!) suggested another solution by using a specific RabbitMQ feature: Alternate Exchanges . The documentation explains the feature like this: It is sometimes desirable to let clients handle messages that an exchange was unable to route (i.e. either because there were no bound queues or no matching bindings). Typical examples of this are detecting when clients accidentally or maliciously publish messages that cannot be routed "or else" routing semantics where some messages are handled specially and the rest by a generic handler Alternate Exchange ("AE") is a feature that addresses these use cases. Whenever an exchange with a configured AE cannot route a message to any queue, it publishes the message to the specified AE instead. If that AE doe...

At one of my clients we had a situation where messages got lost after sending them to RabbitMQ. This is quite bad as the whole point of having a message based solution was to improve the reliability of our solutions(even when of the involved systems is offline or unavailable). In this post I want to explain what got wrong and how we introduced a solution to prevent this from happening in the future. To understand the problem I first have to explain the concept of an exchange. In RabbitMQ, exchanges are message routing agents that are responsible for routing messages to different queues with the help of header attributes, bindings, and routing keys. A producer never sends a message directly to a queue. Instead, it uses an exchange as a routing mediator. Therefore, the exchange decides if the message goes to one queue, to multiple queues, or is simply discarded. Let me emphasize one sentence here: In RabbitMQ , a producer never sends a message directly to a queue Only que...

Mirrored queues have been a feature in RabbitMQ for quite some time. When using mirrored queues messages are replicated across multiple nodes providing high availability in a RabbitMQ cluster. Each mirrored queue has a master and one or more mirrors, and messages are replicated from the master to its mirrors. Mirrored Queues operate on synchronous replication, meaning that the master node waits for at least one mirror to acknowledge the receipt of a message before considering it successfully delivered. This impacts performance and can result in throughput issues due to the synchronous nature of replication. Certain failure scenarios can result in mirrored queues confirming messages too early, potentially resulting in a data loss. Quorum queues Quorum Queues are a more recent addition to RabbitMQ, introduced to address some of the limitations posed by Mirrored Queues. They use a different replication model based on the Raft consensus algorithm. In this model, each queue is repli...

At one of my clients we are using MetricBeats to monitor our RabbitMQ cluster . Of course this can result in a lot of data over time. Recently I was called by one of the system administrators asking why the disk was filling up on the servers hosting ElasticSearch. Let’s find out together… Index lifecycle policies To keep the amount of data on your ElasticSearch cluster under control, you can configure a lifecycle policy . A lifecycle policy moves data through multiple phases; A hot phase: used for your most recent most frequently searched data. It provides the highest indexing AND search performance but comes with the highest cost A warm phase: optimized for search performance over index performance. In this phase it is expected that the data doesn’t change that often anymore. A cold phase: optimized for cost saving over search performance. In this phase the data is read-only. A delete phase: deletes the data you longer need Let’s see how to configure a lifecy...

Today I want to talk about a specific feature in RabbitMQ and how we can use this feature with my favorite messaging library MassTransit. Virtual Hosts With virtual hosts you can create a logical grouping in  your RabbitMQ cluster. Every exchange, queue, binding, user permissions, and so on is bound to a specific virtual host and isolated from the other virtual hosts. This helps to avoid naming collisions. There is always at least one virtual host in use; the default one is ‘/’. To create a new virtual host in your RabbitMQ cluster, you can use the rabbitmqctl 's add_vhost command: rabbitmqctl add_vhost development Or you can use the RabbitMQ Management portal: In the Management Portal go to the Admin tab and click on the Virtual Hosts section on the right: Go to the Add Virtual Host form and enter a name and an optional description and tags : Click on Add virtual host to create the new virtual host. Remark: Be aware...

I recently upgraded the (development) RabbitMQ cluster of one of my clients to RabbitMQ 3.9. The upgrade went smoothly and none of the development teams mentioned any issues. So I was happily preparing for the production upgrade. A few weeks later I was contacted by one of the team leads who was investigating a specific issue he had in one of his applications; he was using a message published to RabbitMQ to trigger a long running task (a batch job). This message was picked by a Windows Service that uses a MassTransit consumer to execute this long running task. The strange this was that the task sometimes failed. The normal behavior in MassTransit is that this message would end up in the error queue (maybe after a few retries). However this didn’t happen and the message was put back on the queue. What was going on? I started by having a look at the error logs and notice a message like this: "Message ACK failed: 258", "The channel was closed: AMQP close...

Yesterday I showed how to remove unacked messages from a RabbitMQ queue through the Management Portal. Today let us leave the Graphical User Interface behind and solve the same problem from the command line. Through the command line RabbitMQ has multiple command line tools available in the sbin folder. The one we need is rabbitmqctl.bat Show queues with unacked connections like this. rabbitmqctl list_queues name messages_unacknowledged The output should be something like this: VLM.eShopExample.Worker-Development 1 VLM.eShopExample.Worker-Production 0 We see that one queue has an unacked message. Let's find out the channel and associated connection that is causing the unacked message. rabbitmqctl list_channels connection messages_unacknowledged This returns the following output: <rabbit@SERVER.1650192371.27249.9> 1 Ok, we found the channel tha...

I got a situation where a message remained unacknowledged in a RabbitMQ queue. I wanted the related consumer to stop and remove the message from the queue. Let me walk you through the steps to get this done. Through the Management Portal In this post I’ll show you how to this through the Management Portal, in a later post I’ll show you how to do this from the commandline. Open the RabbitMQ Management Portal. Go to the Queues tab. Notice that in one of my queues I have an ‘Unacked’ message. First we need to find the connection that is related to the application consuming the message. Therefore go to the Connections tab. Click on the correct connection. Expand the Close this connection section and hit the Force Close button to close the connection and related channels. Click OK when asked for confirmation. Now go back to the Queues tab. Click on the queue with the ‘Unacked’ message. Expand the Purge section and hit the Purge Messages button. ...

Last week I introduced RabbitMQ streams and how you could produce and consume streams through the RabbitMQ.Stream.Client in .NET. Yesterday I showed how you can improve and simplify producing messages by using a Reliable Producer. Today I want to introduce its counterpart on the consumer side; the Reliable Consumer. Introducing Reliable Consumers Reliable Consumers builts on top of Consumer and adds the following features: Auto-Reconnect in case of disconnection Auto restart consuming from the last offset Handle the metadata Update Auto-Reconnect The Reliable Consumer will try to restore the TCP connection when the consumer is disconnected for some reason. Auto restart consuming from the last offset The Reliable Consumer will restart consuming from the last offset stored. So you don’t have to store and query the last offset yourself. Handle the metadata update If the streams  topology changes (ex:Stream deleted or add/remove follower), the client re...

Last week I introduced RabbitMQ streams and how you could produce and consume streams through the RabbitMQ.Stream.Client in .NET. The default Producer is really low-level and leaves a lot of things to be implemented by us. For example, we have to increment the PublishingId ourselves with every Send() operation. Let’s find out how we can improve this through Reliable Producers. Introducing Reliable Producers Reliable Producer builts on top of the Producer and adds the following features: Provide publishingID automatically Auto-Reconnect in case of disconnection Trace sent and received messages Invalidate messages Handle the metadata Update Provide publishingID automatically When using a Reliable Producer it retrieves the last publishingID given the producer name.  This means that it becomes important to choose a good reference value. Auto-Reconnect The Reliable Producer  will try to restore the TCP connection when the Producer is disc...

Yesterday I talked about RabbitMQ Streams, a new persistent and replicated data structure in RabbitMQ 3.9 which models an append-only log with non-destructive consumer semantics. I demonstrated how you could build a small example application in C# to test this stream. I first tried this code against a local cluster I had running in a docker container (check my repo if you want a Dockerfile where this plugin is already enabled: https://github.com/wullemsb/docker-rabbitmq ). At first this failed with the following error message: System.Net.Sockets.SocketException: No such host is known In my configuration you can see that I’m pointing to the local Loopback address which should be localhost: Let’s open the debugger and see what is going on… When I looked at the connection settings I noticed the following: You can see that the advertised host is not ‘localhost’ but a random string. This is the random name assigned to the node in my cluster. To get rid of this pr...

RabbitMQ has been the message broker of my choice for a long time. It has served me well over the years and I still like to use it today. Recently, I was able to add an extra reason to the list why I like RabbitMQ when I noticed that a new feature was added in RabbitMQ 3.9; Streams . RabbitMQ Streams From the documentation : Streams are a new persistent and replicated data structure in RabbitMQ 3.9 which models an append-only log with non-destructive consumer semantics. With streams you get Kafka like functionality in RabbitMQ without all the complexity that comes with maintaining and managing your Kafka cluster. It has been created with the following use cases in mind: Large amount of subscribers; in traditional queuing we use a dedicated queue for each consumers. This becomes ineffective whehn we have large number of consumers. Time-travelling; Streams will allow consumers to attach at any point in the log and read from there. Performance: Streams have been ...

One of the reasons we are using ElasticSearch is to bring all the metrics from multiple systems together in one place. One of the systems we want to monitor is our RabbitMQ cluster. Let's walk through the steps to get our RabbitMQ metrics send to ElasticSearch and available in Kibana. The RabbitMQ Metrics Integration Open up Kibana and click on the hamburger icon in the left corner. This will open up a side menu with a big ‘Add integrations’ button at the bottom. Hit this button to go to the Integrations overview page. Here you can see the list of all available integrations. Let’s search for “rabbitmq”. We get 2 results back. Click on the ‘RabbitMQ Metrics’ integration to go to the installation instructions. Install MetricBeat on Windows The RabbitMQ metrics integration is part of MetricBeat . MetricBeat is a lightweight shipper that you can install on your servers to periodically collect metrics from the operating system and from services running on those ser...

Yesterday I introduced the AsyncApi specification as a way to share message contracts in a language neutral way. Let’s continue today by looking how we can generate a C# messagecontract based on your AsyncAPI specification file. Here is the example specification file again that I shared yesterday: To transform this specification file to a data contract, we can use the Async API generator . The generator uses templates to specify what must be generated. There is a list of official generator templates, unfortunately C# is not part of this list. Thanks to the community a csharp compatible template exists; https://github.com/jonaslagoni/asyncapi-quicktype-template Let’s try that one! Install the generator First install the generator through NPM: npm install -g @asyncapi/generator You can optionally pre-install the quicktype template: npm install -g @lagoni/asyncapi-quicktype-template Generate C# Message contracts Now we can generate the C# messag...

As most of the systems I’m building are .NET based, I typically use NuGet Packages (published on an internal NuGet repository like Azure Artifacts or MyGet ) to share my message contracts between different parts of the system. The main disadvantage of this approach is that it creates platform coupling and is not a good solution if you are in a polyglot environment using different platforms and programming languages. What if we could write our message contracts in a language neutral way? That is exactly what AsyncAPI has to offer. From the AsyncAPI website: AsyncAPI is an open source initiative that seeks to improve the current state of Event-Driven Architectures (EDA). Our long-term goal is to make working with EDAs as easy as it is to work with REST APIs. That goes from documentation to code generation, from discovery to event management. Most of the processes you apply to your REST APIs nowadays would be applicable to your event-driven/asynchronous APIs too. To make ...

If you are interested in message or event based architectures, bookmark the link to the following Github repo: https://github.com/clemensv/messaging This repository is created by Clemens Vaster, the product architect of the messaging and eventing services in the Microsoft Azure cloud. It refers to a lot of talks and articles that help you to get started with message and event based systems and links to the product pages of multiple cloud providers and open source products that exist in this space.

By default when a MassTransit consumer fails to handle a message (and throws an exception), the message is moved to an _error queue (prefixed by the receive endpoint queue name). This is OK for transient exceptions but probably not what you want when you have a bug in your system or there is another reason why none of the messages can be handled succesfully. In that case, another feature of MassTransit becomes handy; the kill switch . A Kill Switch is used to prevent failing consumers from moving all the messages from the input queue to the error queue. By monitoring message consumption and tracking message successes and failures, a Kill Switch stops the receive endpoint when a trip threshold has been reached. You can configure a kill switch for a specific endpoint or for all receiver endpoints on the bus. Here is a short example on how to configure the kill switch for all receiver endpoints:   In the above example, the kill switch will activate after 10 messa...

MassTransit dedicated a whole documentation page to message versioning but it still wasn’t completely clear to me how it worked. Let’s use this blog post to see what’s going on… Publishing messages Let’s first focus on the sending side. Publishing a first version of our message contract We’ll start with a first version of our message contract: Let’s send this to RabbitMQ using: Time to open the RabbitMQ Management portal and take a look how the message payload is constructed: Creating a second version of our message contract Let’s introduce a v2 version of our message contract: If we send it to RabbitMQ in the same way: There isn’t such a big difference when comparing the payloads: The ‘messagetype’ value is different and of course the message itself. But that’s it. Send a backwards compatible version Let’s now construct a message that implements both contracts: And send that one: If we now check the payload, we see that 1 message is put on th...

I couldn’t find a good example on the best way to integrate the RabbitMQ C# client in a .NET core application. So time to write a post about it. I would recommend to use the Hosted Services , either directly through the IHostedService or through the BackgroundService . Let’s see how to do it using the BackgroundService: We create a new class that inherits from BackgroundService : We’ll use the StartAsync method to create the Connectionfactory , Connection and a Channel to listen on: Remark: Notice the DispatchConsumersAsync = true in the ConnectionFactory configuration. This is important to be able to use an async consumer. If you don’t add this configuration no messages will be picked up by the AsyncEventingBasicConsumer . We also implement the StopAsync method to cleanup when our backgroundservice is shutdown: Now the only thing left is to create a consumer to start receiving messages. We’ll use the ExecuteAsync method for that: O...

By default RabbitMQ tries to keep your whole queue in memory. This is OK as long as your messages are processed fast enough but not if your queue becomes very long(many millions of messages). Queues can become very long for various reasons: consumers are offline / have crashed / are down for maintenance there is a sudden message ingress spike, producers are outpacing consumers consumers are slower than normal Lazy Queues can help in these situations- messaged are moved to disk as early as practically possible, and are only loaded in RAM when requested by consumers.This comes at a cost of increased disk I/O. You can configure this in MassTransit when configuring your receive endpoint: More information: https://www.rabbitmq.com/lazy-queues.html