How to submit 100s of events on saga completion?

Axon Framework
, ,
1

I have a batch import process that will load hundreds or potentially a couple thousands of objects from an external system for each user, so I made it into a saga (update: just found out about stateful handlers, will take a look). I’m storing each pending object in a db as {id, importId, value} where value is the pre-constructed event as json, and importId refers to the saga. If the process fails, it has to delete all the objects and restart. On success, for each object I will publish its event and then also cleanup the table.

@EndSaga
@SagaEventHandler(associationProperty = "importId")
fun onImportSucceeded(event: ImportSucceeded) {
  getPendingObjectsPages(this.importId).forEach { page ->
    val nestedUow = BatchingUnitOfWork(page.map { GenericEventMessage.asEventMessage<Any>(it.value) }.toList())
    nestedUow.afterCommit { pendingEventRepo.deleteByIdIn(page.map { it.id!! }) }
    nestedUow.execute { eventBus.publish(nestedUow.message) }
  }
}

I have some concerns about this approach, could someone please help with suggestions?

  1. My assumption here is that by creating a nested unit of work, I will be able to commit each page of events separately without having to load up the whole batch into memory. But I know that event handler already has its own unit of work, so I’m not sure that actually helps.
  2. Is it alright that in a single event handler I publish so many events? Or perhaps it’s better to start a chain of events for each page like importSucceededEvent > importPublishCommand > importChunkPublishedEvent /* the 2 actions above for a single page */ > if (hasNext()) importPublishCommand else saga.end() > ...
  3. I want a single transaction to span both the page-scoped unit of work and the page deletion from the pending objects table. I want both of these to happen atomically. Not quite sure how to do it. I have spring boot and spring data jpa. Can I put those 2 actions into a @Transactional method and call it from here (I know how proxy works, got CTW)? Not sure if unit of work respects that.
  4. This is axon v4 but I barely started adding axon, I don’t even have commands declared yet, so I will be migrating to v5. I just read here that UoW can no longer be nested. So, now I’m not sure how this translates.

Tried asking your AI but this seems too nuanced, it started making stuff up. Otherwise it’s actually been very useful