Unlocking Change Data Capture with Debezium in Spring Boot – No Extra Middleware Needed

Following the Ockham’s razor principle, small projects often avoid adding unnecessary components such as message queues; however, they still need a way to decouple asynchronous business logic.

Change Data Capture (CDC) solves this by monitoring database change logs—most commonly MySQL binlog—to trigger actions whenever a row is inserted, updated, or deleted, without modifying existing application code.

Common CDC frameworks include Canal, Debezium, Maxwell, and Flink CDC. Among them, Debezium is preferred because it requires only a dependency, avoids extra components, supports multiple databases, integrates with Kafka for reliable delivery, and can run in an embedded mode without a separate Kafka cluster.

Why Choose Debezium

Installing Canal adds an extra entity, violating the “no unnecessary components” rule.

Canal only works with MySQL, limiting its applicability.

Flink CDC, though popular in big‑data, also relies on Debezium under the hood.

Debezium can publish change events to Kafka topics, reducing database read pressure and guaranteeing at‑least‑once semantics.

Embedded mode allows running Debezium without deploying a Kafka cluster, adhering to the minimal‑component principle.

Debezium is an open‑source low‑latency streaming platform for CDC. It captures row‑level changes from the binlog, stores them in a durable, replicated log, and lets consumers process events reliably, handling transaction boundaries and schema evolution automatically.

Typical application scenarios include:

Cache invalidation – instantly refresh or remove cache entries when source data changes.

Simplifying monolithic applications – replace dual‑writes with CDC‑driven asynchronous processing.

Sharing databases – multiple services react to the same database changes without a message bus.

Data integration – build lightweight ETL pipelines by consuming change events.

CQRS – feed read‑model updates from captured change events.

Spring Boot Integration

Dependencies

<debezium.version>1.7.0.Final</debezium.version>
<mysql.connector.version>8.0.26</mysql.connector.version>

<dependency>
    <groupId>mysql</groupId>
    <artifactId>mysql-connector-java</artifactId>
    <version>${mysql.connector.version}</version>
    <scope>runtime</scope>
</dependency>

<dependency>
    <groupId>io.debezium</groupId>
    <artifactId>debezium-api</artifactId>
    <version>${debezium.version}</version>
</dependency>

<dependency>
    <groupId>io.debezium</groupId>
    <artifactId>debezium-embedded</artifactId>
    <version>${debezium.version}</version>
</dependency>

<dependency>
    <groupId>io.debezium</groupId>
    <artifactId>debezium-connector-mysql</artifactId>
    <version>${debezium.version}</version>
    <exclusions>
        <exclusion>
            <groupId>mysql</groupId>
            <artifactId>mysql-connector-java</artifactId>
        </exclusion>
    </exclusions>
</dependency>

Configuration

debezium.datasource.hostname = localhost
debezium.datasource.port = 3306
debezium.datasource.user = root
debezium.datasource.password = 123456
debezium.datasource.tableWhitelist = test.test
debezium.datasource.storageFile = E:/debezium/test/offsets/offset.dat
debezium.datasource.historyFile = E:/debezium/test/history/custom-file-db-history.dat
debezium.datasource.flushInterval = 10000
debezium.datasource.serverId = 1
debezium.datasource.serverName = name-1

Key configuration items: connector.class – set to io.debezium.connector.mysql.MySqlConnector for MySQL. offset.storage – choose FileOffsetBackingStore to store offsets locally (no Kafka needed). offset.storage.file.filename – path of the local offset file. offset.flush.interval.ms – frequency of offset persistence; without Kafka the connector provides at‑least‑once semantics. table.whitelist – monitor only specific tables. database.whitelist – monitor all tables of a database.

Snapshot mode determines how the connector initializes: initial – run a snapshot only when no offset exists. when_needed – run when offsets are missing or invalid. never – never snapshot; start from the beginning of binlog. schema_only – capture only schema changes, not data. schema_only_recovery – recover from a corrupted history topic.

The Debezium service must have a unique database.server.id; duplicate IDs cause errors such as:

io.debezium.DebeziumException: A slave with the same server_uuid/server_id as this slave has connected to the master; ...

Listener Implementation

import com.alibaba.fastjson.JSON;
import io.debezium.config.Configuration;
import io.debezium.data.Envelope;
import io.debezium.engine.ChangeEvent;
import io.debezium.engine.DebeziumEngine;
import io.debezium.engine.format.Json;
import lombok.Builder;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import javax.annotation.Resource;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.Executor;

@Component
@Slf4j
public class MysqlBinlogListener {

    @Resource
    private Executor taskExecutor;

    private final List<DebeziumEngine<ChangeEvent<String, String>>> engineList = new ArrayList<>();

    private MysqlBinlogListener(@Qualifier("mysqlConnector") Configuration configuration) {
        this.engineList.add(DebeziumEngine.create(Json.class)
                .using(configuration.asProperties())
                .notifying(record -> receiveChangeEvent(record.value()))
                .build());
    }

    private void receiveChangeEvent(String value) {
        if (Objects.nonNull(value)) {
            Map<String, Object> payload = getPayload(value);
            String op = JSON.parseObject(JSON.toJSONString(payload.get("op")), String.class);
            if (!StringUtils.isBlank(op) && !Envelope.Operation.READ.equals(op)) {
                ChangeData changeData = getChangeData(payload);
                try {
                    mysqlBinlogService.service(changeData);
                } catch (Exception e) {
                    log.error("binlog processing exception, original data: " + changeData, e);
                }
            }
        }
    }

    @PostConstruct
    private void start() {
        for (DebeziumEngine<ChangeEvent<String, String>> engine : engineList) {
            taskExecutor.execute(engine);
        }
    }

    @PreDestroy
    private void stop() {
        for (DebeziumEngine<ChangeEvent<String, String>> engine : engineList) {
            if (engine != null) {
                try {
                    engine.close();
                } catch (IOException e) {
                    log.error("", e);
                }
            }
        }
    }

    public static Map<String, Object> getPayload(String value) {
        Map<String, Object> map = JSON.parseObject(value, Map.class);
        return JSON.parseObject(JSON.toJSONString(map.get("payload")), Map.class);
    }

    public static ChangeData getChangeData(Map<String, Object> payload) {
        Map<String, Object> source = JSON.parseObject(JSON.toJSONString(payload.get("source")), Map.class);
        return ChangeData.builder()
                .op(payload.get("op").toString())
                .table(source.get("table").toString())
                .after(JSON.parseObject(JSON.toJSONString(payload.get("after")), Map.class))
                .source(JSON.parseObject(JSON.toJSONString(payload.get("source")), Map.class))
                .before(JSON.parseObject(JSON.toJSONString(payload.get("before")), Map.class))
                .build();
    }

    @Data
    @Builder
    public static class ChangeData {
        private Map<String, Object> after;
        private Map<String, Object> source;
        private Map<String, Object> before;
        private String table;
        private String op;
    }
}

Test outputs illustrate how the ChangeData object contains the table name, operation type (c‑create, u‑update, d‑delete), and before/after row data.

The offset file ( storageFile) records the last processed binlog position, similar to Kafka offsets; restarting the service resumes from this position.

Conclusion

The article demonstrates that Debezium provides a lightweight CDC solution for small projects, allowing them to achieve some of the benefits of a message queue without introducing heavyweight infrastructure. While it cannot fully replace a dedicated message system, it offers a pragmatic approach to decouple services, keep caches consistent, and build simple data‑integration pipelines.