apache / apache/rocketmq-connect
Topic name is invalid when starting MySQL Source(CDC) Connector
- 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
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