Handling Duplicate Messages (Idempotent Consumers)

Do you create idempotent consumers? "At Least Once" message guarantees that a message will be delivered to a consumer once or many times. This means that you need to develop your consumers to be able to effectively handle duplicate messages. Not doing so could result in some bad outcomes of your system. For example, processing an order twice, would not likely be a good outcome. Why do message brokers support "At Least Once" messaging, and how do you handle duplicates? Here's how to make idempotent consumers and be resilient to duplicate messages.
👍 THANK YOU for supporting my channel!
Venky Venkataraman
Saillesh Pawar
🔔 Subscribe: / @codeopinion
💥 Join this channel to get access to source code & demos!
/ @codeopinion
🔥 Don't have the JOIN button? Support me on Patreon!
/ codeopinion
📝 Blog: codeopinion.com/handling-dupl...
👋 Twitter: / codeopinion
✨ LinkedIn: / dcomartin
0:00 Intro
0:33 Delivery Guarantees
1:19 At Least Once
3:46 Keeping Track of Processed Messages
9:09 Demo Application
10:14 Naturally Idempotent
11:13 Sagas

[@CodeOpinion]

How are you handling duplicate messages? Does your messaging library provide support? Let me know!

[@crhayes]

Product Manager: "Will my user get the message?" Staff Engineer: "Yes, at most once."

[@andersjuul8310]

Hi Derek, just saw a recently posted video on another channel about handling of idempotency. Some cases wouldn't be handled in that version and I rewatched your video, where, sure enough, you demonstrate a more solid version. Thanks for putting your videos out -- inspiring and most appreciated!

[@CodeOpinion]

thanks. I should revisit this topic again at some point.

[@hayala0]

This is epic. Thanks so much for the insight.

[@jameswachira5713]

Excellent

[@mohammadbarbast6524]

excellent and perfect tutorial, thanks so much

[@CodeOpinion]

You're very welcome!

[@Galakyllz]

Great video. Thanks!

[@CodeOpinion]

You're welcome!

[@nmarcel]

A key point here is that you have a unique Message Id (provided in this example by that "CapHeader" component). Not having that, I think that clients could produce a Guid for each of its individual requests (not retries). Unless I'm missing something.

[@CodeOpinion]

Yes. Messages should have IDs generated by the producer. They will be unique to instance of a event/message.

[@shynggyskassen942]

Great videos!

[@CodeOpinion]

Thanks!

[@oguzhan1191]

First of all thanks for the videos they are inspiring. I have 3 question 1- You are checking the database for duplicate records but if there are 1m records in your db the anyAsync query's cost will be heavy am I right? 2- You are working with the CAP framework but CAP supports SqlServer, MySql, PostgreSql，MongoDB databases and it is limiting your db choice. What tools or architecture are you using when you are not working with these databases? I mean how do you apply outbox pattern and idempotent consumer ? 3- You may not catch the records in the database when 2 close request comes into your application(I mean lower then 1 ms let's say) and 2 separated thread query DB but both of them can't find any record and if statement will not work. If I am right how do you handle the situation?

[@CodeOpinion]

1-Indexed, no not really. You can always clear/purge older records as you wouldn't expect the same message to be delivered 1 month later. 2-Use a message library that supports these patterns as well as the data storage you're using. 3-If you're using a serializable transaction, no you'll have consistency and not a dirty read. Also, when you insert the new messageId into the idempotent table, that will have a unique constraint on it, so that's what is going to handle concurrency. The query isn't entirely needed if you're relying on the unique constraint. Hope that answers your questions!

I'm the 500'th like.

[@BondhanNovandy]

Hi, Thanks for sharing. Is this applicable in production, supposed if we keep the events/msg in database such as RDBMS and do update (delete) once consumed will it slow down the database processing?

[@CodeOpinion]

It can as you're adding more work to your DB, but if that will cause any issues are dependent on your workload.

[@Mvision123]

Does it make sense to put this transaction+ idempotence in a mediatr Middleware? Maybe with an attribute [EnsureIdempotent] on the event?

[@CodeOpinion]

What I wrote was pretty cumbersome in the sense you wouldn't want to write what I wrote in every handler/consumer. I do think that you could use some sort of pipeline/middleware for any type of message dispatch to accomplish it, yes. However, you would want to define that on the handler/consumer, not the event.

[@fr3fou]

what if your handler has to talk to some external service (e.g a payment processor)? If for example, you make the external service calls from inside the database transaction and if the request takes a while to respond (e.g service is down, exponential backoff, etc), you'd be starving your database connection pool, due to the uncommited transaction. If you also have retry & timeout logic in your message broker, what you can end up with is an event that was never processed (or partially processed if you do the call to the external service outside your transaction).

[@omerkarahan2482]

Not sure if I understood correctly. But its not a good idea to wrap the database call together with a synchronous service call. If possible and depending on the context it would be better to try to process first the messages. After this process you would make an asynchronous processing of the events which were consumed already successfully and calling your services. This way you have 2 independent steps and eventual consistency.

[@diegogomez6931]

What if I need to publish an event in the consumer? Let's say for this example an event called ShippingLabelCreated. If I understand correctly, in that case inside the the if that is checking if the message was already processed I just can't return because it is possible that the ShippingLabel was inserted in the database but for some reason the ShippingLabelCreated was not published (because the network was down, ASB/RabbitMQ was down, etc.). In that case I guess the solution would be to either save ShippingLabelId in the IdempotentConsumer table or save the MessageId in the ShippingLabel table so that inside that if instead of just returning I can retreive the ShippingLabel entity to retry to send the event.

[@CodeOpinion]

Check out this video on the outbox pattern which addresses your question. kzbin.info/www/bejne/q2nJgKGudt1-fs0

[@catrox1000]

What will be the best way to mix mediatr with Cap¿ Or should it be a replace of mediatr¿ May be you can show us more about this

[@CodeOpinion]

You could use MediatR for synchronous request/response say from ASP.NET Core to invoke commands or queries. And use CAP for out-of-process messaging. Also if you want a single unified library to do both, take a look at Brighter. codeopinion.com/moving-work-out-of-process-using-brighter-and-rabbitmq/

[@tyomidi]

@@CodeOpinion do you prefer or recommend Brighter over Masstransit? A video with your opinions on the various packages would be great.

[@Pleviwow]

Is there a best practice to handle old messages? I mean you don't really need yesterdays message and consumer ids in your database forever. Do you delete them periodically or something like this by a job?

[@CodeOpinion]

I think it totally depends on the system in terms of how long to keep them. I don't think there's a "best practice" other than to either figure out what's the length of time before you want to evict/remove old records. I do think using a period job to remove them is an option. You'd want to also persist the datetime of when you created the record so you know when to remove it obviously.

[@Pleviwow]

@@CodeOpinion Thank you!

[@06A21A0409]

Hi First of all why consumer in a consumer group read duplicate messages, as each partition is assigned to unique consumer in a consumer group?

[@CodeOpinion]

I think you're talking about Kafka specifically since yes, a single consumer in a group is assigned to a partition(s). But that's irrelevant, it's about processing the same message more than once which will occur for any broker with at-least-once semantics.

[@mikaputa2194]

What about race conditions?

[@CodeOpinion]

Concurrency is handled by the unique constraint. First one wins.