Topic name is invalid when starting MySQL Source(CDC) Connector #518
Comments
|
@Ziy1-Tan Rocketmq does not support creating topics with special characters. You can use Transforms to reroute to a new topic |
|
遇到了同样的问题，对着源码看了半天，可能有理解不到位的地方，但是至少解决了创建主题这一步的问题，以下是解决方案： private String maybeCreateAndGetTopic(ConnectRecord record) {
String topic = overwriteTopicFromRecord(record);
topic = taskConfig.getString(SourceConnectorConfig.CONNECT_TOPICNAME);
// 这里把判断条件取消掉，直接从配置中获取主题名
// if (StringUtils.isBlank(topic)) {
// // topic from config
// topic = taskConfig.getString(SourceConnectorConfig.CONNECT_TOPICNAME);
// }
if (StringUtils.isBlank(topic)) {
throw new ConnectException("source connect lack of topic config");
}
if (!workerConfig.isAutoCreateTopicEnable() || topicCache.contains(topic)) {
return topic;
}
if (!ConnectUtil.isTopicExist(workerConfig, topic)) {
ConnectUtil.createTopic(workerConfig, new TopicConfig(topic));
}
topicCache.add(topic);
return topic;
}
```
这里前半部分`overwriteTopicFromRecord`方法没有细看，但是大概理解是将插入数据的 数据库服务名.数据库名.数据库表 名作为主题名，但是因为出现了`.`符号，所以至少在RocketMQ5.x版本会出现创建主题失败的问题，代码这里对于主题名的修改，在传入的配置参数中，优先级比`overwriteTopicFromRecord`方法生成的低，出现配置了"connect.topicname"但是没有生效的问题，不知道后续是否会优化。。。
----
Connect感觉是个比较有用的工具，但是文档中举例的demo，可能是限于个人水平，只运行成功了文件的同步的例子，在做其他demo的时候遇到不少问题，对着源码一点点打断点，也没能完全解决，由于时间原因，最后只能放弃使用。未来还是希望项目能够热度再高一些，维护得更好一些。 |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
I follow the RocketMQ Connect example1.
sh bin/connect-standalone.sh -c conf/connect-standalone.conf &The source connector and task start successfully.
The Source task failed to create topic:
curl http://127.0.0.1:8082/connectors/list{ "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" } } } }The text was updated successfully, but these errors were encountered: