search

apache / rocketmq-connect

A tool for scalable and reliably streaming data between Apache RocketMQ and other systems.
https://rocketmq.apache.org/
Apache License 2.0
121 stars 118 forks source link

创建的topic为null的问题 #424

Closed ityestop closed 1 year ago

ityestop commented 1 year ago

问题描述 在进行rocketmq-connect-runtime测试时，使用文档中的测试用例，结果运行结果是对的，可以正常生成文件，但是传递的topic是 null。

image

配置问题 使用当前master分支（92c14dac8f5667b64aa532bdab411cb95bc18f2c）打包配置，无变化

使用rocketMq 版本 rocketMq版本 4.9.2 rocketMq-client 4.7.1

触发命令

curl -X POST -H "Content-Type: application/json" http://127.0.0.1:8082/connectors/fileSourceConnector -d '{"connector.class":"org.apache.rocketmq.connect.file.FileSourceConnector","filename":"test-source-file.txt","connect.topicname":"fileTopic"}'

排查问题 https://github.com/apache/rocketmq-connect/blob/92c14dac8f5667b64aa532bdab411cb95bc18f2c/rocketmq-connect-sample/src/main/java/org/apache/rocketmq/connect/file/FileSourceTask.java#L149

image

当前代码获取 fileConfig.topic，但是在 connector 创建时传入的是 connect.topicname， 不存在赋值。此时一定为null传入，但是根据代码排查，调用的方法为 io.openmessaging.connector.api.data.ConnectRecord#addExtension(java.lang.String, java.lang.String) 最终执行方法为 /Users/cchu/.m2/repository/io/openmessaging/openmessaging-api/0.3.1-alpha/openmessaging-api-0.3.1-alpha.jar!/io/openmessaging/internal/DefaultKeyValue.class:35

image

根据java.lang.String#valueOf(java.lang.Object) 查看，发现对象传入为null时，返回的是”null“字符串。

https://github.com/apache/rocketmq-connect/blob/92c14dac8f5667b64aa532bdab411cb95bc18f2c/rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/connectorwrapper/WorkerSourceTask.java#L422

image

在org.apache.rocketmq.connect.runtime.connectorwrapper.WorkerSourceTask#maybeCreateAndGetTopic方法中的判断，优先取topic，补偿取 connect.topicname 。由于 topic = "null"，最终导致创建topic为 null

以上是排查的定位，不排除个人用法不当导致的问题，希望有朋友可以一直帮忙double check一下。

ityestop commented 1 year ago

使用文档 https://github.com/apache/rocketmq-connect#source-connector%E7%89%B9%E6%AE%8A%E9%85%8D%E7%BD%AE

image

根据文档的含义与实际代码的追踪，发现在某次的改动后，逻辑出现了变化

image
sunxiaojian commented 1 year ago

使用文档 https://github.com/apache/rocketmq-connect#source-connector%E7%89%B9%E6%AE%8A%E9%85%8D%E7%BD%AE

image

根据文档的含义与实际代码的追踪，发现在某次的改动后，逻辑出现了变化

image

提个 issue修复一下文档？

ityestop commented 1 year ago

我怀疑是这里的代码逻辑问题，不是文档的问题。逻辑上应该优先取 connect.topicname，不存在时再取 overwriteTopicRecord(record)。connect.topicname是配置指定的，topic是兜底时代码写死的

sunxiaojian commented 1 year ago

我怀疑是这里的代码逻辑问题，不是文档的问题。逻辑上应该优先取 connect.topicname，不存在时再取 overwriteTopicRecord(record)。connect.topicname是配置指定的，topic是兜底时代码写死的

connect.topicname配置不能为空，所以可能原始逻辑永远不会使用动态的topic，修改文档就可以了，或者让connect.topicname可以为空，但是不建议

ityestop commented 1 year ago

好的，我再尝试理解一下全流程逻辑后，再进行讨论。感谢