 Welcome everyone and thank you for having me and for attending this session. My name is Sergio Maurenzi. I am the responsible and the founder and the CEO of Peperina Software, a cloud native software engineer firm located in Argentina. Well today I will present a project, a modernization project about a tax information system in the public sector where we use a reactive architecture approach. This is the agenda for the presentation because perhaps we have no time. Well the Q&A session will be outside. Well a little bit of context about the project. Córdoba is the second most populous province of Argentina with almost 3.7 million inhabitants. The government agency called DGR or Dirección General de Rentas and Collector UT is a private health company. They are responsible for collecting the taxes for the province, for the states. The taxes are basically fees related to properties, real estate, vehicles, boats and business income. In 2015 the agency started to move the statement of this of this taxes to the website and they start to have many problems of basically availability, performance and that's why in 2019 they decide and ask or request the IT department to think a different solution, to think out of the box a different solution for providing this service to the taxpayers. At that moment they decide one key business objective, allow the taxpayer to search and pay their statement always and the last word always is very strong, you know. The tax agency required to think this solution as I mentioned out of the box because we don't believe in a big bank project, migration project of the legacy system. That's the reason that we provide or we think in a key technology objective that is decouple the functionality of search and payment from the legacy system as much as possible. Well these were the non-functional requirements of the project, I think they explain themselves but basically the solution must be responsive, resilient, always on, performing and have the capability of track every transaction because it's money of the citizens. This out of the box thinking was the reason of the name of the project, Copernicus. We propose and design from the ground up a solution that is accountable, reactive and microservices and also even driven architecture and this is the high-level architecture of the system. Basically it is a pipeline that ingests and processes the changes that we call them events because they are produced in the legacy system every time an important change happens in the business domain. Of course are you seeing oh yes it's it's good it seems good the image because we have a problem with the sorry with the interface and we have also a feedback look from the consumer apps for the commands and queries of the of the repository. Enter into more detail the pipeline start with the data source with the cdc or change data capture every time an entity state from the data source is changed in this database that is an update and here's insert or a delete a db trigger copy this record of the db of the table in another table called Outbox. Then the pipeline continues with the ETL in this process there is some transformation enrichment of the data there are many processors ingesting these these records and then part of the this data path through and a new component that is a business rule layer this is a business rule engine that processes certain events not not all events certain events to classify the depth of the taxpayer which is defined by more than 45 business rules yes in in our country we we have many creative with the definition and exception of the the depth this in this case the business area can change the rule represent a rows in a dmn file dmn is decision modeling notation is a standard of business rules and then the next stage is the messaging system and as the name suggests the main characteristic of this messaging system is it is an event state event current state transfer pattern and that's mean that the event are stateful events and that's why we can't lose any of of them this then the consumer of this stateful event is the aka aka platform or aka cluster who who built the private replica by storing the state in the actual model is a this is a really short period of time that the data is lagging behind from the data source and that's why it is eventual consistency no strong consistency the actual model for the business logic is the core application which is based on distributed and concurrent processing then we choose event sources and it is because the nfr nfr that I mentioned before that is related to the older log of every transaction also we have a cqrs we provide the read data model to allow to be exposed through apis by the bff to the consumer apps the bff the back for front end well and this is the technology stack for every component through the pipeline a patch and ifa for the etl drills for the business rule engine kafkan the messaging system aka actor model programming in scala and the two repository right side and read side are running into a no sequel database kassandra database and the consumer facing application that is that's are in the front for accepting the query and commands are no gs applications all of the components of all the architecture are cluster aware and in the case of kafka scala or yes all of this the aka kafka and the no gs application are living in the kubernetes cluster before starting to code we spend several workshops days with the business and it area trying trying to understand the process the business process we use even storming for trying to capture the whole information about this process and then we have different sessions to moth modeling the data and here here there is a simplified view of this relationship between the the main entities that are in this kind of system the person or taxpayer we represent here we represent here with well the name is to head in spanish is represented the taxpayer the assets of the person are called of head in spanish and the liability or obligations are related these three entities are very related and that's why we are putting in in an aggregate for example if you have a message of pain a liability the message entered by the aggregate root and is root to the corresponding liability to update in this case the balance this also update her parent or in this case the object or asset and then update the balance of the person sujeto and the last step is to emit an event this is a simplified workflow of what happened between these entities well during the project we have many challenges this photo is from cordon mountains i love mountains but guess what this guy is not me i prefer i still prefer technical challenge well during the execution the project well as i mentioned many problems but mainly well here we will talk about throughput observability and scheme evolution one of the requirements of the project was to process the full entity states previous to the beginning of the next year because every year the tax agency generates the liability for the full year for every taxpayer this is a batch process we call it initialization it includes all more 100 million of new records to update during the weekend window on the other hand on the ongoing process every day we process two to eight million of records or events per day when we talk about improve the throughput in the into the pipeline we have many places to tune in are improved for this presentation i just talk about this uh one one of these improvement is related to the message and processing stage i i think that you already knows uh about aka but i don't know if are you familiar with aka streams here yeah well aka stream just a little as a model built in top of aka actor to make the ingestion and processing of stream easy and the alpaca project is a kind of library that's bring us a bunch of connectors to integrate various system in aka streams here we will talk about alpaca only the kafka connector so here's the basic aka stream pipeline you need a little a source and a sink obviously a source has an outlet a sink has an inlet and then you have an arbitrary number of flows in between message flow downstream from the source to the sink but you also have the other channel going to the upstream this is for back pressure in kafka the main level of parallelism is the number of partition in a topic here we have almost 90 topics in this project one for each entity and also we have uh topics for retry retry processing certain events error handling and prioritization each partition is shared by the entity id as we will sell later it is possible to expand the number of consumer instance to keep reading data from the same topic in parallel the consumers subscribe to kafka topic and pass the message into the aka stream in aka streams the kafka consumer is also named source well we have two approach for doing this the first one was um the first approach was about the consumer kafka transaction why because kafka transaction allowed exactly one's processing of message message are consumed transformed and produces to another kafka topic in one atomic operation and all message including the transaction will be successfully written or none of them will be this is the the core of kafka transaction and according to our first test it really worked well but the problem arose when we put many events in the pipeline when we increased the the number of message we changed many settings the number of partition a quantity of brokers the number of aka cluster nodes the processing power of this node but it changed widely we realized that we have a bottleneck in another in another area in another component well the first issue was related to the latency when building transactional pipeline that spawned more than an entity update that that is in the specification or in the the the talks that shone glove for example uh talk about this issue the latency between the components in a transaction but the most important we discover that is also in the documentation is that this kind of consumer is not partition aware and that's mean that if it cannot manage each partition in parallel and that's why there is only one consumer for the 90 partition and that's why we tried with the second consumer a committable partition as source configuration this config use at least one semantics different the source use a single kafka consumer internally but the committing works with partition that's mean that it's partition aware the least one processing means that if the broker that's not received confirmation that the event is consumed successfully the broker will resend the event while this ensure that the event is a get process it also means that can happen multiple times and we solve this problem introducing an identity identity with this configuration we can technically technically size our kafka consumer with the number of partition and the same consumer group run in parallel to achieve high throughput and high rate of the dating gesture is like the the slide show like one consumer for each partition well here's the code example of our consumer main processor i will explain briefly the committable partition source method supports tracking automatic partition assigning from kafka when a topic partition is assigned to a consumer this source will emit a tuple with the assigning topic partition and a corresponding source and when a topic partition is revoked the corresponding source completes here we can identify two map async operators the map async operators if you remember allow us to process the flow in different actors separating the boundaries and allowing us to execute the flow in parallel it's also important to remark that this parallelization of event is done in order we have another method map async and order but in this case we use order because it's necessary for the events that are stateful the first map async that is in the code is the source of each partition and has a parallelization parameter the number of partition and the second map async is the operator to process the message inside this partition here you we use consumer parallelism environment viable to increase the parallel tasks to achieve a better throughput but we couldn't achieve these better results with this increasing this parameter the application turned unstable when we increased this parameter more than five continuing with the code we have the mapping of error in processing the message we have two kind of errors one is related to the timeouts into the pipeline because for example the database is one issue for example in this case the queue for trying to retry this event we implement this way and the other one is normally related to serialization issues when we produce in this case message for another topic is for error topics here we can see the case for successful processing which produce message into the correspondence of sex topic at the end of this code you can see also the sync method and the stream materialization or the writing of the graph well the next challenge is talk about observability of course we have no time in this presentation to show you about every tool every dashboard or graph it's just a glimpse but as you know in distributed system data and application observability is essential because the message that flow in the pipeline are events that represent some state state changes that are important from the legacy application the source of truth in this case we need to ensure that we have no data loss every stage in the pipeline must be observable we need metrics log and traces all of them to look for the trace and each event we use one unique identifier that we define for the cdc we call evid which includes the time stamp when the trigger activate and a sequence number from the source database in this presentation again we will focus on the observability just in main components of the the message system and the business logic and this is a slide well we are showing control center from confluent platform and we can see the consumer lag of message from one of the topics we see the total number of message behind and by partition and also you can see the bar in the middle each bubble represents a partition and a position represent the number of message behind because this tool shows the message consumption in live mode we could discover the latency issue with the transaction consumer that i mentioned before because we saw how the consumer process one partition at a time we we can see this movement one partition at a time but when we change the consumer incommittable partition we could see how the processing of the of the all the topics are in parallel here we can see kafka ui an open source tool to manage uh or to uh to show the message content through the kafka and here you can see the evid that is remark it contains a timestamp and we can compare the timestamp that was generated by the database and what it enters into the kafka topic and then we can see if we have any luck there fortunately liven provides several dashboards uh to help you to help us to observe and monitor the behavior of the actor system in terms of throughput latency and consumption this is one of them of actors and and we have another about akka streams numbers to running throughput latency i will start to move forward sorry another useful tool to capture the trace between the components inside the application open tracing and siebkin and of course we define our own dashboard with cinnamon and kyman we use these libraries to include our metrics of our application next and last challenge present here is uh scheme evolution working with event sourcing system we discover some issues and challenge the first one what is related to the number of read events for recovery operation every time we need to recover the state of an entity the persistent library reads every event from the beginning and replay them to compose the state the problem with entities related as an aggregate is the number of events that they produce for instance when you have a tree like this the parents of the events of the all of the the children all of the three in this example is just five but we have a big family we have thousand of events running of grandchild or child or children because this is a common issue in event sourcing akka persistent library bring us a solution called snapshots you can define for instance that after n events you can take a snapshot of the day of the state and with this feature you don't need to read all the events from the beginning you just read the latest the latest snapshot and the following events this solution reduce raster drastically the response time on recovery the state but introduce another issue or challenge the scheme evolution is about changing the structure of the message according to life cycle of this application the snapshot we use in akka stream are not compatible with the library we use for serialization we use cryo library we try to change the library for Jackson but the performance was worse in the next slide we are going to explain how we solve this issue first of all well we have the snapshot of event of one of the entities in this case the person each snapshot contains the state of the entity and other fields that compose what we call a register this register could include for example in this example three fields other a type address an email now you want to add a fourth field for example the phone number if we change the definition of the schema when we recover the state from the snapshot it fails because the snapshot doesn't know anything about this field what we did here is to create a new case class at the application level to manage this like a new version of the schema and when we read the schema we compare with the different versions and well these are the two case classes the version one called sujeto three contains the all register the version two called sujeto three two is the one that contains the new field and take into account that in our case we have optional fields into the register and that is that is why it's possible to evolve this register this is the code where we do the schema transformation aka persistent support event sourcing with persistent actor the behavior of persistent actor is defined by implementing receive recover receive command and here we override the method to change the behavior the persistent actor receive recover define how the state is updating during the recover by handling event and the snaps a snapshot offer message and here's the way we compare the snapshot with the corresponding schema version one or two and in the case it is the old version the snapshot is copied and updated final slides since the start of the project in 2019 the engineer team had many challenge and on learnings the rollout in production starting in 2021 last year mid-night mid-night last year the system won't live with the initialization process and immediately starting to ongoing the processing gesting of all the events from the source system but in terms of use from the consumer point of view a blockchain based digital wallet it was the first consumer just started to use the data of our platform in a light way this was part of our strategy to allow to tune in the platform but then in the second half of this year the system started to be integrated with different other platform and systems IVRs website new revenue system etc this platform is considered the glue between the legacy system and the new customer facing systems we are continuing developing new features business features like a discount model module that is by nature reactive need to be reactive and of course we continue to evolve in and tune in the platform especially increasing the performance of data processing a robustness of the events in the pipeline well key takeaways there is one universal truth all things that can fail will fail you know it's not matter of if but it's matter of when if a single consumer face that is an accountant for in the design state replicas can divert permanently that's why do not underestimate the complexity of maintaining a consistent state across microservice in a high availability environment at the scale the second key lesson is about restriction and non-functional requirements review carefully for non-functional requirements that could change your architecture decisions for example the security audit logs never the business partner never then asked to show this part this part we choose event sourcing for this the third key lesson is about the resilience and responsiveness as you mentioned today in the morning with these asynchronous and decoupled components we could see how the system reacts to failures like application nodes going down or issue with the network or with the database without disrupting the service because well in one case they activate the circuit breaker or a spring bearing resolver other part of the platform this is really amazing in conventional system if you have an issue like this the application becomes unavailable in any case the observability is essential as I mentioned before in distributed system okay are you already here working with distributed system is hard but worried at the end we have more resilient and robot system and our taxpayer and citizen deserve it thank you very much