[rocketmq-connect-jdbc] Possible enhancements recorder

[gripson]

The issue tracker is ONLY used for bug report and feature request.

Any question or RocketMQ proposal please use our mailing lists.

BUG LIST

This is a recording for issues I've met when deploying rocketmq-connect-runtime, and there are not many core-related or performance enhancements I can observe now, so I'll just put some simple enhancements as candidates.

• [x] Remove unnecessary key-value for config request used to start/create connector
• [ ] Update jdbc connector unit test
• [ ] Fix checkstyle problem and issue of mvn -B clean apache-rat:check findbugs:findbugs checkstyle:checkstyle failure
• [x] Update Restful API template README.md file of jdbc connector. for example, Source connector Restful config due to current config guide in README.md is not lastest version:

http://${runtime-ip}:${runtime-port}/connectors/${rocketmq-jdbc-source-connector-name}?config={"source-rocketmq":"NameSrv-addr:9876","source-cluster":"${rocketmqTopic-topicRoute-cluster-value}","connector-class":"org.apache.rocketmq.connect.jdbc.connector.JdbcSourceConnector","dbUrl":"${source-db-ip}",dbPort”:"${source-db-port}",dbUsername":"${source-db-username}","dbPassword”:"${source-db-password}","rocketmqTopic":"jdbcTopic","mode":"bulk","whiteDataBase":{"${source-db-name}":{"${source-table-name}":{"${source-column-name}":"${source-column-value}"}}},"source-record-converter":"org.apache.rocketmq.connect.runtime.converter.JsonConverter"}

• [ ] Issue of sending/receiving many messages than expectation number of messages: it produces duplicate data into target database, even there is only one row data in source database table. Test environment: both of source and target DBMS are MySQL.

• [ ] Issue of query table error due to unknow column "id". It looks column "id" is hard-code in the source code. Maybe I am wrong, will double check it.

2020-04-06 10:20:20 ERROR pool-5-thread-2 - query table error,{}
java.sql.SQLSyntaxErrorException: Unknown column 'id' in 'field list'
        at com.mysql.cj.jdbc.exceptions.SQLError.createSQLException(SQLError.java:118) ~[rocketmq-connect-jdbc-0.0.1-SNAPSHOT-jar-with-dependencies.jar:na]
        at com.mysql.cj.jdbc.exceptions.SQLError.createSQLException(SQLError.java:95) ~[rocketmq-connect-jdbc-0.0.1-SNAPSHOT-jar-with-dependencies.jar:na]
        at com.mysql.cj.jdbc.exceptions.SQLExceptionsMapping.translateException(SQLExceptionsMapping.java:122) ~[rocketmq-connect-jdbc-0.0.1-SNAPSHOT-jar-with-dependencies.jar:na]
        at com.mysql.cj.jdbc.ClientPreparedStatement.executeInternal(ClientPreparedStatement.java:960) ~[rocketmq-connect-jdbc-0.0.1-SNAPSHOT-jar-with-dependencies.jar:na]
        at com.mysql.cj.jdbc.ClientPreparedStatement.executeQuery(ClientPreparedStatement.java:1019) ~[rocketmq-connect-jdbc-0.0.1-SNAPSHOT-jar-with-dependencies.jar:na]
        at com.alibaba.druid.pool.DruidPooledPreparedStatement.executeQuery(DruidPooledPreparedStatement.java:227) ~[rocketmq-connect-jdbc-0.0.1-SNAPSHOT-jar-with-dependencies.jar:na]
        at org.apache.rocketmq.connect.jdbc.sink.Updater.queryAfterUpdateRowId(Updater.java:176) [rocketmq-connect-jdbc-0.0.1-SNAPSHOT-jar-with-dependencies.jar:na]
        at org.apache.rocketmq.connect.jdbc.sink.Updater.push(Updater.java:47) [rocketmq-connect-jdbc-0.0.1-SNAPSHOT-jar-with-dependencies.jar:na]
        at org.apache.rocketmq.connect.jdbc.connector.JdbcSinkTask.put(JdbcSinkTask.java:74) [rocketmq-connect-jdbc-0.0.1-SNAPSHOT-jar-with-dependencies.jar:na]
        at org.apache.rocketmq.connect.runtime.connectorwrapper.WorkerSinkTask.receiveMessages(WorkerSinkTask.java:336) [rocketmq-connect-runtime-0.0.1-SNAPSHOT.jar:na]
        at org.apache.rocketmq.connect.runtime.connectorwrapper.WorkerSinkTask.pullMessageFromQueues(WorkerSinkTask.java:276) [rocketmq-connect-runtime-0.0.1-SNAPSHOT.jar:na]
        at org.apache.rocketmq.connect.runtime.connectorwrapper.WorkerSinkTask.run(WorkerSinkTask.java:262) [rocketmq-connect-runtime-0.0.1-SNAPSHOT.jar:na]
        at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511) [na:1.8.0_242]
        at java.util.concurrent.FutureTask.run(FutureTask.java:266) [na:1.8.0_242]
        at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149) [na:1.8.0_242]
        at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624) [na:1.8.0_242]
        at java.lang.Thread.run(Thread.java:748) [na:1.8.0_242]

• [ ] It is highly likely to get "Fetch topic list error" when jdbc sink connector is building topic routes . The root cause could be that the schedule thread created by JdbcSinkConnector tries getting data from srcMQAdminExt, but the data is not there since srcMQAdminExt shut down after first time creation.

The solution could be movingsrcMQAdminExt.shutdown() funtion from buildRoute() to stop(), while adding adminStarted = false; and cancel logic of executor.scheduleAtFixedRate(..) to the jdbcSinkConnector stop() function, as well as replace startMQAdminTools(); and buildRoute(); with alternative function to get latest topic route info in taskConfigs() function of jdbcSinkConnector.

However, it will bring about a consequence that the performance of the system reduces because if there will be more than one jdbcSinkConnectors created, each of them will have a schedule thread to check topic route info regually. Therefore I am going to think if there will be bette solution.

2020-04-06 09:53:34 ERROR JdbcSinkConnector-SinkWatcher-1 - Fetch topic list error.
org.apache.rocketmq.remoting.exception.RemotingConnectException: connect to <null> failed
        at org.apache.rocketmq.remoting.netty.NettyRemotingClient.invokeSync(NettyRemotingClient.java:392) ~[rocketmq-connect-jdbc-0.0.1-SNAPSHOT-jar-with-dependencies.jar:4.5.2]
        at org.apache.rocketmq.client.impl.MQClientAPIImpl.getTopicRouteInfoFromNameServer(MQClientAPIImpl.java:1331) ~[rocketmq-connect-jdbc-0.0.1-SNAPSHOT-jar-with-dependencies.jar:4.5.2]
        at org.apache.rocketmq.client.impl.MQClientAPIImpl.getTopicRouteInfoFromNameServer(MQClientAPIImpl.java:1321) ~[rocketmq-connect-jdbc-0.0.1-SNAPSHOT-jar-with-dependencies.jar:4.5.2]
        at org.apache.rocketmq.tools.admin.DefaultMQAdminExtImpl.examineTopicRouteInfo(DefaultMQAdminExtImpl.java:305) ~[rocketmq-connect-jdbc-0.0.1-SNAPSHOT-jar-with-dependencies.jar:na]
        at org.apache.rocketmq.tools.admin.DefaultMQAdminExt.examineTopicRouteInfo(DefaultMQAdminExt.java:251) ~[rocketmq-connect-jdbc-0.0.1-SNAPSHOT-jar-with-dependencies.jar:na]
        at org.apache.rocketmq.connect.jdbc.common.Utils.examineBrokerData(Utils.java:63) ~[rocketmq-connect-jdbc-0.0.1-SNAPSHOT-jar-with-dependencies.jar:na]
        at org.apache.rocketmq.connect.jdbc.connector.JdbcSinkConnector.buildRoute(JdbcSinkConnector.java:146) ~[rocketmq-connect-jdbc-0.0.1-SNAPSHOT-jar-with-dependencies.jar:na]
        at org.apache.rocketmq.connect.jdbc.connector.JdbcSinkConnector$1.run(JdbcSinkConnector.java:103) [rocketmq-connect-jdbc-0.0.1-SNAPSHOT-jar-with-dependencies.jar:na]
        at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511) [na:1.8.0_242]
        at java.util.concurrent.FutureTask.runAndReset(FutureTask.java:308) [na:1.8.0_242]
        at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$301(ScheduledThreadPoolExecutor.java:180) [na:1.8.0_242]
        at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:294) [na:1.8.0_242]
        at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149) [na:1.8.0_242]
        at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624) [na:1.8.0_242]
        at java.lang.Thread.run(Thread.java:748) [na:1.8.0_242]

POSSIBLE ENHANCEMENT/FEATURE REQUEST

• [ ] Release the binding relation between value of jdbc connector topics and database table name. Currently, jdbc connector only can work when rocketmqTopic of source connector and topicNames of sink connector are same as ${source-table-name} of whiteDataBase, and target DBMS has same name database and table.

• [ ] Add some other DBMS drivers and enhance related modules to make jdbc connector support multiple DBMS such as PostgreSQL.

[imaffe]

I can help with the "fetch topic list error" problem, feels like it would help me understand the codebase better :)