Upgrade to Pro — share decks privately, control downloads, hide ads and more …

Dealing with change in event sourced applicatio...

Dealing with change in event sourced applications (ConFoo 2023)

In software development, change is pretty much the only constant factor. In fact, embracing change is one of the twelve principles behind the Agile Manifesto. However, changing event sourced applications can be very challenging.

In this talk, we’ll explore how to deal with projection updates, event updates and versioning, and existing privacy legislation (such as the GDPR).

Michiel Rook

February 24, 2023
Tweet

More Decks by Michiel Rook

Other Decks in Programming

Transcript

  1. @michieltcs #StandWithUkraine - M A RT I N FOW L

    E R "Event Sourcing ensures that all changes to application state are stored as a sequence of events."
  2. @michieltcs #StandWithUkraine ACTIVE RECORD VS. EVENT SOURCING Account Id Account

    number Balance 1234 12345678 € 50,00 ... ... ... Money Withdrawn Account Id 1234 Amount € 50,00 Money Deposited Account Id 1234 Amount € 100,00 Account Opened Account Id 1234 Account number 12345678
  3. @michieltcs #StandWithUkraine COMMANDS TO EVENTS Deposit Money Account Id 1234

    Amount € 100,00 data class DepositMoney( @TargetAggregateIdentifier val accountId: String, val amount: BigDecimal )
  4. @michieltcs #StandWithUkraine COMMANDS TO EVENTS Deposit Money Account Id 1234

    Amount € 100,00 command 
 handler @CommandHandler fun depositMoney(command: DepositMoney) { apply(MoneyDeposited( command.accountId, command.amount, GenericEventMessage.clock.instant())) }
  5. @michieltcs #StandWithUkraine COMMANDS TO EVENTS Deposit Money Account Id 1234

    Amount € 100,00 Money Deposited Account Id 1234 Amount € 100,00 command 
 handler data class MoneyDeposited( val accountId: String, val amount: BigDecimal, val instant: Instant )
  6. @michieltcs #StandWithUkraine AGGREGATES @Aggregate class BankAccount { @AggregateIdentifier private lateinit

    var accountId: String private lateinit var accountNumber: String private lateinit var balance: BigDecimal @CommandHandler fun depositMoney(command: DepositMoney) { apply(MoneyDeposited( command.accountId, command.amount, GenericEventMessage.clock.instant())) } @EventHandler fun moneyDeposited(event: MoneyDeposited) { balance = balance.add(event.amount) } }
  7. @michieltcs #StandWithUkraine AGGREGATE STATE Account number Balance 12345678 € 0,00

    Account number Balance 12345678 € 100,00 Account number Balance 12345678 € 50,00 event 
 handler event 
 handler event 
 handler Money Withdrawn Account Id 1234 Amount € 50,00 Money Deposited Account Id 1234 Amount € 100,00 Account Opened Account Id 1234 Account number 12345678
  8. @michieltcs #StandWithUkraine VALIDATING COMMANDS @CommandHandler @Throws(OverdraftDetectedException::class) fun withdrawMoney(command: WithdrawMoney) {

    if (balance.compareTo(command.amount) >= 0) { apply(MoneyWithdrawn( command.accountId, command.amount, GenericEventMessage.clock.instant())) } else { throw OverdraftDetectedException(accountNumber, balance, command.amount) } }
  9. @michieltcs #StandWithUkraine TESTING AGGREGATES @TestInstance(TestInstance.Lifecycle.PER_CLASS) class BankAccountTest { private lateinit

    var fixture: AggregateTestFixture<BankAccount> @BeforeEach fun createFixture() { fixture = AggregateTestFixture(BankAccount::class.java) } @Test fun noOverdraftsOnEmptyAccount() { fixture.given(accountOpened()) .`when`(WithdrawMoney(ACCOUNT_ID, WITHDRAW_AMOUNT)) .expectException(OverdraftDetectedException::class.java) } const val ACCOUNT_ID = "accountId" const val ACCOUNT_NUMBER = "3856625" }
  10. @michieltcs #StandWithUkraine -AG I L E M A N I

    F ES TO "Welcome changing requirements, even late in development."
  11. @michieltcs #StandWithUkraine PROJECTION ACCOUNT OPENED EVENT HANDLER # OF ACTIVE

    ACCOUNTS +1 ACCOUNT CLOSED EVENT HANDLER # OF ACTIVE ACCOUNTS -1
  12. @michieltcs #StandWithUkraine DOMAIN UI EVENT BUS EVENT HANDLERS COMMAND HANDLERS

    REPOSITORY DATABASE DATABASE EVENT STORE AGGREGATES commands events events
  13. @michieltcs #StandWithUkraine DOMAIN UI EVENT BUS EVENT HANDLERS COMMAND HANDLERS

    REPOSITORY DATA LAYER DATABASE DATABASE EVENT STORE commands events events queries DTOs AGGREGATES
  14. @michieltcs #StandWithUkraine PROJECTION class BankAccountProjections { private val activeAccounts: MutableMap<String,

    BankAccount> = HashMap() @EventHandler fun onAccountOpened(accountOpened: AccountOpened) { val bankAccount = BankAccount( accountOpened.accountId, accountOpened.accountNumber, BigDecimal.ZERO) activeAccounts[accountOpened.accountId] = bankAccount } @EventHandler fun onAccountClosed(accountClosed: AccountClosed) { activeAccounts.remove(accountClosed.accountId) }
  15. @michieltcs #StandWithUkraine PROJECTION @GetMapping("accounts/{accountId}") fun getAccountNumber(@PathVariable("accountId") accountId: String?): Optional<String> {

    return bankAccountProjections.findAccountById(accountId!!) .map { obj: BankAccount -> obj.accountNumber } }
  16. @michieltcs #StandWithUkraine ZERO DOWNTIME NEW EVENTS QUEUE LOOP OVER EXISTING

    EVENTS APPLY TO NEW PROJECTION APPLY QUEUED EVENTS USE PROJECTION
  17. @michieltcs #StandWithUkraine DIVIDING THE WORK EVENT EVENT EVENT EVENT EVENT

    AGGREGATE INSTANCE EVENT EVENT EVENT EVENT INSTANCE
  18. @michieltcs #StandWithUkraine DIVIDING THE WORK EVENT EVENT EVENT EVENT EVENT

    AGGREGATE INSTANCE EVENT EVENT EVENT INSTANCE EVENT EVENT EVENT EVENT
  19. @michieltcs #StandWithUkraine DIVIDING THE WORK SELLER EVENT SELLER EVENT SELLER

    EVENT SELLER EVENT SELLER EVENT LISTING EVENT LISTING EVENT LISTING EVENT LISTING EVENT LISTING EVENT SELLER LISTING Seller Name Listing Date Listing Description ... ... ... ... ... ... ... ... ... ... ... ... ... ... ... ... ... ... ... ... ...
  20. @michieltcs #StandWithUkraine DIVIDING THE WORK SELLER EVENT SELLER EVENT SELLER

    EVENT SELLER EVENT SELLER EVENT LISTING EVENT LISTING EVENT LISTING EVENT LISTING EVENT LISTING EVENT SELLER LISTING INSTANCE INSTANCE
  21. @michieltcs #StandWithUkraine DIVIDING THE WORK SELLER EVENT SELLER EVENT SELLER

    EVENT SELLER EVENT SELLER EVENT LISTING EVENT LISTING EVENT LISTING EVENT LISTING EVENT LISTING EVENT SELLER LISTING INSTANCE INSTANCE ?
  22. @michieltcs #StandWithUkraine TRACKING EVENTS EVENT EVENT EVENT EVENT EVENT EVENT

    . . . . . . . . . . . . . . EVENT EVENT EVENT TOKEN STORE
  23. @michieltcs #StandWithUkraine TRACKING EVENT PROCESSOR @Autowired fun startTrackingProjections(configurer: EventProcessingConfigurer) {

    val scanner = ClassPathScanningCandidateComponentProvider(false) scanner.addIncludeFilter(AnnotationTypeFilter(RebuildableProjection::class.java)) for (bd in scanner.findCandidateComponents("org.demo")) { val projectionClass = Class.forName(bd.beanClassName) val rebuildableProjection = projectionClass.getAnnotation(RebuildableProjection::class.java) if (rebuildableProjection.rebuild) { registerRebuildableProjection( configurer, projectionClass, rebuildableProjection) } } }
  24. @michieltcs #StandWithUkraine TRACKING EVENT PROCESSOR private fun registerRebuildableProjection(configurer: EventProcessingConfigurer, projectionClass:

    Class<*>, rebuildableProjection: RebuildableProjection) { val processingGroup = projectionClass.getAnnotation(ProcessingGroup::class.java) val name = processingGroup?.let(ProcessingGroup::value) ?: (projectionClass.name + "/" + rebuildableProjection.version) configurer.assignHandlerTypesMatching(name) { eventHandler -> projectionClass.isAssignableFrom(ClassUtils.getUserClass(eventHandler)) } configurer.registerTrackingEventProcessor(name) }
  25. @michieltcs #StandWithUkraine TRACKING EVENT PROCESSOR class BankAccountProjections { private val

    activeAccounts: MutableMap<String, BankAccount> = HashMap() @ResetHandler fun resetProjections() { activeAccounts.clear() } ////
 }
  26. @michieltcs #StandWithUkraine @michieltcs Ledger Entry Aug 14 Inventory € 15600,00

    Accounts Payable € 15600,00 Ledger Entry Aug 14 Inventory € 16500,00 Accounts Payable € 16500,00
  27. @michieltcs #StandWithUkraine Ledger Entry Aug 14 Inventory € 15600,00 Accounts

    Payable € 15600,00 Ledger Entry Aug 14 Inventory € 16500,00 Accounts Payable € 16500,00 Ledger Correction Entry Aug 14 Inventory € 900,00 Accounts Payable € 900,00
  28. @michieltcs #StandWithUkraine COMPENSATING ACTIONS class MoneyWithdrawn {
 String accountId;
 BigDecimal

    amount;
 } class WithdrawalRolledBack {
 String accountId;
 BigDecimal amount;
 } Typo: too much withdrawn!
  29. @michieltcs #StandWithUkraine COMPENSATING ACTIONS class AccountOpened {
 String accountId;
 String

    accountNumber;
 } class DuplicateAccountClosed {
 String accountId;
 } Duplicate account number!
  30. @michieltcs #StandWithUkraine UPCASTING class AccountOpenedUpcaster : SingleEventUpcaster() { private val

    typeConsumed = SimpleSerializedType("AccountOpened", "1.0") private val typeProduced = SimpleSerializedType("AccountOpened", "2.0") override fun canUpcast(intermediateRepresentation: IntermediateEventRepresentation): Boolean { return intermediateRepresentation.type == typeConsumed } override fun doUpcast(intermediateRepresentation: IntermediateEventRepresentation): IntermediateEventRepresentation { return intermediateRepresentation.upcastPayload( typeProduced, JsonNode::class.java ) { event: JsonNode -> val node = event as ObjectNode val accountNumber = node["accountNumber"].asText() node.put("accountNumberIban", toIban(accountNumber)) node.remove("accountNumber") event } }
  31. @michieltcs #StandWithUkraine events_v1 [
 {
 "id": "12345678",
 "type": "AccountOpened",
 "aggregateType":

    "Account",
 "aggregateIdentifier": "1234",
 "sequenceNumber": 0,
 "payloadRevision": "1.0",
 "payload": { ... },
 "timestamp": ...
 ...
 },
 ...
 ]
  32. @michieltcs #StandWithUkraine VERSIONED EVENT STORE LOOP OVER EXISTING EVENTS APPLY

    UPCASTER ADD QUEUED EVENTS USE NEW EVENT STORE NEW EVENTS QUEUE
  33. @michieltcs #StandWithUkraine events_v2 [
 {
 "id": "12345678",
 "type": "AccountOpened",
 "aggregateType":

    "Account",
 "aggregateIdentifier": "1234",
 "sequenceNumber": 0,
 "payloadRevision": "2.0",
 "payload": { ... },
 "timestamp": ...
 ...
 },
 ...
 ]
  34. @michieltcs #StandWithUkraine - G D P R , A RT

    I C L E 17 "... shall have the right to obtain ... the erasure of personal data concerning him or her without undue delay"
  35. @michieltcs #StandWithUkraine STORE PII EXTERNALLY ACCOUNTOPENED EXTERNAL STORAGE Account Id

    Account number Name 1234 12345678 John Doe ... ... ... data class AccountOpened( @TargetAggregateIdentifier
 val accountId: String )
  36. @michieltcs #StandWithUkraine ENCRYPTING EVENTS <org.demo.AccountOpened> 
 <accountId>80f49161</accountId> 
 <accountNumberIban>2dqjHkY8Mc8+cek4vs/9hzgkob4J3fZJNIJh2sAXlJ0=</ accountNumberIban>

    
 <firstName>N5Y27vd0UbKo6FIu5c7QGQ==</firstName> 
 <lastName>OSKrzfuuuayuUNXYS5YUug==</lastName> ... 
 </org.demo.AccountOpened>
  37. @michieltcs #StandWithUkraine SHEDDING THE KEY LOAD 
 EVENT FIND ASSOCIATEDENC

    RYPTION KEY DECRYPT PAYLOAD VALUES PROCESS 
 EVENT X