How to Use Canal to Monitor Table Schema Changes

Canal — a CDC framework built on MySQL binlog — offers native support for detecting not only DML (INSERT/UPDATE/DELETE) but also DDL (CREATE/ALTER/DROP) events. Capturing schema changes in real time is essential for any system that streams data downstream, because unnoticed column additions, type modifications, or index adjustments can cause parsing failures, corrupt downstream stores, or break hard‑coded business logic. This article explains how Canal listens to table‑structure changes, walks through the necessary configuration, shows concrete client implementations (Java and Go), and demonstrates a full pipeline that pushes DDL events to Kafka.
Canal’s architecture consists of three core components:

1. Canal Server – connects to MySQL, reads the binary log, and translates raw events into a unified Entry protocol.
2. Canal Client SDKs – Java, Go, Python, etc., pull Entry objects from the server and apply custom logic.
3. Instance – a logical unit inside the server that isolates a specific database or set of tables. Each instance can be tuned with filter rules (filter.regex) and mode settings.

When MySQL emits a DDL statement, it appears in the binlog as a QUERY event. Canal converts this event into an Entry whose EntryType is ROWDATA and whose RowChange payload has isDdl = true. By checking RowChange.isDdl() on the client side, you can instantly recognise a schema‑change event.

Enabling DDL Capture

Server‑side Settings

Add the following lines to the instance’s instance.properties file:

# Monitor all tables in the `mydb` database
canal.instance.filter.regex = mydb\\..*

# Turn on DDL parsing – the server will emit RowChange entries for DDL
canal.instance.filter.query.dml = true

# Optional: push only DDL events (set to false to receive DML as well)
canal.instance.filter.query.only = false

• filter.regex limits the scope of monitoring, preventing noise from unrelated databases.
• filter.query.dml = true tells Canal to treat DDL as regular entries rather than discarding them.
• filter.query.only is handy for a dedicated metadata‑sync service that cares solely about schema changes.

Client‑side Detection (Java)

import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.client.CanalConnectors;
import com.alibaba.otter.canal.protocol.CanalEntry.*;
import com.alibaba.otter.canal.protocol.Message;

import java.net.InetSocketAddress;
import java.util.List;

public class DdlListener {
    public static void main(String[] args) {
        // 1️⃣ Create a connector to the Canal server
        CanalConnector connector = CanalConnectors.newSingleConnector(
                new InetSocketAddress("127.0.0.1", 11111),
                "example",   // instance name defined in instance.properties
                "",          // username (empty for default)
                ""           // password
        );

        int batchSize = 1000;
        connector.connect();
        connector.subscribe(".*\\..*");   // subscribe to all schemas/tables
        connector.rollback();

        while (true) {
            // 2️⃣ Pull a batch of entries
            Message message = connector.getWithoutAck(batchSize);
            long batchId = message.getId();
            int size = message.getEntries().size();

            if (batchId == -1 || size == 0) {
                try { Thread.sleep(100); } catch (InterruptedException ignored) {}
                continue;
            }

            // 3️⃣ Iterate over entries
            for (Entry entry : message.getEntries()) {
                if (entry.getEntryType() != EntryType.ROWDATA) {
                    continue;   // skip transaction markers, heartbeat, etc.
                }

                RowChange rowChange;
                try {
                    rowChange = RowChange.parseFrom(entry.getStoreValue());
                } catch (Exception e) {
                    throw new RuntimeException("Failed to parse RowChange", e);
                }

                // 4️⃣ Detect DDL
                if (rowChange.getIsDdl()) {
                    String sql = rowChange.getSql();                     // original DDL statement
                    String schema = entry.getHeader().getSchemaName();   // affected database
                    String table = entry.getHeader().getTableName();     // affected table
                    System.out.printf("[DDL] %s.%s : %s%n", schema, table, sql);

                    // 5️⃣ Optional: further parsing (e.g., extract column list) or
                    //      persist the DDL to a metadata store.
                }
            }

            // 6️⃣ Acknowledge the batch
            connector.ack(batchId);
        }
    }
}

Key points

• RowChange.getIsDdl() is the flag that distinguishes DDL from ordinary row changes.
• RowChange.getSql() returns the raw DDL string, which can be logged, audited, or re‑executed on a downstream system.
• Header fields (schemaName, tableName) give you the exact target of the change without parsing the SQL yourself.

Client‑side Detection (Go)

package main

import (
    "fmt"
    canal "github.com/alibaba/canal-go/client"
    "github.com/alibaba/canal-go/protocol"
)

func main() {
    conn := canal.NewSimpleCanalConnector("127.0.0.1:11111", "example", "", "")
    conn.Connect()
    defer conn.Disconnect()

    conn.Subscribe(".*\\..*")
    for {
        msg, err := conn.Get(1000)
        if err != nil || msg.Id == -1 {
            continue
        }
        for _, entry := range msg.Entries {
            if entry.EntryType != protocol.EntryType_ROWDATA {
                continue
            }
            rowChange, _ := protocol.RowChangeParse(entry.StoreValue)
            if rowChange.IsDdl {
                fmt.Printf("[DDL] %s.%s : %s\n",
                    entry.Header.SchemaName,
                    entry.Header.TableName,
                    rowChange.Sql)
            }
        }
        conn.Ack(msg.Id)
    }
}

Both snippets illustrate the same workflow: connect → subscribe → fetch → filter ROWDATA → check isDdl → act.

End‑to‑End Example: Broadcasting DDL to Kafka

Below is a practical pipeline that captures DDL events with Canal and publishes them to a Kafka topic named ddl-changes. Downstream services can then replay the DDL on other databases, update schema registries, or trigger alerting workflows.

1️⃣ Infrastructure Overview

2️⃣ Docker‑Compose Skeleton

version: '3'
services:
  mysql:
    image: mysql:8.0
    environment:
      MYSQL_ROOT_PASSWORD: root
    command: >
      --default-authentication-plugin=mysql_native_password
      --log-bin=mysql-bin
      --binlog-format=ROW
    ports:
      - "3306:3306"

  canal:
    image: canal/canal-server
    depends_on:
      - mysql
    environment:
      - canal.instance.master.address=mysql:3306
      - canal.instance.dbUsername=root
      - canal.instance.dbPassword=root
      - canal.instance.filter.regex=mydb\\..*
      - canal.instance.filter.query.dml=true
    ports:
      - "11111:11111"

Running docker-compose up -d brings up a MySQL instance with binlog enabled and a Canal server already listening for changes in the mydb schema.

3️⃣ Kafka Producer (Java)

import org.apache.kafka.clients.producer.*;
import java.util.Properties;

public class DdlKafkaProducer {
    private final Producer<String, String> producer;
    private final String topic = "ddl-changes";

    public DdlKafkaProducer() {
        Properties props = new Properties();
        props.put("bootstrap.servers", "localhost:9092");
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        // Optional tuning
        props.put("linger.ms", "5");
        props.put("batch.size", "16384");
        producer = new KafkaProducer<>(props);
    }

    public void send(String schema, String table, String sql) {
        String key = schema + "." + table;
        ProducerRecord<String, String> record = new ProducerRecord<>(topic, key, sql);
        producer.send(record, (metadata, exception) -> {
            if (exception != null) {
                exception.printStackTrace();
            } else {
                System.out.printf("DDL sent → %s-%d offset %d%n",
                        metadata.topic(), metadata.partition(), metadata.offset());
            }
        });
    }

    public void close() {
        producer.close();
    }
}

Integrate the producer into the earlier DdlListener:

DdlKafkaProducer kafkaProducer = new DdlKafkaProducer();
...
if (rowChange.getIsDdl()) {
    String sql = rowChange.getSql();
    String schema = entry.getHeader().getSchemaName();
    String table = entry.getHeader().getTableName();
    kafkaProducer.send(schema, table, sql);
}
...
kafkaProducer.close();

4️⃣ Consuming the DDL Stream (Python)

from kafka import KafkaConsumer

consumer = KafkaConsumer(
    'ddl-changes',
    bootstrap_servers='localhost:9092',
    group_id='ddl-sync',
    value_deserializer=lambda m: m.decode('utf-8')
)

for msg in consumer:
    key = msg.key.decode('utf-8')   # schema.table
    sql = msg.value
    print(f"[{key}] DDL: {sql}")
    # Example: execute the DDL on a replica database
    # replica_cursor.execute(sql)

With this setup, any ALTER TABLE, CREATE INDEX, or DROP COLUMN executed on the source MySQL instance instantly appears on the ddl-changes topic, ready for downstream consumption.

Common Pitfalls & Tuning Tips

Performance Nuggets

• Granular Filtering – The more specific the filter.regex, the fewer irrelevant events Canal has to parse and forward.
• Batch Size – A batchSize of 500‑2000 offers a good trade‑off between latency and network overhead.
• Retry Logic – Wrap Kafka sends and MySQL replays in retry loops with exponential back‑off to survive transient failures.
• Metadata Persistence – Persist captured DDL into a dedicated metadata store (e.g., a separate MySQL schema or a version‑controlled file) for audit trails and offline schema reconstruction.

Recap

1. Activate DDL parsing on the Canal server via filter.query.dml.
2. Consume entries on the client side and use RowChange.getIsDdl() to separate schema changes from row changes.
3. Extract the raw SQL (RowChange.getSql()) and context (schemaName, tableName).
4. Push the DDL to downstream systems such as Kafka, where it can be replayed, logged, or used to keep multiple databases in sync.
5. Address operational concerns—filtering, batch sizing, error handling, and idempotency—to keep the pipeline robust in production.

By mastering these steps, you can build a CDC pipeline that not only streams data changes but also gracefully handles the evolution of the underlying schema, ensuring downstream services stay in lockstep with the source database’s structure.