So I have a batch processing which will send a command to aggregate member at the end of the month. it may look like this.
@Aggregate
class ProductAggregate {
@AggregateMember
val invoices = listOf<Invoice>()
}
class Invoice {
@CommandHandler
fun handle(cmd: ClearInvoiceCommand){
// do whatever at the end of the month
}
}
So I did a few research and found that when a command is sent to aggregate, The aggregate will be locked until the command is executed succesfuly. I wonder if it’s also locked for the aggregate member. Let’s say I have 2 invoices inside of ProductAggregate. At the end of the month, my batch processing will send the same command to the different invoice. Can these commands execute at the same time or it has to wait until one is done. I got weird result from doing this and not sure what’s wrong with it so I have to find out.
Update: After doing a lot of debugging now I found a really bad problem with my own aggregate design. So I was trying to decouple these aggregates since I’m doing microservices architecture I don’t wanna put everything in a single service. So I made this approach.
// we run a lot of products with different features,
// so I don't want to put everything in the same aggregate or the same service
// I also want resource to be shared among these service
// So this is a base factory service which we design
abstract class BaseFactory {
companion object {
const val aggregateType = "FactoryAggregate"
}
@AggregateIdentifier
lateinit var factoryId: String
var resource = 0
// this will automatically duplicate state from **FactoryCreatedEvent**
// to every factories that extend this class
// So I can have shared state with such strong consistency
@EventSourcingHandler
fun on(event: FactoryCreatedEvent){
resource = event.resource
// every sub factory use the same id
factoryId = event.factoryId
}
// any factory that use the resource need to apply this event so other factories
// resource state will also be changed
@EventSourcingHandler
fun on(event: ResourceUsedEvent){
resource -= event.amount
}
fun hasEnoughResource(required: Int): Boolean {
return resource >= required
}
}
@Aggregate(type = BaseFactory.aggregateType)
class FactoryAggregate: BaseFactory() {
@CommandHandler
fun handle(command: CreateFactoryCommand){
AggregateLifecycle.apply(event)
}
}
// let's say this service only care about plane factory deploy on the different node
@Aggregate(type = BaseFactory.aggregateType)
class PlaneFactoryAggregate: BaseFactory() {
@AggregateMember
val wings = listOf<Wing>()
fun hanlde(command: CreateWingCommand){
if (!hasEnoughResource(command.required)){
throw ...
}
}
}
// let's say this service only care about car factory deploy on the different node
@Aggregate(type = BaseFactory.aggregateType)
class CarFactoryAggregate: BaseFactory() {
@AggregateMember
val wheels = listOf<Wheel>()
fun hanlde(command: CreateWheelCommand) {
if (!hasEnoughResource(command.required)){
throw ...
}
}
}
// there can be more factories C, D, E, F, G which has different features,
// but require the same resource
At first it works just fine, but little did I know. Since there’s requirement where I have to run batch processing at the end of the month. Every factories send the same command at the same time so It’s only lock within its own aggregate. What I mean is, when the CarFactoryAggregate sent the command it might conflict with PlaneFactoryAggregate since these two aggregates only lock its own and duplicate aggregateSequenceNumber. Is there a way to solve this problem I really feel bad of myself since my design breaks the entire business. Or should I redesign my aggregate and store everything I needed in the same class.
Another update: Ok after doing a few research I found this post Saga/Aggregate: multiple threads and locking - #3 by vab2048. So I think I should give up on my approach.