ApsaraMQ for Kafka:Synchronize MySQL data to ApsaraMQ for Kafka by using Canal

instance.properties parameter reference

Internet access (SASL_SSL)

When connecting over the Internet, Canal authenticates with ApsaraMQ for Kafka through SASL_SSL. You need to modify three files.

File 1: conf/canal.properties

# Set Canal to use Kafka as the output
canal.serverMode = kafka

# SSL endpoint of your ApsaraMQ for Kafka instance
# Get this from the Endpoint Information section on the Instance Details page
kafka.bootstrap.servers = alikafka-pre-cn-zv**********-1.alikafka.aliyuncs.com:9093,alikafka-pre-cn-zv**********-2.alikafka.aliyuncs.com:9093,alikafka-pre-cn-zv**********-3.alikafka.aliyuncs.com:9093

# Kafka producer settings (adjust as needed)
kafka.acks = all
kafka.compression.type = none
kafka.batch.size = 16384
kafka.linger.ms = 1
kafka.max.request.size = 1048576
kafka.buffer.memory = 33554432
kafka.max.in.flight.requests.per.connection = 1
kafka.retries = 0

# SASL_SSL authentication
kafka.ssl.truststore.location = ../conf/kafka_client_truststore_jks
kafka.ssl.truststore.password = KafkaOnsClient
kafka.security.protocol = SASL_SSL
kafka.sasl.mechanism = PLAIN
kafka.ssl.endpoint.identification.algorithm =

File 2: bin/startup.sh

Add the JAAS configuration path to the JAVA_OPTS variable:

JAVA_OPTS=" $JAVA_OPTS -Djava.awt.headless=true -Djava.net.preferIPv4Stack=true -Dfile.encoding=UTF-8 -Djava.security.auth.login.config=/home/doc/tools/canal.deployer-1.1.5/conf/kafka_client_jaas.conf"

File 3: conf/kafka_client_producer_jaas.conf

Specify the SASL credentials for your ApsaraMQ for Kafka instance:

KafkaClient {
  org.apache.kafka.common.security.plain.PlainLoginModule required
  username="<your-sasl-username>"
  password="<your-sasl-password>";
};

Replace the placeholders with your actual values:

Download the SSL truststore file kafka.client.truststore.jks and save it to the conf/ directory.

VPC access (PLAINTEXT)

When connecting from within a VPC, no authentication or encryption is required. Set canal.serverMode and kafka.bootstrap.servers in conf/canal.properties:

# Set Canal to use Kafka as the output
canal.serverMode = kafka

# Default (VPC) endpoint of your ApsaraMQ for Kafka instance
# Get this from the Endpoint Information section on the Instance Details page
kafka.bootstrap.servers = alikafka-pre-cn-zv**********-1-vpc.alikafka.aliyuncs.com:9092,alikafka-pre-cn-zv**********-2-vpc.alikafka.aliyuncs.com:9092,alikafka-pre-cn-zv**********-3-vpc.alikafka.aliyuncs.com:9092

# Kafka producer settings (adjust as needed)
kafka.acks = all
kafka.compression.type = none
kafka.batch.size = 16384
kafka.linger.ms = 1
kafka.max.request.size = 1048576
kafka.buffer.memory = 33554432
kafka.max.in.flight.requests.per.connection = 1
kafka.retries = 0

canal.properties parameter reference

Verify that Canal started

Check the Canal server log:

tail -f logs/canal/canal.log

Expected output:

2013-02-05 22:45:27.967 [main] INFO  com.alibaba.otter.canal.deployer.CanalLauncher - ## start the canal server.
2013-02-05 22:45:28.113 [main] INFO  com.alibaba.otter.canal.deployer.CanalController - ## start the canal server[10.1.XX.XX:11111]
2013-02-05 22:45:28.210 [main] INFO  com.alibaba.otter.canal.deployer.CanalLauncher - ## the canal server is running now ......

Check the Canal instance log:

tail -f logs/example/example.log

Expected output:

2013-02-05 22:50:45.636 [main] INFO  c.a.o.c.i.spring.support.PropertyPlaceholderConfigurer - Loading properties file from class path resource [canal.properties]
2013-02-05 22:50:45.641 [main] INFO  c.a.o.c.i.spring.support.PropertyPlaceholderConfigurer - Loading properties file from class path resource [example/instance.properties]
2013-02-05 22:50:45.803 [main] INFO  c.a.otter.canal.instance.spring.CanalInstanceWithSpring - start CannalInstance for 1-example
2013-02-05 22:50:45.810 [main] INFO  c.a.otter.canal.instance.spring.CanalInstanceWithSpring - start successful....

Test with a sample table

1. In a MySQL database named mysql, create a test table and insert sample data:

       mysql> SELECT * FROM T_Student;
       +--------+---------+------+------+
       | stuNum | stuName | age  | sex  |
       +--------+---------+------+------+
       |      1 | Wang    |   18 | girl |
       |      2 | Zhang   |   17 | boy  |
       +--------+---------+------+------+
       2 rows in set (0.00 sec)

2. Confirm that Canal captured the changes by checking the meta log. Each INSERT, UPDATE, or DELETE generates a new entry: Expected output: Each line shows the binary log file (log.000001), the byte offset (for example, 29723), and the source MySQL address -- confirming that Canal is reading changes.

       tail -f logs/example/meta.log

       2020-07-29 09:21:05.110 - clientId:1001 cursor:[log.000001,29723,1591190230000,1,] address[/192.168.XX.XX:3306]
       2020-07-29 09:23:46.109 - clientId:1001 cursor:[log.000001,30047,1595985825000,1,] address[localhost/192.168.XX.XX:3306]

3. Verify that messages arrived in ApsaraMQ for Kafka. Open the ApsaraMQ for Kafka console and query messages for your topic. For more information, see Query messages.

   

4. After you finish testing, stop Canal:

       sh bin/stop.sh