RocketMQ Connect in Action 2

PostgreSQL Source(CDC) - >RocketMQ Connect -> MySQL Sink(JDBC)

Preparation

Start RocketMQ

  1. Linux/Unix/Mac
  2. 64bit JDK 1.8+;
  3. Maven 3.2.x+;
  4. Start RocketMQ;

tips : ${ROCKETMQ_HOME} locational instructions

bin-release.zip version：/rocketmq-all-4.9.4-bin-release

source-release.zip version：/rocketmq-all-4.9.4-source-release/distribution

Start Connect

Compiling Connector Plugin

Debezium RocketMQ Connector

  1. $ cd rocketmq-connect/connectors/rocketmq-connect-debezium/
  2. $ mvn clean package -Dmaven.test.skip=true

Move the compiled Debezium PostgreSQL RocketMQ Connector package into the Runtime loading directory. The command is as follows ：

  1. mkdir -p /usr/local/connector-plugins
  2. cp rocketmq-connect-debezium-postgresql/target/rocketmq-connect-debezium-postgresql-0.0.1-SNAPSHOT-jar-with-dependencies.jar /usr/local/connector-plugins

JDBC Connector

Move the compiled JDBC Connector package into the Runtime loading directory. The command is as follows：

  1. $ cd rocketmq-connect/connectors/rocketmq-connect-jdbc/
  2. $ mvn clean package -Dmaven.test.skip=true
  3. cp rocketmq-connect-jdbc/target/rocketmq-connect-jdbc-0.0.1-SNAPSHOT-jar-with-dependencies.jar /usr/local/connector-plugins

Start Connect Runtime

  1. cd  rocketmq-connect
  3. mvn -Prelease-connect -DskipTests clean install -U

Modify the configuration connect-standalone.conf, the main configuration is as follows

  1. $ cd distribution/target/rocketmq-connect-0.0.1-SNAPSHOT/rocketmq-connect-0.0.1-SNAPSHOT
  2. $ vim conf/connect-standalone.conf
  1. $ cd distribution/target/rocketmq-connect-0.0.1-SNAPSHOT/rocketmq-connect-0.0.1-SNAPSHOT
  2. $ vim conf/connect-standalone.conf
  1. workerId=standalone-worker
  2. storePathRootDir=/tmp/storeRoot
  4. ## Http port for user to access REST API
  5. httpPort=8082
  7. # Rocketmq namesrvAddr
  8. namesrvAddr=localhost:9876
  10. # RocketMQ acl
  11. aclEnable=false
  12. accessKey=rocketmq
  13. secretKey=12345678
  15. autoCreateGroupEnable=false
  16. clusterName="DefaultCluster"
  18. # Core configuration, configure the plugin directory of the previously compiled debezium package here
  19. # Source or sink connector jar file dir,The default value is rocketmq-connect-sample
  20. pluginPaths=/usr/local/connector-plugins
  1. cd distribution/target/rocketmq-connect-0.0.1-SNAPSHOT/rocketmq-connect-0.0.1-SNAPSHOT
  3. sh bin/connect-standalone.sh -c conf/connect-standalone.conf &

Postgres image

Use debezium’s Postgres docker environment to set up the Postgres database

  1. # starting a pg instance
  2. docker run -d --name postgres -p 5432:5432 -e POSTGRES_USER=start_data_engineer -e POSTGRES_PASSWORD=password debezium/postgres:14
  4. # bash into postgres instance
  5. docker exec -ti postgres /bin/bash

Postgres information Port：5432 Aaccount：start_data_engineer/password Synchronize original database：bank.holding Target database table：bank1.holding

MySQL image

Use debezium’s MySQL docker environment to set up the MySQL database.

  1. docker run -it --rm --name mysql -p 3306:3306 -e MYSQL_ROOT_PASSWORD=debezium -e MYSQL_USER=mysqluser -e MYSQL_PASSWORD=mysqlpw quay.io/debezium/example-mysql:1.9

MySQL information

Port：3306

Account：root/debezium

Test data

Log in to the database with the start_data_engineer/password account

Source database table：bank.holding

  1. CREATE SCHEMA bank;
  2. SET search_path TO bank,public;
  3. CREATE TABLE bank.holding (
  4.                               holding_id int,
  5.                               user_id int,
  6.                               holding_stock varchar(8),
  7.                               holding_quantity int,
  8.                               datetime_created timestamp,
  9.                               datetime_updated timestamp,
  10.                               primary key(holding_id)
  11. );
  12. ALTER TABLE bank.holding replica identity FULL;
  13. insert into bank.holding values (1000, 1, 'VFIAX', 10, now(), now());
  14. \q
  15. insert into bank.holding values (1000, 1, 'VFIAX', 10, now(), now());
  16. insert into bank.holding values (1001, 2, 'SP500', 1, now(), now());
  17. insert into bank.holding values (1003, 3, 'SP500', 1, now(), now());
  18. update bank.holding set holding_quantity = 300 where holding_id=1000;

Target database table：bank1.holding

  1. create database bank1;
  2. CREATE TABLE holding (
  3.                           holding_id int,
  4.                           user_id int,
  5.                           holding_stock varchar(8),
  6.                           holding_quantity int,
  7.                           datetime_created bigint,
  8.                           datetime_updated bigint,
  9.                           primary key(holding_id)
  10. );

Start Connector

Start Debezium source connector

Synchronize original table data：bank.holding Purpose: Parse Postgres binlog and encapsulate it into a common ConnectRecord object, which is sent to the RocketMQ Topic.

  1. curl -X POST -H "Content-Type: application/json" http://127.0.0.1:8082/connectors/postgres-connector -d  '{
  2.   "connector.class": "org.apache.rocketmq.connect.debezium.postgres.DebeziumPostgresConnector",
  3.   "max.task": "1",
  4.   "connect.topicname": "debezium-postgres-source-01",
  5.   "kafka.transforms": "Unwrap",
  6.   "kafka.transforms.Unwrap.delete.handling.mode": "none",
  7.   "kafka.transforms.Unwrap.type": "io.debezium.transforms.ExtractNewRecordState",
  8.   "kafka.transforms.Unwrap.add.headers": "op,source.db,source.table",
  9.   "database.history.skip.unparseable.ddl": true,
  10.   "database.server.name": "bankserver1",
  11.   "database.port": 5432,
  12.   "database.hostname": "database ip",
  13.   "database.connectionTimeZone": "UTC",
  14.   "database.user": "start_data_engineer",
  15.   "database.dbname": "start_data_engineer",
  16.   "database.password": "password",
  17.   "table.whitelist": "bank.holding",
  18.   "key.converter": "org.apache.rocketmq.connect.runtime.converter.record.json.JsonConverter",
  19.   "value.converter": "org.apache.rocketmq.connect.runtime.converter.record.json.JsonConverter"
  20. }'

Start JDBC sink connector

Purpose: Consume data from the Topic and write it to the target table through JDBC protocol.

  1. curl -X POST -H "Content-Type: application/json" http://127.0.0.1:8082/connectors/jdbcmysqlsinktest201 -d '{
  2.   "connector.class": "org.apache.rocketmq.connect.jdbc.connector.JdbcSinkConnector",
  3.   "max.task": "2",
  4.   "connect.topicnames": "debezium-postgres-source-01",
  5.   "connection.url": "jdbc:mysql://database ip:3306/bank1",
  6.   "connection.user": "root",
  7.   "connection.password": "debezium",
  8.   "pk.fields": "holding_id",
  9.   "table.name.from.header": "true",
  10.   "pk.mode": "record_key",
  11.   "insert.mode": "UPSERT",
  12.   "db.timezone": "UTC",
  13.   "table.types": "TABLE",
  14.   "errors.deadletterqueue.topic.name": "dlq-topic",
  15.   "errors.log.enable": "true",
  16.   "errors.tolerance": "ALL",
  17.   "delete.enabled": "true",
  18.   "key.converter": "org.apache.rocketmq.connect.runtime.converter.record.json.JsonConverter",
  19.   "value.converter": "org.apache.rocketmq.connect.runtime.converter.record.json.JsonConverter"
  20. }'

After the creation of the above two Connector tasks, log in to the database using the start_data_engineer/password account.

Any add, delete, or modification made to the source database table bankholding will be synced to the target table bank1.holding.