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 the 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 the Connectors configuration section. All the fields are mapped to the appropriate columns of the outbox table:

The save method stores OutboxMessageEntity through OutboxMessageRepository in the 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 WAL). 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.

In the next section, we will look at another implementation of the Transactional outbox pattern, which stores outbox directly in the WAL, that is, without using the outbox table. We will also look at the advantages and disadvantages of each implementation.

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 into 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 the 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 the 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 to process them. If the microservice processes the messages in the same transaction, it can only process them in the sequential manner because it is technically difficult in Spring to process 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 following algorithm of messages processing:

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:

The configuration of transaction management specifies that:

I wrapped incomingEventService.process method call in try/catch to prevent possibility of endless number of attempts to process an inbox message; otherwise, in case of an exception in that method, the status of the inbox message would never change and always be ReadyForProcessing. The considered implementation makes only one attempt to process a message.

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

The configuration of transaction management specifies that the original transaction started in InboxMessageService.process() method should be suspended; thus, any downstream error won’t affect the original transaction.

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:

Saving an outbox message is implemented in a separate transaction and looks like this:

The considered implementation of the Transactional outbox pattern has two changes compared to the one considered in Book service section:

To send an outbox message directly to the WAL, I implemented the following repository:

This and other possible applications of pg_logical_emit_message() function are considered in this blog post. We will discuss WAL and how it works in detail in the Persistence layer section.

Both approaches to implementing the Transactional outbox pattern result in a message appearing in the WAL, so both approaches can be used in conjunction with a Debezium source connector. Let’s look at the main differences between them:

If you use the latter approach, you need to define a common message format for all the emitted events, since there is no table schema that a message must conform to. You can, for example, derive that format from a set of outbox table columns. I implement OutboxMessage DTO to have a structure similar to the outbox table considered in the previous section:

The DTO has an additional topic field 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 into inbox table in the same order in which they were inserted into 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 the 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 the 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 thc tool looks only at the HTTP status: a service is considered OK when it is >= 200 and < 300. The health-check command will be used by Docker containers of the services as follows:

THC_PATH configures a path for health checks.

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 into 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. 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; by default, in-memory storage is used. 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 is accessible at http://localhost:8080 in your browser; to explore artifacts, you can use Apicurio Registry UI accessible at http://localhost:8103. If you initiate several domain events through REST API of book-service, the page with the list of existing artifacts will look like this (the screenshots below are from Apicurio Registry 2.x version; currently, Apicurio Registry 3.x version and its UI are used in the project):

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 the 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: