Event-driven architecture on the modern stack of Java technologies

common-model module unsurprisingly contains the code that is shared by all three microservices, specifically:

This section covers:

The service stores data about all books and their authors, allows to access and manage that data by its REST API and allows a user to borrow and return a book. When some domain event occurs (for example, a new book was added to the library or a book was lent by a user), the service stores an appropriate message in its outbox table implementing thus the first part of Transactional outbox pattern. The second part of the pattern implementation is to deliver the message from a microservice’s database to Kafka; it is done using Kafka Connect and Debezium and is discussed in Connectors configuration section.

The main class looks like this:

@ImportRuntimeHints tells the service that additional reflection data should be used by importing it from common-model module.

@EnableScheduling is needed to enable execution of scheduled tasks; this service, as the other two, polls periodically inbox table of its database for the new messages. This is the part of Inbox pattern implementation that is covered in detail in the User service section.

REST API is described in OpenAPI format. openApiGenerate task of org.openapi.generator Gradle plugin is configured as shown in the build script and produces controllers, DTOs, and delegate interfaces. You need to implement the delegate interfaces to specify how to process incoming requests. This project contains implementations of delegates for:

A generated controller calls an implementation (or one of implementations) of an appropriate delegate interface to process an incoming request.

The config for testing environment also enables limited converters for books and authors; that means that a user is allowed to change fewer fields compared to default converters. Also, the values for these fields will be coerced to lie in a given range, for example, when updating a book, a user is allowed to change only its publication year and the value of the field will lie in a range between 1800 and 1950 years even if the user entered a value not in that range.

In the introduction, I described the high-level algorithm of what book-service (as a part of a whole system) does when it processes a request to its REST API. Now we will look at what the service does in detail, including how the Transactional outbox pattern is implemented on the microservice side. Again, we consider an update of a book scenario; it is initiated by a call of PUT {{baseUrl}}/book-service/books/{bookId} REST API method. You can start the process locally by calling PUT https://localhost/book-service/books/3 and passing the following request body:

both the considered implementations of BooksApiDelegate interface call BookService.update() method which opens a new transaction:

As said in the introduction, the following two actions are performed inside the transaction:

The first action is quite common; let’s consider the second. saveBookChangedEventMessage does the following:

At first, the saveBookChangedEventMessage method creates an instance of OutboxMessageEntity:

by using the following method:

BookChanged is one of event types in this project. There are two main types of messages in event-driven architectures:

In different sources, in addition to the above, you can find also the following message types:

According to this classification, events in this project are documents because they contain the latest state (and some can also contain a previous state of an entity), but I decided to leave familiar naming and call them events.

In the considered scenario, the payload field has the type of the mentioned CurrentAndPreviousState model containing both new and old states of the book; using it, any consumer can process the message in any way. For example, one consumer can be only interested in the latest state of a book, while another needs to receive the delta between the new and the old states of the book. In this project, the only consumer of BookChanged type of event is user-service that needs the delta to notify a user about changes in the library but still I preferred more flexible data model which may be suitable for consumers with other needs in the future. Such an approach also allows to use just one message type (BookChanged) to describe any change in a book instead of multiple events (BookPublicationYearChanged, BookNameChanged, etc.); but such a structure doesn’t give information about what change triggered the event, unless we specifically add it to the payload. It should also be noted that such messages need more storage space for Kafka. Do your own research to find out which message type is best suited for your system: events or documents.

All the fields used in OutboxMessageEntity constructor above except the type field are required by Outbox Event Router single message transformation; the type field is required by CloudEvents converter. This will be covered in Connectors configuration section. All the fields are mapped to the appropriate columns of the outbox table:

The save method stores OutboxMessageEntity through OutboxMessageRepository to outbox table:

As you see above, the outbox message is deleted right after its creation. This might be surprising at first, but it makes sense when remembering how log-based CDC works: it doesn’t examine the actual contents of the table in the database, but instead it tails the append-only transaction log. The calls to save() and deleteById() will create one INSERT and one DELETE entries in the log once the transaction commits. After that, Debezium will process these events: for any INSERT, a message with the event’s payload will be sent to Apache Kafka; DELETE events, on the other hand, can be ignored because they don’t require any propagation to the message broker. So we can capture the message added to the outbox table by means of CDC, but when looking at the contents of the table itself, it will always be empty. This means that no additional disk space is needed for the table and also no separate house-keeping process is required to stop it from growing indefinitely.

Further, the transaction is committed and the following happens:

The service also is an example of a consumer of data replicated from a database of another service. The database of book-service contains user_replica table with just two columns:

The source table, library_user of the database of user-service, has more fields but in the source connector we specify that only id and status fields will be included to change event record values. Then the record will be published to the streaming.users Kafka topic which is listened by the sink connector; it will insert the records to the user_replica table.

book-service can use the data from the table in a scenario when a user take a book from the library. To lend a book, it is needed to perform POST {{baseUrl}}/book-service/books/{bookId}/loans request (for example, POST https://localhost/book-service/books/5/loans with the following body:

The method to borrow a book acts similar to the method of a book update: it starts a new transaction that includes a change of a business data and insertion of an appropriate outbox message:

If you set user.check.use-streaming-data: true in the service’s config, then useStreamingDataToCheckUser is true which tells the method to use streaming data to check a user before creation of a book loan: the method tries to find an active user with the specified id. If it is successful, the method changes business data (that is, it inserts into the database BookLoan entity with specified the book and the user) and stores an outbox message of BookLent type. Otherwise, the method throws an exception and returns it to a user in a response body.

If you set user.check.use-streaming-data: false, the scenario differs from the one discussed above only in that the lendBook() method skips checking that the specified user exists and is active despite the fact that the user_replica table still exists, that is, this simulates the case where you have not configured replication of the user data table. However, the further continuation of the saga may differ because now we rely on the user verification to happen in user-service. If we use the request body above with "userId": 2, everything is fine, and the saga continues in the next service, notification-service. But if we use in the request body certainly not existing in the database userId (for example, 502), the user-service receives a message of BookLent type containing userId value that doesn’t exist in its library_user table; therefore, user-service publishes a message of RollbackBookLentCommand type which will mean that book-service should perform a compensating transaction that is to rollback somehow the creation of a BookLoan entity. The message eventually occurs in inbox table of the database of book-service through user.source and book.sink connectors and is processed using Inbox pattern. When processing the message, the cancelBookLoan() is called:

The method sets the status of the BookLoan entity to Canceled. After that, it publishes a message of BookLoanCanceled type that user-service will process as well as one of BookLent type.

You can see that for cancelBookLoan() method transaction propagation differs from other methods of BookService: it is MANDATORY instead of REQUIRES_NEW. This is because the method is initiated by processing of an inbox message of RollbackBookLentCommand type (the only inbox message type processed by book-service) and the only transaction should include not just changed business data and an appropriate outbox message but also a change of the status of the InboxMessageEntity; the transaction consisting of these three changes starts when the processing of the message begins. Otherwise, if there would two separate transactions, for implementation of the Inbox pattern and Transactional outbox pattern, there could be a case when the latter completes successfully but the inbox message can’t be updated in its table with a new status (Completed or Error) so it stays in the New status which leads to an infinite loop of processing the same event and therefore to infinite number of produced outbox messages.

user-service processes all messages published by book-service and we will consider the implementation of Inbox pattern in more detail on its example (see User service section), specifically, using messages of the mentioned BookChanged, BookLent, and BookLoanCanceled types.

The following is a combined table containing a full list of message types produced by the book-service and a full list of REST API methods of the service (Initiated by REST API method table column):

Because of, as mentioned above, the service implements Inbox pattern, it would be possible to put all commands for a change in the library initiated by methods of the REST API to the inbox table for further processing. That would allow to return the response to a user almost immediately; the response would contain the information that the service accepted the request and will process it soon. However, it is unknown whether the processing will complete successfully or will fail, so I decided to make the service respond to a user after committing the changes (updated business data with an appropriate outbox event) to the database. The only event type processed by book-service is RollbackBookLentCommand (it is published by user-service).

This section covers:

This service stores information about users of the library. It also processes all the messages from inbox table (the messages are published by book-service (see the previous section)) implementing thus the second part of Inbox pattern. The first part of the pattern implementation is to deliver the message from Kafka to a microservice’s database; it is done using Kafka Connect and Debezium and is discussed in Connectors configuration section. As a result of message processing, the service creates commands to send appropriate notifications to the users. For example, if a user borrowed a book and later the book’s data has changed, the user will receive a notification about that. There is also a scenario when the service shouldn’t proceed with the message and should publish thus a compensating transaction (a message of RollbackBookLentCommand type); that was considered in the previous section. As book-service, this service sends the messages using Transactional outbox pattern. The service’s database contains library_user table which is the data source for user_replica table of the database of book-service.

The main class is the same as in book-service and looks like this:

The high-level algorithm of what is done inside user-service (as a part of a whole system) when it processes an inbox message was described in the introduction. We will now take a detailed look at what the service does, including how the Inbox pattern is implemented on the microservice side. We will continue to use an update of a book scenario; now it will help us to consider Inbox pattern implementation.

The structure of the inbox table looks like this:

The table is populated with the messages passing in this way: book-service → book database → book.source connector → Kafka → user.sink connector → user database. As you see from the listing above, the messages in inbox table are not in CloudEvents format; instead, they have a custom structure containing only four columns corresponding to appropriate fields of CloudEvents envelope: id, source, type, and payload (data field of the envelope). These columns are sufficient for the user-service to process an event. The processing algorithm expects that the payload column contains a message in JSON format. Using Hibernate, the mapping between the table and InboxMessageEntity looks as follows:

I designed the processing of inbox messages based on the following requirements to the parallelization of the processing:

Obviously, we don’t want a message to be processed by more than one instance. For that, I use pessimistic write lock that ensures that the current service instance obtains an exclusive lock on each record (message) from the list to be processed and, therefore, prevents the records from being read, updated or deleted by another service instance.

Further, considering the processing on a service instance, there should be a transaction that acquires the pessimistic write lock when retrieving a batch of inbox messages. The processing of the messages should be performed in the same transaction but if we do so, we can only process the messages in the sequential manner because it is technically difficult in Spring to execute the processing of each message in a separate thread but inside the original transaction. Therefore, to support the parallel processing of the messages by a service instance, each message should be processed inside its own separate transaction. Based on the above, I designed the algorithm of messages processing consisting of two steps:

If there is only one instance of our service, or we don’t need to process the messages by multiple instances, the algorithm will be much simpler, specifically, we won’t need the first step of the algorithm as we don’t need to define which inbox messages are processed by which instances. If there are multiple instances of our service, we can use, for example, shedlock library which prevents the task from being executed by more than one instance.

Let’s look at how these thoughts are implemented in code.

InboxProcessingTask starts processing new messages in the inbox table at a specified interval (5 seconds) and looks like this:

The method to mark inbox messages to be processed by the current instance receives batchSize — configuration parameter — which tells how many new messages (ordered by created_at column’s value) in the inbox table should be marked. It looks like this:

Eventually, each message in the batch is updated (see saveReadyForProcessing method) with the new status (ReadyForProcessing) and the name of the instance that should process the message. The method returns the number of messages that were marked.

The messages to process are retrieved using InboxMessageRepository with the aforementioned pessimistic write lock:

The lock is in effect within the transaction that wraps markInboxMessagesAsReadyForProcessingByInstance method.

Next, the algorithm retrieves messages that are intended to pe processed by the current instance using getBatchForProcessing method. The messages are processed in parallel using the following approach:

Usage of applicationTaskExecutor bean of AsyncTaskExecutor type allows to process each inbox message in a separate thread. The subtaskTimeout parameter specifies how long to wait before completing a subtask exceptionally with a TimeoutException. The thread of the main task will wait until all subtasks finish, successfully or exceptionally. The applicationTaskExecutor bean is constructed as follows:

This way we provide custom applicationTaskExecutor bean that overrides the default one; it uses JDK 21’s Virtual Threads starting a new virtual Thread for each task.

Each inbox message is processed as follows:

Next, InboxMessageService delegates processing of a business event inside a message (inboxMessage.payload) to IncomingEventService:

In the considered scenario of a book update, the processing of the event looks like this:

If the book is borrowed by a user, IncomingEventService:

It turns out that in the scenario under consideration, the final step in implementing the Inbox pattern is using the Transactional outbox pattern. But in this case, it is a truncated version of the pattern because, unlike book-service, it doesn’t include storing "business" data. The other important difference is that outbox table has an additional topic column that determines the last part of a topic name to which a message should be sent; it can be either notifications or rollback depending on a message type so a resulting topic name derived by the appropriate connector will be either library.notifications or library.rollback.

In this project, the order of inbox messages processing is not the main concern: it is supposed that the main library entities such as books and authors are not changed very often. But if in your case it is important to process inbox message in the same order in which they were produced, you need to do your own research and change the algorithm. While we assume that the order is preserved when the messages from outbox table of book database are passed to inbox table of user database (Kafka guarantees the order within a partition, and it is supposed that both source and sink connectors also do it), we need to ensure that the inbox messages are processed by user-service in the same order. For example, in the considered scenario, how can we ensure that two notifications are created by user-service in the same order in which two updates of the same book occurred in book-service (let’s suppose that the updates occurred almost simultaneously)? First, in this case, the processing of the inbox messages must not be parallelized neither between multiple instances of the service nor between multiple threads inside an instance. For example, if we don’t meet the first condition, and the contents of the inbox table looks like this:

it is not guaranteed that uuid1 message will be processed by one instance before uuid51 message will be processed by another instance (considering the batch size parameter is 50). Therefore, it is possible that notifications will be delivered to a user that borrowed Book 1 in a different order than the original.

The same logic applies to the case when one instance processes inbox messages about changes in the same book in two threads.

Therefore, both updates of the same book should be processed by the same instance and sequentially. To do that, considering that we are using Kafka, we should use the same mechanism by which it is guaranteed that messages within a Kafka partition are inserted to inbox table in the same order in which they were inserted to outbox table. It is keys of Kafka messages. The produced Kafka messages about two updates should have the same key; for that, I use id of the book by configuring Outbox Event Router in book.source connector (see transforms.outbox.table.field.event.key property) that will be discussed in detail in Connectors configuration section. We can add event_key column to inbox table and, use it in where clause when selecting messages to process by a service instance, and then process messages with the same key sequentially.

If you understand that your service will not process a large number of messages, and one instance of the service is enough for that, you might consider processing all new messages (that is, from all Kafka partitions of a topic) sequentially by that instance.

Messages of other types are processed similarly, but it should be noted who is notified in different cases:

Let’s talk about how to initiate a compensating transaction for the scenario discussed in the previous section. Remember, when the user-service receives a message of BookLent type containing userId value that doesn’t exist in its library_user table it should publish a message of RollbackBookLentCommand type that should initiate execution of a compensating transaction in book-service that is to rollback somehow the creation of a BookLoan entity. There is nothing unusual in the implementation of this scenario: the service performs user check and if a user is not found, a message of RollbackBookLentCommand type is stored in the outbox table (in the same way as is stored a command to send a notification):

I don’t remove the messages from inbox table immediately after the processing because I think, there are cases when you may need a message some time after processing, for example, to repeat it or investigate an error. But definitely, after some time, you can delete the message; for that, you may need to implement separate house-keeping process to stop the table from growing indefinitely.

In this project, formatting the user notification is done using kotlinx.html library. The library allows to create an HTML document along with custom CSS that further, in notification-service, can be sent to a user’s email or over WebSocket. We will see that formatting in Testing section.

Note: when considering book-service we have seen that there is an alternative to the connector to pass messages from outbox table of book database to Kafka. Similarly, an alternative to pass messages from Kafka to inbox table of user database is to create a component in user-service that listens an appropriate Kafka topic and puts messages from the topic to the inbox table.

This section covers:

This service receives commands to send notifications and delivers them to users through email or, for demo purposes, through WebSocket. As well as book-service and user-service, notification-service is an example of implementation of Inbox and Saga (as the final participant) patterns; but unlike those services, it doesn’t need Transactional outbox pattern. The implementation of these and other event-driven architecture patterns is a key part of microservices implementation in this project and was considered in Book service and User service sections. So in this section, we will focus on the message delivery, specifically, through WebSocket, because the email delivery is quite straightforward.

If you want to test the email delivery, you need to:

Please note that email delivery implementation is only for testing purposes; it sends an email to the sender.

WebSocket is an application level protocol providing full-duplex communication channels over a single TCP connection that is using TCP as transport layer. STOMP is a simple text-based messaging protocol that can be used on top of the lower-level WebSocket.

To implement WebSocket server, first, we need to add the org.springframework.boot:spring-boot-starter-websocket dependency to the build script.

In this project, WebSocket is only used to notify users about some changes in the books in the library, and we don’t need to receive messages from users that is the messaging is unidirectional. The application is configured to enable WebSocket and STOMP messaging as follows:

WebSocketConfig is annotated with @Configuration to indicate that it is a Spring configuration class and with @EnableWebSocketMessageBroker that enables WebSocket message handling, backed by a message broker. The registerStompEndpoints method registers the /ws-notifications endpoint, enabling SockJS fallback options so that alternate transports can be used if WebSocket is not available. The SockJS client will attempt to connect to /ws-notifications and use the best available transport (WebSocket, XHR streaming, XHR polling, and so on). The configureMessageBroker method implements the default method in WebSocketMessageBrokerConfigurer to configure the message broker. It starts by calling enableSimpleBroker to enable a simple memory-based message broker to carry the messages on destinations prefixed with /topic.

A command to send notification from the inbox table is processed as follows:

At any environment except test, the emailService is an instance of EmailServiceWebSocketStub:

The WebSocket client is a part of the user interface that was developed within the notification-service for demo purposes. The client receives messages the sending of which was shown in the previous section; the messages are displayed in the UI.

The following dependencies are used to create the UI:

The /resources/static folder of the service contains the resources for the frontend application: app.js, index.html, and main.css. The most interesting is app.js:

The main pieces of this JavaScript file to understand are connect and showNotification functions.

The connect function uses SockJS and stomp.js (these libraries are imported in index.html) to open a connection to /ws-notifications (STOMP over WebSocket endpoint configured in the previous section). Upon a successful connection, the client subscribes to the /topic/library destination, where the service publishes notifications to all users. When a notification is received on that destination, it is appended to the DOM:

Notification delivery will be demonstrated in the Testing section.

It would be also possible to set up email streaming from the database to some email provider (in this project, it is Gmail). It could be done, for example, by storing notifications in outbox table of the considered service and setting up two connectors: Debezium PostgreSQL connector that would read messages containing the notifications from outbox table and send them to Kafka and some Email sink connector that would put notification to the provider users' inboxes. I decided though that it would be over-engineering and current implementation is enough for this project.

This section covers how to set up the following:

GraalVM native images are standalone executables that can be generated by processing compiled Java applications ahead-of-time. Native images generally have a smaller memory footprint and start faster than their JVM counterparts.

The Spring Boot Gradle plugin automatically configures AOT tasks when the Gradle Plugin for GraalVM Native Image is applied. In such case, bootBuildImage task will generate a native image rather than a JVM one.

While there are many benefits from using native images, please note:

Building a standalone binary with the native-image tool takes place under a closed world assumption (see Native Image docs). The native-image tool performs a static analysis to see which classes, methods, and fields within your application are reachable and must be included in the native image. The analysis cannot always exhaustively predict all uses of dynamic features of Java, specifically, it can only partially detect application elements that are accessed using the Java Reflection API. So, you need to provide it with details about reflectively accessed classes, methods, and fields. In this project, it is done by Spring AOT engine that generates hint files containing reachability metadata — JSON data that describes how GraalVM should deal with things that it can’t understand by directly inspecting the code. To provide my own hints, I register some classes and their public methods and constructors for reflection explicitly:

In the future, that approach may change.

Then, CommonRuntimeHints is imported by all microservices, for example:

In this project, generated hint files can be found in <service-name>/build/generated/aotResources folder. You can see that besides the custom reflection hints it also contains automatically generated hints, for example, for the resources that are needed for the microservices such as configuration (application.yaml) and database migration (/db/migration/*) files.

To build a Docker image containing a native image of Spring Boot application, you need to execute ./gradlew :<service name>:bootBuildImage in the root directory of the project. On Windows, you can use this script to rebuild all three microservices and restart the whole project. The script contains commands for parallel and sequential building of all microservices. Please note that using this approach you don’t need Dockerfile for any microservice.

Let’s consider the build configuration on the example of book-service: the command is gradlew :book-service:bootBuildImage. bootBuildImage is a task provided by Spring Boot Gradle Plugin (org.springframework.boot); under the hood it uses Gradle Plugin for GraalVM Native Image (org.graalvm.buildtools.native) to generate the native image (rather than a JVM one) and packs it into OCI image using Cloud Native Buildpacks (CNB).

The build config of the service looks like this:

In the config of BootBuildImage task, I set:

You can find more properties to customize an image in the docs.

Of course, you can still build a Docker image of a regular JVM-based application by applying an appropriate build configuration which can help you to reduce build time of the services and thus speed up local development. A quick way to do that is to remove org.graalvm.buildtools.native from the list of plugins and remove buildpacks and environment settings from bootBuildImage task configuration; that will allow you to get JVM-based Docker images but without health check support; to build JVM-based Docker image with health check support, do your own research. To speed up the local development, you can also start the microservices from your IDE.

If you want beans that are created based on a condition in the native image, you have to set up the environment when building the application. Spring profiles are implemented through conditions, so I have to customize ProcessAot Gradle task as shown above. test profile defines several beans implementing some REST API limitations intended for testing environment. So if you deploy this project locally, don’t forget to remove this customization.

book-service defines only profile-specific beans all of which are annotated @Primary so you don’t need to specify the profile at runtime. Unlike book-service, notification-service defines not only profile-specific beans but also profile-specific application properties; to fully enable its test profile, you also need to specify the profile’s name as en environment variable as you do for a regular JVM-based Spring Boot application; for example, you can do it by adding the environment variable to Docker Compose service definition as follows:

health-checker buildpack contributes a process called health-check. It is recommended that you use this abstract name because it won’t change if for some reason the underlying health checker process (thc) will change.

The health-check command will be used by Docker containers of the services as follows:

The thc tool looks only at the HTTP status: a service is considered OK when it is >= 200 or < 300.

We start with the persistence layer because this is where all the data changes initially occur. In this project, I use the official Docker image.

As you can see from the listing above, Docker containers of databases of all three microservices are configured the same, for instance:

I think most readers are familiar with most of this listing that is with the Docker Compose service definition; let’s just focus on the database configuration. You can find different options on how to configure PostgreSQL database running in a Docker container; I use the simplest approach customizing postgres database server program through command option. The approach allows me to not create a full configuration file, but only specify the properties that I want to override:

You can find more Postgres configuration parameters, particularly for WAL and replication, in the docs.

It is not required to set wal_level for notification database because, unlike the other two databases, its appropriate microservice (notification-service) doesn’t need to implement Transactional outbox pattern or other type of data streaming so there is no source connector that reads WAL of notification database. I decided to leave the setting for uniformity of databases configuration.

Write-ahead logging is a standard method for ensuring data integrity. A detailed description can be found in most (if not all) books about transaction processing. Briefly, WAL’s central concept is that changes to data files (where tables and indexes reside) must be written only after those changes have been logged, that is, after WAL records describing the changes have been flushed to permanent storage that. WAL files can be used for database backup, crash recovery, and data replication; the latter is the most interesting for us, as will be shown below.

The WAL (or transaction log) is written in a format your non-Postgres consumers won’t understand. Logical decoding is the way Postgres enables you to translate (decode) the WAL contents into a form that the consumers can use. When a row is changed in a Postgres table, that change is recorded in the WAL. If logical decoding is enabled, the record of that change is passed to the output plugin. The output plugin transforms that record from the WAL internal representation into the format a consumer of a replication slot desires (for example, SQL statement, JSON, or Protocol Buffers). In the context of logical replication, a slot represents a stream of changes that can be replayed to a client in the order they were made on the origin server. The consumer can be any application of your choice or other component such as an instance of Debezium PostgreSQL connector as in this project. Then the transformed data exits Postgres via the replication slot. Finally, the consumer connected to Postgres receives the logical decoding output.

In this project, I use native pgoutput plugin for the Debezium PostgreSQL connectors; if you want to use another plugin, you need to install it. When using the pgoutput plugin, in addition to replication slot, a publication should also be created to stream events. Similar to replication slot, publication is at database level and is defined for a set of tables.

In Postgres, replication slots persist independently of the connection using them and know nothing about the state of their consumer(s). The replication slots will prevent removal of WAL files and other resources even when there is no connection using them. If a consumer fails forever and cannot be recovered (an orphaned replication slot) or when a replica cannot replay the WAL segments fast enough, the WAL files will just pile up and eventually pg_wal directory may run out of space. So if a slot is no longer required it should be dropped.

Let’s try to put the knowledge we have gained into practice. First, open pgAdmin (localhost:8102) or connect to the database Docker container by doing the following:

At first, we ensure that the settings specified above were applied. The pg_file_settings view provides a summary of the contents of the server’s configuration file(s); select name, setting from pg_file_settings; query returns the following:

timezone setting has the default value and there is no wal_level setting because passing command-line options, as specified above, have changed no configuration file. We can found those settings using select name, setting, source from pg_settings where name like 'wal_%' or name ilike 'timezone'; query:

Among WAL-related settings, we see two custom parameters provided through the command line.

Further, let’s look at WAL files and its contents. In Postgres, WAL files are stored in pg_wal directory under the data directory (PGDATA). In the selected Docker image, it is /var/lib/postgresql/data/pg_wal directory. For examining its contents, you can use file system or pg_ls_waldir() system function. select * from pg_ls_waldir(); query returns the following:

To see the content of a WAL file, you need to exit from psql terminal and use pg_waldump command that displays a human-readable rendering of a WAL file: pg_waldump -p /var/lib/postgresql/data/pg_wal 000000010000000000000001 (the file is too large to list its contents here).

Let’s now see how we can query the existing publications and replication slots in the system. For more clarity, connect to user-db Docker container and user database:

Now you can:

As you see above, there are two publications and two replications slots in the user-db. This is because two Debezium Postgres source connectors are used in the project: the first is needed for streaming messages from outbox table, the second is needed to replicate data from library_user table. If you return to book-db Docker container and book database, you will see that it contains one publication and one replication slot because there is only one source connector to it.

Now, let’s use the book database to see in practice how logical decoding works and, accordingly, how a consumer (one of Debezium PostgreSQL connectors) receives the logical decoding output that is events containing changes occurred in database tables. We’ll do it with test_decoding that is an example of a logical decoding output plugin. It receives WAL through the logical decoding mechanism and decodes it into text representations of the operations performed. A publication is not needed, unlike pgoutput plugin. For testing, we will use pg_recvlogical utility. In the bash terminal, do the following:

Now you can use REST API of book-service: for example, if you create and update a book, you will probably get something like the following:

As you see, two REST requests initiated two database transactions, each of which contains two main parts:

In this project, the consumer is only interested in the second part, more precisely, only the INSERT to outbox table, so the connector is configured to ignore all change events except those in the outbox table through table.include.list property. Additionally, as noted earlier, when using the pgoutput plugin, there must be a publication for the replication slot. If the publication doesn’t exist, it can be created by the connector; publication.autocreate.mode property defines change events in which tables should be emitted by the created publication. By default, it publishes all changes from all tables: the property’s default value is all_tables; to make sure that it emits change events only from tables listed in the table.include.list property, set publication.autocreate.mode property’s value to filtered, as in the considered connector. Therefore, filtering is done by the publication, that is, on the database side. The advantage of using the filtered publication with pgoutput is that you avoid the network bandwidth costs of sending unnecessary WAL entries over the network to Kafka Connect and Debezium. All other source connectors, user.source and user.source.streaming are configured similarly. To make sure that the publication is configured as required, execute table pg_publication_tables; query for book database:

Do not forget to delete the replication slot for the reasons stated above in this section; you can do it using pg_recvlogical -U postgres -h localhost -d book --slot test_slot --drop-slot.

This section covers:

Kafka Connect is a tool for scalable and reliable data streaming between Apache Kafka and other systems, such as databases, key-value stores, search indexes, and file systems, using connectors.

In this project, I use Kafka Connect Docker image maintained by Debezium community. There is also Confluent’s Docker image.

Kafka Connect Docker container runs as follows:

More information about these and other environment variables can be found here.

In this project, I use Kafka image maintained by Bitnami. It runs in KRaft mode, that is, without ZooKeeper. It is configured like this:

As mentioned earlier, I disable automatic topic creation, as it is done by Kafka Connect.

Kafka, at its core, only transfers data in byte format. There is no data verification that’s being done at the Kafka cluster level. But consumers of events/messages emitted by Debezium (for example, from outbox tables like in this project) should be aware of their structure and data types. This problem can be solved with the help of schemas. As a result, we have the following options:

In the context of an event-driven architecture using Kafka, schema registry is a component of your infrastructure that provides a centralized repository for managing and validating schemas for topic message data, and for serialization and deserialization of the data over the network. Producers and consumers to Kafka topics can use schemas to ensure data consistency and compatibility as schemas evolve. Schema registry is a key component for data governance, helping to ensure data quality, adherence to standards, etc. Kafka Connect can use a schema registry to store schemas representing structure of messages. By using a schema registry in combination with Apache Avro data serialization format, it is possible to significantly decrease the overall message size.

A producer (a source connector as in this project), before sending the data to Kafka, talks to the schema registry first and checks if the schema is available. If it doesn’t find the schema, it registers and stores the schema in the schema registry. Further, the producer serializes the data with the schema and sends it to Kafka in binary format; a unique schema ID is included in the message (in its payload or headers depending on the configuration of the producer). When the consumer (a sink connector) processes this message, it will communicate with the schema registry using the schema ID it received from the producer and deserialize it using the same schema.

In this project, I use Apicurio Registry; an alternative is Confluent Schema Registry. Both platforms support JSON, Google Protobuf, and Apache Avro schema formats, while Apicurio Registry supports much more, such as OpenAPI, AsyncAPI, GraphQL, Kafka Connect, and other schemas. The items stored in Apicurio Registry are known as artifacts.

Apicurio Registry provides several options for the underlying storage of registry data: in-memory, PostgreSQL, and Apache Kafka; in this project, I use the version with in-memory storage:

Note that this storage option is suitable for a development environment only because all data is lost when restarting Apicurio Registry.

Let’s consider properties of a source connector configuration considered earlier that are related to Apicurio Registry:

First, the configuration tells that an emitted message should be in CloudEvents format and serialized using Avro as well as its data field. Next, Apicurio-specific properties follow:

You can find more info about Apicurio Registry configuration options in the docs.

In this project, Apicurio Registry UI is accessible at http://localhost:8080 in your browser; if you will initiate several domain events through REST API of book-service, the page with the list of existing artifacts will look like this:

Here are a single CloudEvents schema and three schemas of data models, that is, of a content in CloudEvents' data field. We have discussed how their names are derived in Connectors configuration section.

Page of an artifact looks like this:

Every version of every artifact has a single globally unique identifier that can be used to retrieve the content of that artifact. The artifact ID and other parameters under which a message schema is registered in Apicurio Registry are determined with lookup strategies. The strategies are used by the Kafka producer serializer and implement ArtifactReferenceResolverStrategy. There are several lookup strategies to use with Avro; if you change value.converter.avro.apicurio.registry.artifact-resolver-strategy from RecordIdStrategy to TopicRecordIdStrategy, the IDs of the artifacts will also include a topic name, so there will be a schema of a CloudEvents envelope per each topic:

Returning to the page of an artifact, you can see that Book artifact has two versions, and they are very different from each other. This is because when you create a book, a single book model is published, and when you update a book, a model contains current and previous states (see Book service for the reasons). You can change that behavior by setting a different name of the aggregate for the case of a book update, for example, CurrentAndPreviousBookStates so there will be two different artifacts for cases of a book creation and update. But even if models for both cases would contain a single book, there could be multiple versions of the book model because they could change in optional fields: for example, a created book has empty value for its currentLoan field or doesn’t include the field at all so Kafka Connect/Debezium can’t define a type of the property or doesn’t include it to the schema; at the same time, an updated book can be previously borrowed by a user so its currentLoan field is not empty and has a certain type. Addressing this question, I created this issue. Notification artifact has a single version as expected.

If in your project it is possible that multiple services can publish different schemas with the same name you probably need to do one of the following:

As for CloudEvents schema in a schema registry, I think it is worth to at least consider a possibility to provide the predefined schema as in my opinion there should be a single version of the schema in a schema registry, and it should be the same in any project. That follows from CloudEvents concept.

In this project, the reverse proxy is needed to route requests from a user to book-service (that is, to its REST API) and notification-service (index.html with CSS and JS resources and WebSocket endpoint). Initially, I used Nginx for that. During the implementation of the project, I formulated the following requirements for the reverse proxy module in addition to HTTP and WebSocket proxying:

It turned out that making project accessible through HTTPS is much easier using Caddy web server, that is, you don’t have to do anything special because HTTPS is used automatically and by default on any environment including localhost. But let’s start with the proxying:

DOMAIN environment variable can be eda-demo.romankudryashov.com or localhost depending on the environment.

As you can see, configuration for reverse proxying using reverse_proxy directive is equally simple for both HTTP/HTTPS and WebSocket protocols.

The only non-obvious thing for me was that rate limiting functionality is not included to the web server by default; for that, if you run Caddy in a Docker container, you need to create a custom Dockerfile where you add HTTP Rate Limit Module, and build Docker image from it. Then you can order rate_limit before basicauth in the Caddyfile.

If you try to access UI to receive user notifications, you may see a warning screen from a browser:

You can proceed with that but if you want to access the UI locally through plain HTTP, add localhost:80 (or use other port) by editing the list of addresses of a site in Caddyfile.

Caddy Docker container is defined like this:

It is very important to persist the data directory because, in particular, TLS certificates are stored in that directory. If you run Caddy in a Docker container, as in this project, use a volume for persisting data generated by and used by the container. Otherwise, after each restart of a project, and therefore recreating the container, Caddy won’t find a certificate locally and thus will try to obtain it; if you are actively developing, testing, or deploying your project, you may exceed Let’s Encrypt’s rate limits. In this project, certificates are persisted on a virtual machine’s storage; the VM is recreated quite rarely, so the rate limits won’t be exceeded.

Also, during the aforementioned phases of your project, it is recommended to use Let’s Encrypt’s staging environment. You can do that by adding ACME URL for the staging environment to Caddyfile as follows:

You can find the Nginx configuration for the project in the commit history. It does the same things as Caddy config does except HTTPS support. If you want to set up Nginx to make a project accessible through HTTPS, it is worth noting mkcert (for local development) and Certbot (for deployment to other environments) tools.

Locally, it is not mandatory to use the reverse proxy: you can also use http://localhost:8091 for book-service and http://localhost:8093 for notification-service.

For the monitoring, I use Kafka UI and pgAdmin running as Docker containers that are accessible at http://localhost:8101 and http://localhost:8102 respectively.

Using Kafka UI, you can manage your Kafka cluster, specifically, observe existing consumers, structure of messages, and existing topics:

Using pgAdmin, you can monitor what is happening in your databases, in particular, the contents of tables:

To save myself the trouble of entering master password, configuring three database connections, and entering passwords for them, I do the following:

Make sure you do not use this approach in your production environment as it is not secure.

Don’t forget to use docker logs <container-name> -f to see what’s going on inside a Docker container.

There are some limitations to functionality when deploying a project to a test environment (see Book service section), so we start with a local environment. For the testing, you need a utility that allows to perform requests to REST API. For that, I use Postman and a Postman collection containing examples of requests to REST API of book-service exposed by Caddy. The collection also contains a couple of requests to Kafka Connect REST API described earlier). URLs in the Postman collection use baseUrl environment variable, for example, {{baseUrl}}/book-service/books; by default, it is set to https://localhost.

After you have launched the project, you can execute requests against REST API of book-service exposed by Caddy and see their results that is notifications displayed in the UI. Let’s start by getting a list of all the books in the library; for that, execute GET {{baseUrl}}/book-service/books:

Similarly, you can get a list of all the authors using GET {{baseUrl}}/book-service/authors:

Let’s see how user notifications delivery works. For demo purposes, I’ll show the delivery through WebSocket. In your browser, navigate to https://localhost which is served by Caddy and routes to the page served by notification-service; press the "Connect" button:

That initiates a connection to our WebSocket/SockJS server.

Now you can add a book to the library by executing POST {{baseUrl}}/book-service/books:

The changes caused by the initial call to the REST API have passed through the chain of several microservices, databases, connectors, and Kafka, and have been eventually delivered to users. Two notifications arrived because there are two users in user-db and by business logic, an event of book creation is "broadcasted" to all users.

You can delete the book by using DELETE {{baseUrl}}/book-service/books/{id}:

This event is also delivered to all users.

When a user borrows a book, POST {{baseUrl}}/book-service/books/{id}/loans should be executed. As said earlier, in book-service, there is a configurable parameter whether to use streaming data of users of the library (see user.check.use-streaming-data parameter in the config). By default, the data are not used, so if a book is lent by a user who obviously does not exist in the database, at first the book loan entity will be successfully created by book-service but eventually the created book loan will be canceled:

This case is explained in Book service and User service sections: such setting of book-service results in user check being performed by user-service; if it determines that a message of type BookLent contains non-existing user id, it will publish RollbackBookLentCommand, and book-service will cancel the book loan. Therefore, the executed request will eventually lead to two events in the inbox table processed by user-service — BookLent and BookLoanCanceled:

Now I change user.check.use-streaming-data to true and restart book-service. The same request is rejected immediately because the check is performed against user_replica table of book database:

The request will only succeed if the user exists and is active:

When a user returns a book to the library, DELETE {{baseUrl}}/book-service/books/{id}/loans/{id} should be executed:

Finally, I switch to the testing environment where I will execute available write methods of the REST API. First, change the base URL to https://eda-demo.romankudryashov.com. If you use Postman, you can either edit the mentioned baseUrl environment variable in the Postman collection or hardcode URLs for REST API methods by replacing {{baseUrl}} with the specified value. Then, navigate to eda-demo.romankudryashov.com in your browser.

A book can be changed by PUT {{baseUrl}}/book-service/books/{id}:

On the testing environment, you can only change publication year of the book.

An author can be changed by PUT {{baseUrl}}/book-service/authors/{id}:

Similarly to the book entity, on the testing environment, you can only change date of birth of an author.

Two notifications received because two users borrowed two books of this author.

You can try to do it yourself here. On this environment, after a periodical project redeployment, it may take some time for user notifications to be delivered, but they should be delivered eventually.

In this project, user-service is the only service that has two instances running. The second instance is needed to demonstrate how inbox messages processing can be parallelized. It is configured with different SPRING_APPLICATION_NAME, user-service-2, that overrides the default value in the service’s config.

To simulate a heavy load on the service, I use this script that inserts 10,000 identical messages that differ only in their id. I expect each service to eventually process about 5000 messages.

Before starting the processing, I slightly change the default values of configuration parameters to decrease the processing time: the scheduled task will run every 3 seconds (instead of 5) and batch.size is 70 (instead of 50). I will observe the processing using the following query:

The query and the script to insert 10,000 events are performed against user database; you can enter the database using docker exec -it user-db bash.

Real time processing looks like this:

That is, it took about 3 minutes and 36 seconds to process 10,000 events by two instances of the service. Or, the speed of message processing by each microservice is 70 messages / 3 seconds ≈ 23 messages per second. I haven’t done any more intensive load testing yet.

You can play around with the processing time by changing the following configurable parameters of the service:

By invalid message, I mean that it is in an unexpected format for some connector. To produce such a message, I will change value.converter* properties in configuration of book.source connector.

First, I change the encoding type used to serialize a message in CloudEvents format and its data field by changing both value.converter.serializer.type and value.converter.data.serializer.type configuration parameters from avro to json and restart the project. Now user.sink can’t process such messages because it expects them as well as their data field to be serialized to Avro. I initiate producing of such a message by calling one of REST API methods of book-service, for example, to update a book; that causes an error in user.sink:

You can see that the message appears in library.events topic as well as in library.events.dlq.ce-json topic:

Eventually, this message is processed by user.sink.dlq-ce-json connector, stored in inbox table, and successfully processed by user-service:

For the second scenario, I change value.converter property of book.source connector to org.apache.kafka.connect.json.JsonConverter, restart the project again, and use the same REST API method to produce an invalid message. Now, in addition to user.sink, user.sink.dlq-ce-json also fails at processing the message:

The message appears in three topics: library.events, library.events.dlq.ce-json, and library.events.dlq.unprocessed:

In the latest topic, headers of a message contain multiple connect.errors.* headers including a stacktrace of an error (connect.errors.exception.stacktrace header) occurred in user.sink.dlq-ce-json connector.

Eventually, this message is processed by user.sink.dlq-unprocessed and stored in inbox_unprocessed:

Messages from this table won’t be processed by user-service and are intended for manual processing. The error column of the table is filled with a value of the mentioned header containing a stacktrace of an error occurred in the previous connector.