Using Maxwell, we read data changes from the primary database and put them on RabbitMQ, and finally build an indexer service that processes the messages placed on RabbitMQ and indexes the data on Typesense.
In this part, we use a small prebuilt blog management service and implement fast search on this project using CDC.
Initial Overview
Our service is a simple blog that gives us the ability to create, delete, and edit content. This project uses Golang as the backend language and MySQL as the primary database. Using CDC, we can read data changes from the blog table and store these changes in the typesense database.
You can see and download the project's final source code on GitHub. To run the project, first go to the project folder using the terminal, and then run the project with Docker using the command below:
docker compose up -d
And finally, you can shut down the project using the command below:
docker compose down --remove-orphans
If you're not familiar with Docker or haven't installed it, it's best to install Docker first, and this will also be a good opportunity to get a bit familiar with Docker.
And if you have make on your system, you can use the commands below inside the project folder to start up and shut down the services:
// start the project
make up
// shut down the services
make down
Project Architecture
Before we start implementing, let's look at the final structure of what we want to have.
Overall, we'll use the following tools to build our project:
We use MySQL as the primary database for storing and maintaining content.
We use the Typesense database to full-text index the content for use in search. This way, our application can search the desired content very quickly and provide the results to our application's users.
We use RabbitMQ as a queue. This way, we put changes made on the primary database into this queue, and then, by processing the messages in the queue, we update the data on Typesense.
This way, we'll index all the data on Typesense instantly, without putting any load on our primary database.
We use Maxwell to read MySQL's logs and turn those logs into messages on RabbitMQ. This way, the changes we make to data in MySQL are stored as messages inside RabbitMQ.
We build an indexer application that reads the messages written to RabbitMQ and, based on these messages, indexes and updates the data on Typesense.
Our main application, whose responsibilities are split into two main parts:
Dashboard-related functionality for managing blog posts, which stores and updates data on the primary database (MySQL, in this case).
General functionality, including the home page, post detail pages, and searching content, which reads and displays data from Typesense.
The overall shape of our application will be as follows:
The project's final architecture
Setting Up Maxwell on MySQL
To set up Maxwell on MySQL, according to the documentation, we need to go through the following steps:
Enabling the binlog in MySQL and setting the log format to Row
Creating the permissions Maxwell needs.
Enabling the settings for the initial data load (bootstrap).
Running maxwell and the bootstrapper
Enabling the Binlog in MySQL and Setting the Log Format to Row
To do this, we can create a file with a conf extension at /etc/mysql/conf.d/. For example, I'll create this file named cdc.conf at /etc/mysql/conf.d/cdc.conf.
Then we put the following settings inside it:
[mysqld]
server_id=1
binlog_format=row
log-bin=mysql-bin
# Only log `blog` DB
# Remove this line if you want to log everything.
# Or add a new line like this if you want to also log another specific DB
binlog_do_db=blog
In the settings above, we set the log format to row and enabled logging using log-bin. Whenever we use the log-bin directive, we also need to assign an ID to the MySQL server. Here, we assigned the number 1 as the ID for this MySQL server. If you have multiple servers or replicas, you should use different IDs.
In the last line, we specified that logging should be enabled only for the blog database. By default, using log-bin enables logging for every database. Using bin_do_db, we can restrict logging to just the databases we want. You can repeat this line once per database.
By removing every line that contains bin_do_db, we can enable logging for every database.
Quite often, these settings are already enabled by default, and we don't need to configure them again! To check whether they're enabled or not, you can run the following queries on your current database and check the result:
SHOW VARIABLES LIKE 'binlog_format'; -- should be ROW
SHOW VARIABLES LIKE 'server_id'; -- should show a number
SHOW VARIABLES LIKE 'log_bin'; -- should be ON
Creating the Permissions Maxwell Needs
To do this, you need to connect to your database and run the following commands inside it:
-- Create Maxwell user for CDC
CREATE USER IF NOT EXISTS 'maxwell'@'%' IDENTIFIED BY 'maxwell123';
-- Grant Maxwell database permissions (for storing state)
GRANT ALL ON maxwell.* TO 'maxwell'@'%';
-- Grant replication permissions to all database tables for CDC
GRANT SELECT, REPLICATION CLIENT, REPLICATION SLAVE ON *.* TO 'maxwell'@'%';
-- Maxwell schema (required for Maxwell to track its state)
CREATE DATABASE IF NOT EXISTS maxwell;
-- Flush privileges
FLUSH PRIVILEGES;
In the first line, you can put your own password instead of maxwell123. Running the commands above creates a user named maxwell as well as a database named maxwell. This user will have full access to the maxwell database. We also grant this user read and replication access on every database. If you want maxwell to have access only to the blog database, you can use blog.* instead of *.*.
What Is the Initial Data Load, or Bootstrap?
After Maxwell is enabled, this tool detects changes that happen in the database (such as INSERT, UPDATE, and DELETE) and produces each change as a message (usually in JSON format).
These messages are then sent to RabbitMQ so other services can use them. However, by default, Maxwell only publishes new changes from the moment it's enabled onward. Because of this, data that already existed in the database beforehand isn't sent to RabbitMQ. To cover this case, there's a feature called Bootstrap. Using Bootstrap, you can also extract the existing data in selected tables and send it to RabbitMQ just like new changes.
Here, we use this feature to index, on Typesense, the data that existed before Maxwell was enabled.
Running Maxwell and the Bootstrapper
After completing the configuration, we can run maxwell and its bootstrapper separately:
Above, we ran two separate commands. The first command runs maxwell as a long-running process, meaning this service keeps running continuously and sends database changes to RabbitMQ.
The second command is maxwell-bootstrap, and running it creates a record in the bootstrap table, which causes maxwell to start the initial data load. You can also do this manually yourself, in which case you won't need the maxwell-bootstrap command.
The record created in the bootstrap table
Note: If you're using the project created in our GitHub repo, after running the project (how to run it is mentioned at the start of this post), you can go to http://localhost:8081/ to see the PHPMyAdmin dashboard and use it to inspect the database structure.
After running bootstrap and maxwell, if we have data in the current database, or if we make changes to the data, we should be able to see the messages that were sent in the RabbitMQ dashboard, under the exchange named maxwell.
In the chart above, we can see that in a short time window, 60 messages per second were sent to RabbitMQ. Next, we'll build a service that processes these messages and indexes the result on Typesense.
Note: If you're using the project created in our GitHub repo, after running the project you can go to http://localhost:15672/ and log in with the username admin and the password admin123.
Building the Indexer Service
This service is responsible for processing the messages that Maxwell writes onto RabbitMQ. Processing each message consists of creating, editing, or deleting an index entry.
If the user creates new content, that content is placed as a message on RabbitMQ, and then the indexer service processes this message and creates a new index entry on Typesense.
If the user edits existing content, that edit is placed as a message on RabbitMQ, and then the indexer service processes this message and updates the current indexed content on Typesense.
And finally, if content is deleted, by processing the message placed on RabbitMQ, the index entry corresponding to that message is removed from typesense.
// StartCDC starts the CDC pipeline
func (s *CDCService) StartCDC(ctx context.Context, queueName string) error {
log.Printf("Starting CDC service, listening to queue: %s", queueName)
// Connect to RabbitMQ
if err := s.messageQueue.Connect(); err != nil {
return fmt.Errorf("failed to connect to RabbitMQ: %w", err)
}
defer s.messageQueue.Close()
// Connect to Typesense√
if err := s.searchIndex.Connect(); err != nil {
return fmt.Errorf("failed to connect to Typesense: %w", err)
}
defer s.searchIndex.Close()
// Ensure the posts collection exists
if err := s.ensurePostsCollection(); err != nil {
return fmt.Errorf("failed to ensure posts collection: %w", err)
}
// Start consuming messages
return s.messageQueue.ConsumeMessages(queueName, s.handleMessage)
}
// handleMessage processes individual messages from RabbitMQ
func (s *CDCService) handleMessage(message []byte) error {
// Parse the CDC event
event, err := domain.FromJSON(message)
if err != nil {
log.Printf("Failed to parse CDC event: %v", err)
return err
}
// Validate the event
if !event.IsValid() {
log.Printf("Invalid CDC event: %+v", event)
return fmt.Errorf("invalid CDC event")
}
// Only process posts table events
if event.Table != "posts" {
log.Printf("Skipping non-posts table event: %s.%s", event.Database, event.Table)
return nil
}
// Process the event based on its type
switch event.Type {
case domain.EventTypeInsert, domain.EventTypeUpdate, domain.EventTypeBootstrapInsert:
return s.handleUpsert(event)
case domain.EventTypeDelete:
return s.handleDelete(event)
case domain.EventTypeBootstrapStart:
log.Printf("Bootstrap insert event received")
return nil
case domain.EventTypeBootstrapComplete:
log.Printf("Bootstrap complete event received")
return nil
default:
log.Printf("Unknown event type: %s", event.Type)
return fmt.Errorf("unknown event type: %s", event.Type)
}
}
Based on the code above, depending on the type of message placed on RabbitMQ, we create, edit, or delete the index entry in Typesense. You can see the complete code on GitHub.
By running the indexer service, the data should get indexed in Typesense, and you can see the result in the typesense dashboard.
The typesense dashboard for viewing indexed data.
For example, in the image we can see that 290 items have been indexed in the posts collection.
Then, our main application can search through the indexed data and show the results to users:
Note: If you're using the project created in our GitHub repo, after running the project you can go to http://localhost:8085/ to see the blog service.
In the image above, whatever we search for in the search box, if it exists, the list of related posts will be shown to us.