NestJs EventBus duplicates event at EventHandler

I am trying out Event-Sourcing and CQRS using NestJs using Kafka as the Event Store.

The application is a small and simple one having 2 parts, customer and order. You first create a customer with some initial balance and then using the customer id you create order, if the amount of order is less than balance then it is approved otherwise rejected.

Here is the code in question: https://github.com/Ashniu123/nestjs-customer-order-eventsourcing-cqrs

I use KafkaJs as the EventBus (created my own KafkaModule under libs/)

When I run this with Kafka and MongoDB, the app starts up just fine. And when I create a customer too, the event CreateCustomerEvent is published as expected and pushed onto Kafka by the CommandHandler. (Checked using the landoop UI)

The problem comes in when the Event is read from kafka and pushed onto the EventBus to be picked up and executed by the EventHandler. Like CreateCustomerEventHandler.

My configuration for EventBus to use Kafka is in AppModule for each service. For instance, Customer.

And the EventBus observable subject$ is configured for the events in KafkaService.

Here are the application logs (added // for my comments).

customer-svc (Command side)

[Nest] 657306   - 09/13/2020, 12:54:47 AM   [CreateCustomerCommandHandler] Running command handler with: {"customerId":"900ee3e9-33aa-431c-bbd0-eea91cafb673","createCustomerDto":{"email":"someemail@gmail.com","password":"abc123","firstName":"john","lastName":"doe","balance":1000}}
[Nest] 657306   - 09/13/2020, 12:54:47 AM   [KafkaService] Published event: {"customerId":"900ee3e9-33aa-431c-bbd0-eea91cafb673","createCustomerDto":{"email":"someemail@gmail.com","password":"abc123","firstName":"john","lastName":"doe","balance":1000},"eventType":"CreateCustomerEvent"}

customer-view-svc (Query/View side)

[Nest] 657550   - 09/13/2020, 12:54:47 AM   [KafkaService] Bridged event payload value: {"customerId":"900ee3e9-33aa-431c-bbd0-eea91cafb673","createCustomerDto":{"email":"someemail@gmail.com","password":"abc123","firstName":"john","lastName":"doe","balance":1000},"eventType":"CreateCustomerEvent"}
[Nest] 657550   - 09/13/2020, 12:54:47 AM   [CreateCustomerEventHandler] Running event handler with: {"customerId":"900ee3e9-33aa-431c-bbd0-eea91cafb673","createCustomerDto":{"email":"someemail@gmail.com","password":"abc123","firstName":"john","lastName":"doe","balance":1000}}
[Nest] 657550   - 09/13/2020, 12:54:47 AM   [CreateCustomerEventHandler] Running event handler with: {"customerId":"900ee3e9-33aa-431c-bbd0-eea91cafb673","createCustomerDto":{"email":"someemail@gmail.com","password":"abc123","firstName":"john","lastName":"doe","balance":1000}}
[Nest] 657550   - 09/13/2020, 12:54:47 AM   [CreateCustomerEventHandler] Created customer: {"_id":"5f5d207f57cd5f089895867a","id":"900ee3e9-33aa-431c-bbd0-eea91cafb673","email":"someemail@gmail.com","password":"$2b$10$NzEnAHRsfh/7QnczB3p/MepPl0fD44G/6sFtzKsjpwudjYlNjGacG","firstName":"john","lastName":"doe","balance":1000,"salt":"$2b$10$NzEnAHRsfh/7QnczB3p/Me"}
[Nest] 657550   - 09/13/2020, 12:54:48 AM   [CreateCustomerEventHandler] Created customer: {"_id":"5f5d207f57cd5f089895867b","id":"900ee3e9-33aa-431c-bbd0-eea91cafb673","email":"someemail@gmail.com","password":"$2b$10$w0.mShhI3cMys7XAPLHRFusy63Fqlzj9s95JuSGdDpy.g5n5nt/8O","firstName":"john","lastName":"doe","balance":1000,"salt":"$2b$10$w0.mShhI3cMys7XAPLHRFu"}
// for some reason the another customer of same email is created even though in `customer.schema.ts` I have specified that it should be unique (not a priority at the moment)

What I could infer from the logs is that the Kafka event is received only once by the consumer as expected but is moved using subject$.next to the EventHandler twice.

Also, to clarify, the Events are pushed to the EventHandler two separate times, as suggested by the different values of the customer._id when created.

Using the debugger I could see that the subject.observers has 2 values in the array of class FilterSubscriber. I don't know if this is useful or no but just wanted to layout my efforts to cure this issue myself, and after 6 hours of nothing I have come here for help :).

I have added the launch.json to use with VSCode in the repo if you guys have a better use of it. Just attach using the processId of the running application.

P.S. I have configured the EventBus of both customer-view-svc and order-view-svc in a similar manner and the issue exists in both (i.e. duplicate events). I hope that you guys will be able to help me resolve the issue.

Thanks.

from NestJs EventBus duplicates event at EventHandler