ApsaraMQ for RocketMQ:Stream MySQL change data to ApsaraMQ for RocketMQ with Canal

Version compatibility

Enable binary logging

ApsaraDB RDS for MySQL

Binary logging is enabled by default, and your Alibaba Cloud account automatically has the required dump permissions. Skip to Create a test database.

Self-managed MySQL

1. Edit the MySQL configuration file (my.cnf) to enable ROW-format binary logging:

       [mysqld]
       log-bin=mysql-bin   # Enable binary logging
       binlog-format=ROW   # Required format for Canal
       server_id=1         # Must differ from the Canal secondary node ID

2. Create a dedicated MySQL user for Canal with replication permissions:

       CREATE USER canal IDENTIFIED BY 'canal';
       GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'canal'@'%';
       FLUSH PRIVILEGES;

Create a test database

CREATE DATABASE canal DEFAULT CHARACTER SET utf8mb4 COLLATE utf8mb4_general_ci;

Create a test table

Create a students table in the canal database:

CREATE TABLE students (
    id INT AUTO_INCREMENT,
    name VARCHAR(100) NOT NULL,
    age INT,
    gender VARCHAR(10),
    PRIMARY KEY (id)
);

Install JDK

sudo yum install java-1.8.0-openjdk

Download and extract Canal Deployer

# Download Canal Deployer 1.1.6
sudo wget https://github.com/alibaba/canal/releases/download/canal-1.1.6/canal.deployer-1.1.6.tar.gz

# Create the installation directory and extract
sudo mkdir -p /usr/local/canal-server
sudo tar -zxvf canal.deployer-1.1.6.tar.gz -C /usr/local/canal-server

Configure Canal for RocketMQ

Two configuration files control Canal behavior: canal.properties for global settings and instance.properties for per-instance settings.

sudo vi /usr/local/canal-server/conf/canal.properties

# Output destination: send change events to RocketMQ
canal.serverMode = rocketMQ

# Alibaba Cloud authentication
canal.aliyun.accessKey = <your-access-key-id>
canal.aliyun.secretKey = <your-access-key-secret>

# Access channel: set to "cloud" for Alibaba Cloud managed RocketMQ
canal.mq.accessChannel = cloud

# Message format: must be false for ApsaraMQ for RocketMQ
canal.mq.flatMessage = false

# RocketMQ producer settings
rocketmq.producer.group = canal_test
rocketmq.enable.message.trace = false
rocketmq.customized.trace.topic =
rocketmq.namespace =
rocketmq.namesrv.addr = <your-rocketmq-endpoint>
rocketmq.retry.times.when.send.failed = 0
rocketmq.vip.channel.enabled = false
rocketmq.tag =

sudo vi /usr/local/canal-server/conf/example/instance.properties

# MySQL connection
canal.instance.master.address = <your-mysql-endpoint>:3306
canal.instance.dbUsername = <your-db-username>
canal.instance.dbPassword = <your-db-password>

# Table filter: capture all tables in the "canal" schema
canal.instance.filter.regex = canal\\..*

# Target RocketMQ topic
canal.mq.topic = canal_topic

Start Canal Deployer

/usr/local/canal-server/bin/startup.sh

Verify the startup

Check the Canal server log to confirm a successful start:

sudo cat /usr/local/canal-server/logs/canal/canal.log

A successful startup produces output similar to the following:

2024-07-15 17:24:12.154 [main] INFO  com.alibaba.otter.canal.deployer.CanalLauncher - ## set default uncaught exception handler
2024-07-15 17:24:12.202 [main] INFO  com.alibaba.otter.canal.deployer.CanalLauncher - ## load canal configurations
2024-07-15 17:24:12.497 [main] INFO  c.a.o.c.c.rocketmq.producer.CanalRocketMQProducer - ##Start RocketMQ producer##
2024-07-15 17:24:12.799 [main] INFO  com.alibaba.otter.canal.deployer.CanalStarter - ## start the canal server.
2024-07-15 17:24:12.984 [main] INFO  com.alibaba.otter.canal.deployer.CanalController - ## start the canal server[172.17.XX.XX:11111]
2024-07-15 17:24:16.208 [main] INFO  com.alibaba.otter.canal.deployer.CanalStarter - ## the canal server is running now ......

The final the canal server is running now line confirms that Canal started successfully and connected to RocketMQ.

Next, check the instance log to confirm Canal connected to MySQL and started monitoring:

sudo cat /usr/local/canal-server/logs/example/example.log

Look for these key indicators of a successful connection:

2024-07-15 18:22:15.667 [main] INFO  c.a.otter.canal.instance.spring.CanalInstanceWithSpring - start CannalInstance for 1-example
2024-07-15 18:22:15.699 [main] WARN  c.a.o.canal.parse.inbound.mysql.dbsync.LogEventConvert - --> init table filter : ^canal\..*$
2024-07-15 18:22:15.699 [main] WARN  c.a.o.canal.parse.inbound.mysql.dbsync.LogEventConvert - --> init table black filter : ^mysql\.slave_.*$

If the instance log shows errors instead, check the following common causes:

Insert test data

Insert a row into the students table to trigger a change event:

INSERT INTO `students` (`name`, `age`, `gender`) VALUES ('Tome', 18, 'male');

Check the message in ApsaraMQ for RocketMQ

1. Log on to the ApsaraMQ for RocketMQ console.

2. In the left-side navigation pane, click Instances.

3. On the Instances page, click the name of the instance you configured.

4. In the left-side navigation pane of the Instance Details page, click Message Query.

5. Verify that a message from Canal appears in the query results.

If no message appears, wait a few seconds for Canal to process the binary log event and retry the query. If messages still do not appear after 30 seconds, review the Canal instance log at /usr/local/canal-server/logs/example/example.log for errors.

canal.properties parameters

instance.properties parameters