Dealing with change in event sourced applications (ConFoo 2023)

DEALING WITH CHANGE   IN EVENT SOURCED APPLICATIONS Michiel Rook
  @michieltcs

@michieltcs #StandWithUkraine YOU?

@michieltcs #StandWithUkraine FAMILIAR WITH CQRS/ES THEORY?

@michieltcs #StandWithUkraine HOBBY PROJECT?

@michieltcs #StandWithUkraine APP IN PRODUCTION?

@michieltcs #StandWithUkraine QUICK RECAP

@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."

@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

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

Amount 100,00 command   handler @CommandHandler fun depositMoney(command: DepositMoney) { apply(MoneyDeposited( command.accountId, command.amount, GenericEventMessage.clock.instant())) }

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 )

@michieltcs #StandWithUkraine AGGREGATES An Aggregate handles Commands and generates Events
based on the current state

@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) } }

@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

@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) } }

@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" }

@michieltcs #StandWithUkraine THIS IS ABOUT CHANGE

@michieltcs #StandWithUkraine -AG I L E M A N I
F ES TO "Welcome changing requirements, even late in development."

@michieltcs #StandWithUkraine MODIFICATIONS

@michieltcs #StandWithUkraine EVENT SOURCING

@michieltcs #StandWithUkraine CHALLENGE!

@michieltcs #StandWithUkraine REPLAYS AND REBUILDS

@michieltcs #StandWithUkraine ANSWERING QUERIES

@michieltcs #StandWithUkraine BASED ON EVENTS

@michieltcs #StandWithUkraine QUERIES ACCOUNT OPENED ACCOUNT OPENED ACCOUNT CLOSED Number
of active accounts?

@michieltcs #StandWithUkraine QUERIES MONEY DEPOSITED MONEY WITHDRAWN INTEREST RECEIVED Accounts
with balance > €100? MONEY DEPOSITED MONEY WITHDRAWN

@michieltcs #StandWithUkraine PROJECTION ACCOUNT OPENED EVENT HANDLER # OF ACTIVE
ACCOUNTS +1 ACCOUNT CLOSED EVENT HANDLER # OF ACTIVE ACCOUNTS -1

@michieltcs #StandWithUkraine PROJECTION EVENTS EVENT HANDLER(S) STORAGE

@michieltcs #StandWithUkraine CQRS

@michieltcs #StandWithUkraine UI

@michieltcs #StandWithUkraine DOMAIN UI COMMAND HANDLERS commands AGGREGATES

@michieltcs #StandWithUkraine DOMAIN UI COMMAND HANDLERS REPOSITORY EVENT STORE commands
events AGGREGATES

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

@michieltcs #StandWithUkraine DOMAIN UI EVENT BUS EVENT HANDLERS COMMAND HANDLERS
REPOSITORY DATA LAYER DATABASE DATABASE EVENT STORE commands events events queries DTOs AGGREGATES

@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) }

@michieltcs #StandWithUkraine PROJECTION fun findAccountById(accountId: String): Optional<BankAccount> { return Optional.ofNullable(activeAccounts[accountId])
} fun getNumberOfAccounts(): Int { return activeAccounts.size } }

@michieltcs #StandWithUkraine PROJECTION @GetMapping("accounts/{accountId}") fun getAccountNumber(@PathVariable("accountId") accountId: String?): Optional<String> {
return bankAccountProjections.findAccountById(accountId!!) .map { obj: BankAccount -> obj.accountNumber } }

@michieltcs #StandWithUkraine PROJECTION @GetMapping("accounts/active") fun getNumberOfAccounts(): Int { return bankAccountProjections.getNumberOfAccounts()
}

@michieltcs #StandWithUkraine NEW PROJECTION

@michieltcs #StandWithUkraine NEW STRUCTURE

@michieltcs #StandWithUkraine BASED ON EXISTING EVENTS

@michieltcs #StandWithUkraine REBUILDING STOP APP CLEANUP LOOP OVER EVENTS APPLY
TO PROJECTION START APP

@michieltcs #StandWithUkraine ZERO DOWNTIME LOOP OVER EXISTING EVENTS APPLY TO
NEW PROJECTION USE PROJECTION

@michieltcs #StandWithUkraine ZERO DOWNTIME NEW EVENTS QUEUE LOOP OVER EXISTING
EVENTS APPLY TO NEW PROJECTION APPLY QUEUED EVENTS USE PROJECTION

@michieltcs #StandWithUkraine LONG RUNNING REBUILDS

@michieltcs #StandWithUkraine DISTRIBUTED

@michieltcs #StandWithUkraine DIVIDING THE WORK EVENT EVENT EVENT EVENT EVENT
AGGREGATE EVENT

AGGREGATE INSTANCE EVENT INSTANCE

AGGREGATE INSTANCE EVENT EVENT EVENT EVENT INSTANCE

AGGREGATE INSTANCE EVENT EVENT EVENT INSTANCE EVENT EVENT EVENT EVENT

@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 ... ... ... ... ... ... ... ... ... ... ... ... ... ... ... ... ... ... ... ... ...

EVENT SELLER EVENT SELLER EVENT LISTING EVENT LISTING EVENT LISTING EVENT LISTING EVENT LISTING EVENT SELLER LISTING INSTANCE INSTANCE

EVENT SELLER EVENT SELLER EVENT LISTING EVENT LISTING EVENT LISTING EVENT LISTING EVENT LISTING EVENT SELLER LISTING INSTANCE INSTANCE ?

@michieltcs #StandWithUkraine BACKGROUND TASKS

@michieltcs #StandWithUkraine TRACKING EVENTS EVENT EVENT EVENT EVENT EVENT EVENT
. . . . . . . . . . . . . . EVENT EVENT EVENT

. . . . . . . . . . . . . . EVENT EVENT EVENT TOKEN STORE

. . . . . . . . . . . . . . EVENT EVENT EVENT

@michieltcs #StandWithUkraine TRACKING EVENT PROCESSOR GET NEXT EVENT APPLY TO
NEW PROJECTION LAST EVENT? USE PROJECTION yes no

@michieltcs #StandWithUkraine TRACKING EVENT PROCESSOR @Target(AnnotationTarget.CLASS) @Retention(AnnotationRetention.RUNTIME) annotation class RebuildableProjection(
val version: String = "", val rebuild: Boolean = false )

@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) } } }

@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) }

@michieltcs #StandWithUkraine TRACKING EVENT PROCESSOR trackingEventProcessor.shutDown() trackingEventProcessor.resetTokens() trackingEventProcessor.start()

@michieltcs #StandWithUkraine TRACKING EVENT PROCESSOR class BankAccountProjections { private val
activeAccounts: MutableMap<String, BankAccount> = HashMap() @ResetHandler fun resetProjections() { activeAccounts.clear() } ////  }

@michieltcs #StandWithUkraine EVENT VERSIONING

@michieltcs #StandWithUkraine NEW BUSINESS REQUIREMENTS

@michieltcs #StandWithUkraine CHANGING VIEW ON EVENTS

@michieltcs #StandWithUkraine IRRELEVANT

@michieltcs #StandWithUkraine DIFFERENT FIELDS

@michieltcs #StandWithUkraine WRONG NAME

@michieltcs #StandWithUkraine TOO COARSE

@michieltcs #StandWithUkraine TOO FINE

@michieltcs #StandWithUkraine HOW TO SUPPORT YOUR LEGACY?

@michieltcs #StandWithUkraine Commands can be renamed 1

@michieltcs #StandWithUkraine Commands can be renamed 1 Events are immutable
2

@michieltcs #StandWithUkraine Commands can be renamed 1 Events are immutable
Correct old events with new events 2 3

@michieltcs #StandWithUkraine Ledger Entry Aug 14 Inventory 15600,00 Accounts
Payable 15600,00

@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

@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

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

@michieltcs #StandWithUkraine COMPENSATING ACTIONS class AccountOpened {  String accountId;  String
accountNumber;  } class DuplicateAccountClosed {  String accountId;  } Duplicate account number!

@michieltcs #StandWithUkraine UPCASTING

@michieltcs #StandWithUkraine UPCASTING EVENT STORE OLD EVENT TYPE UPCASTER NEW
EVENT TYPE EVENT HANDLER

@michieltcs #StandWithUkraine UPCASTING EVENT STORE

@michieltcs #StandWithUkraine UPCASTING @Revision("1.0") data class AccountOpened( val accountId: String,
val accountNumber: String )

@michieltcs #StandWithUkraine UPCASTING <org.demo.AccountOpened>   <accountId>80f49161</accountId>   <accountNumber>12345678</accountNumber> </org.demo.AccountOpened>

@michieltcs #StandWithUkraine UPCASTING EVENT STORE ACCOUNTOPENED (1.0) EVENT HANDLER

@michieltcs #StandWithUkraine UPCASTING @Revision("2.0") data class AccountOpened( val accountId: String,
val accountNumberIban: String )

@michieltcs #StandWithUkraine UPCASTING <org.demo.AccountOpened>   <accountId>80f49161</accountId>   <accountNumberIban>NL00ABNA012345678</accountNumberIban>   </org.demo.AccountOpened>

@michieltcs #StandWithUkraine UPCASTING EVENT STORE ACCOUNTOPENED (1.0) UPCASTER ACCOUNTOPENED (2.0)
EVENT HANDLER

@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 } }

@michieltcs #StandWithUkraine VERSIONED EVENT STORE

@michieltcs #StandWithUkraine events_v1 [  {  "id": "12345678",  "type": "AccountOpened",  "aggregateType":
"Account",  "aggregateIdentifier": "1234",  "sequenceNumber": 0,  "payloadRevision": "1.0",  "payload": { ... },  "timestamp": ...  ...  },  ...  ]

@michieltcs #StandWithUkraine COPY & REPLACE

@michieltcs #StandWithUkraine VERSIONED EVENT STORE LOOP OVER EXISTING EVENTS APPLY
UPCASTER ADD QUEUED EVENTS USE NEW EVENT STORE NEW EVENTS QUEUE

@michieltcs #StandWithUkraine events_v2 [  {  "id": "12345678",  "type": "AccountOpened",  "aggregateType":
"Account",  "aggregateIdentifier": "1234",  "sequenceNumber": 0,  "payloadRevision": "2.0",  "payload": { ... },  "timestamp": ...  ...  },  ...  ]

@michieltcs #StandWithUkraine GDPR

@michieltcs #StandWithUkraine "RIGHT TO ERASURE"

@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"

@michieltcs #StandWithUkraine PERSONALLY IDENTIFIABLE INFORMATION

@michieltcs #StandWithUkraine REMOVED

@michieltcs #StandWithUkraine ANONYMIZED?

@michieltcs #StandWithUkraine PROCESSING GDPR ART. 17 REQUESTS RIGHTTOERASUREINVOKED REMOVE FROM
EVENT STORE REMOVE FROM READ MODELS NOTIFY 3RD PARTIES

@michieltcs #StandWithUkraine PROCESSING GDPR ART. 17 REQUESTS RIGHTTOERASUREINVOKED REMOVE FROM
EVENT STORE ? REMOVE FROM READ MODELS NOTIFY 3RD PARTIES

@michieltcs #StandWithUkraine IMMUTABLE EVENTS?

@michieltcs #StandWithUkraine MODIFY DIRECTLY

@michieltcs #StandWithUkraine COPY WITH FILTER

@michieltcs #StandWithUkraine STORE PII EXTERNALLY

@michieltcs #StandWithUkraine STORE PII EXTERNALLY ACCOUNTOPENED EXTERNAL STORAGE Account Id
Account number Name 1234 12345678 John Doe ... ... ... data class AccountOpened( @TargetAggregateIdentifier  val accountId: String )

@michieltcs #StandWithUkraine CRYPTO SHEDDING

@michieltcs #StandWithUkraine ENCRYPTING EVENTS <org.demo.AccountOpened>   <accountId>80f49161</accountId>   <accountNumberIban>NL00ABNA012345678</accountNumberIban>  
<firstName>Foo</firstName>   <lastName>Bar</lastName> ...   </org.demo.AccountOpened>

@michieltcs #StandWithUkraine ENCRYPTING EVENTS <org.demo.AccountOpened>   <accountId>80f49161</accountId>   <accountNumberIban>2dqjHkY8Mc8+cek4vs/9hzgkob4J3fZJNIJh2sAXlJ0=</ accountNumberIban>
  <firstName>N5Y27vd0UbKo6FIu5c7QGQ==</firstName>   <lastName>OSKrzfuuuayuUNXYS5YUug==</lastName> ...   </org.demo.AccountOpened>

@michieltcs #StandWithUkraine ENCRYPTING EVENTS GENERATE EVENT FIND / CREATE ENCRYPTION
KEY ENCRYPT PAYLOAD VALUES STORE   EVENT

@michieltcs #StandWithUkraine DECRYPTING EVENTS LOAD   EVENT FIND ASSOCIATEDENC RYPTION
KEY DECRYPT PAYLOAD VALUES PROCESS   EVENT

@michieltcs #StandWithUkraine SHEDDING THE KEY LOAD   EVENT FIND ASSOCIATEDENC
RYPTION KEY DECRYPT PAYLOAD VALUES PROCESS   EVENT X

@michieltcs #StandWithUkraine OPTIONS https://developer.axoniq.io/axon-data-protection/overview https://github.com/everest-engineering/axon-crypto-shredding-extension

@michieltcs #StandWithUkraine CLOSING WORDS

@michieltcs #StandWithUkraine NO SILVER BULLET

@michieltcs #StandWithUkraine LOTS OF CHALLENGES

@michieltcs #StandWithUkraine (IM)MUTABILITY

@michieltcs #StandWithUkraine AUDIT TRAIL

@michieltcs #StandWithUkraine SCALABILITY

@michieltcs #StandWithUkraine TESTING

@michieltcs @michieltcs #StandWithUkraine THANK YOU FOR LISTENING! twitter: @michieltcs  
mastodon: @[email protected]   email: [email protected] www.michielrook.nl

Dealing with change in event sourced applicatio...

Dealing with change in event sourced applications (ConFoo 2023)

More Decks by Michiel Rook

Other Decks in Programming

Featured

Transcript