apache / apache/rocketmq-connect

Topic name is invalid when starting MySQL Source(CDC) Connector

Open
#518 3 comments 0 reactions 0 assignees View on GitHub
Dominant language
Java
Stars
141
Forks
120
PR merge metrics
No merged PRs in 30d

Description

1. Please describe the issue you observed:

- What did you do (The steps to reproduce)?

I follow the RocketMQ Connect [example1](https://rocketmq.apache.org/zh/docs/connect/04RocketMQ%20Connect%20In%20Action1/).
- Start RocketMQ
- Start Connect Runtime `sh bin/connect-standalone.sh -c conf/connect-standalone.conf &`
- Start MySQL Source(CDC) Connector:
```bash
curl -X POST -H "Content-Type: application/json" http://127.0.0.1:8082/connectors/MySQLCDCSource -d '{
"connector.class": "org.apache.rocketmq.connect.debezium.mysql.DebeziumMysqlConnector",
"max.task": "1",
"connect.topicname": "debezium-mysql-source-topic",
"kafka.transforms": "Unwrap",
"kafka.transforms.Unwrap.delete.handling.mode": "none",
"kafka.transforms.Unwrap.type": "io.debezium.transforms.ExtractNewRecordState",
"kafka.transforms.Unwrap.add.headers": "op,source.db,source.table",
"database.history.skip.unparseable.ddl": true,
"database.history.name.srv.addr": "localhost:9876",
"database.history.rocketmq.topic": "db-history-debezium-topic",
"database.history.store.only.monitored.tables.ddl": true,
"include.schema.changes": false,
"database.server.name": "dbserver1",
"database.port": 3306,
"database.hostname": "127.0.0.1",
"database.connectionTimeZone": "UTC",
"database.user": "debezium",
"database.password": "dbz",
"table.include.list": "inventory.employee",
"max.batch.size": 50,
"database.include.list": "inventory",
"snapshot.mode": "when_needed",
"database.server.id": "184054",
"key.converter": "org.apache.rocketmq.connect.runtime.converter.record.json.JsonConverter",
"value.converter": "org.apache.rocketmq.connect.runtime.converter.record.json.JsonConverter"
}'
```

- What is expected to see?

The source connector and task start successfully.

- What did you see instead?

The Source task failed to create topic: `curl http://127.0.0.1:8082/connectors/list`
```json
{
"status": 200,
"body": {
"MySQLCDCSource": {
"status": {
"name": "MySQLCDCSource",
"connector": {
"state": "RUNNING",
"trace": null,
"workerId": "standalone-worker"
},
"tasks": [
{
"state": "FAILED",
"trace": "java.lang.RuntimeException: Create topic [dbserver1.inventory.employee] failed\n\tat org.apache.rocketmq.connect.runtime.utils.ConnectUtil.createTopic(ConnectUtil.java:189)\n\tat org.apache.rocketmq.connect.runtime.connectorwrapper.WorkerSourceTask.maybeCreateAndGetTopic(WorkerSourceTask.java:439)\n\tat org.apache.rocketmq.connect.runtime.connectorwrapper.WorkerSourceTask.sendRecord(WorkerSourceTask.java:237)\n\tat org.apache.rocketmq.connect.runtime.connectorwrapper.WorkerSourceTask.execute(WorkerSourceTask.java:518)\n\tat org.apache.rocketmq.connect.runtime.connectorwrapper.WorkerTask.doExecute(WorkerTask.java:119)\n\tat org.apache.rocketmq.connect.runtime.connectorwrapper.WorkerTask.doRun(WorkerTask.java:224)\n\tat org.apache.rocketmq.connect.runtime.connectorwrapper.WorkerTask.run(WorkerTask.java:244)\n\tat java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:515)\n\tat java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264)\n\tat java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128)\n\tat java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628)\n\tat io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30)\n\tat java.base/java.lang.Thread.run(Thread.java:829)\nCaused by: org.apache.rocketmq.client.exception.MQClientException: CODE: 1 DESC: The specified topic contains illegal characters, allowing only ^[%|a-zA-Z0-9_-]+$\nFor more information, please visit the url, http://rocketmq.apache.org/docs/faq/\n\tat org.apache.rocketmq.client.impl.MQClientAPIImpl.createTopic(MQClientAPIImpl.java:306)\n\tat org.apache.rocketmq.tools.admin.DefaultMQAdminExtImpl.createAndUpdateTopicConfig(DefaultMQAdminExtImpl.java:201)\n\tat org.apache.rocketmq.tools.admin.DefaultMQAdminExt.createAndUpdateTopicConfig(DefaultMQAdminExt.java:170)\n\tat org.apache.rocketmq.connect.runtime.utils.ConnectUtil.createTopic(ConnectUtil.java:185)\n\t... 12 more\n",
"workerId": "standalone-worker",
"id": 0
}
],
"type": "SOURCE"
},
"info": {
"name": "MySQLCDCSource",
"config": {
"connector.class": "org.apache.rocketmq.connect.debezium.mysql.DebeziumMysqlConnector",
"kafka.transforms.Unwrap.type": "io.debezium.transforms.ExtractNewRecordState",
"include.schema.changes": "false",
"kafka.transforms.Unwrap.add.headers": "op,source.db,source.table",
"connect.topicname": "debezium-mysql-source-topic",
"database.history.skip.unparseable.ddl": "true",
"value.converter": "org.apache.rocketmq.connect.runtime.converter.record.json.JsonConverter",
"key.converter": "org.apache.rocketmq.connect.runtime.converter.record.json.JsonConverter",
"database.user": "debezium",
"kafka.transforms.Unwrap.delete.handling.mode": "none",
"database.server.id": "184054",
"database.server.name": "dbserver1",
"database.port": "3306",
"max.task": "1",
"database.history.rocketmq.topic": "db-history-debezium-topic",
"database.history.name.srv.addr": "localhost:9876",
"database.hostname": "127.0.0.1",
"database.connectionTimeZone": "UTC",
"database.password": "dbz",
"database.history.store.only.monitored.tables.ddl": "true",
"table.include.list": "inventory.employee",
"max.batch.size": "50",
"kafka.transforms": "Unwrap",
"database.include.list": "inventory",
"snapshot.mode": "when_needed"
},
"tasks": [
{
"connector": "MySQLCDCSource",
"task": 0
}
],
"type": "SOURCE"
}
}
}
}
```

2. Please tell us about your environment:

- Ubuntu 20.04
- RocketMQ: develop
- RocketMQ-Connect: master

Contributor guide

Open the contributing guide

Research direction

Start at ConnectUtil.createTopic and WorkerSourceTask.maybeCreateAndGetTopic, using the supplied MySQL CDC configuration to reproduce the failed topic creation. Trace how the topic name is formed and verify the connector and task start successfully with a RocketMQ-accepted topic name; add or update focused tests if the surrounding code provides them.

Written by the indexing model from the issue text.

Assessment

Tech stack
java, mysql
Domain
backend, databases
Issue type
Bug
Difficulty
3/5
Estimated time
1-2 days
Activity status
Quiet
Clarity
Mostly clear
Newbie friendliness
48/100

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.