TDengine Kafka Connector将 Kafka 中指定 topic 的数据（批量或实时）同步到 TDengine

教程放在这里：TDengine Java Connector，官方文档已经写的很清晰了，不再赘述。

这里记录一下踩坑：

1.报错

java.lang.UnsatisfiedLinkError: no taos in java.library.pathat java.lang.ClassLoader.loadLibrary(ClassLoader.java:1860)at java.lang.Runtime.loadLibrary0(Runtime.java:843)at java.lang.System.loadLibrary(System.java:1136)at com.taosdata.jdbc.TSDBJNIConnector.<clinit>(TSDBJNIConnector.java:30)at com.taosdata.jdbc.TSDBDriver.connect(TSDBDriver.java:159)at java.sql.DriverManager.getConnection(DriverManager.java:664)at java.sql.DriverManager.getConnection(DriverManager.java:208)at com.taosdata.kafka.connect.db.TSDBConnectionProvider.getConnection(TSDBConnectionProvider.java:35)at com.taosdata.kafka.connect.db.CacheProcessor.getConnection(CacheProcessor.java:40)at com.taosdata.kafka.connect.db.CacheProcessor.execute(CacheProcessor.java:66)at com.taosdata.kafka.connect.db.CacheProcessor.initDB(CacheProcessor.java:55)at com.taosdata.kafka.connect.db.CacheProcessor.setDbName(CacheProcessor.java:33)...

这是由于没有按照taos客户端，连机器内部调用了taos客户端程序去连接TDengine服务，只要安装TDengine Client即可。安装TDengine Client教程

2.报错

Caused by: java.lang.NoClassDefFoundError: Could not initialize class com.taosdata.jdbc.TSDBJNIConnectorat com.taosdata.jdbc.TSDBDriver.connect(TSDBDriver.java:159)at java.sql.DriverManager.getConnection(DriverManager.java:664)at java.sql.DriverManager.getConnection(DriverManager.java:208)at com.taosdata.kafka.connect.db.TSDBConnectionProvider.getConnection(TSDBConnectionProvider.java:35)at com.taosdata.kafka.connect.db.CacheProcessor.getConnection(CacheProcessor.java:40)at com.taosdata.kafka.connect.db.CacheProcessor.execute(CacheProcessor.java:66)at com.taosdata.kafka.connect.db.CacheProcessor.initDB(CacheProcessor.java:55)at com.taosdata.kafka.connect.db.CacheProcessor.setDbName(CacheProcessor.java:33)at com.taosdata.kafka.connect.sink.TDengineSinkTask.bulkWriteBatch(TDengineSinkTask.java:94)at com.taosdata.kafka.connect.sink.TDengineSinkTask.put(TDengineSinkTask.java:85)at org.apache.kafka.connect.runtime.WorkerSinkTask.deliverMessages(WorkerSinkTask.java:601)... 11 more

这个报错是因为我部署TDengine服务的时候用的是docker的方式部署，用原生连接的方式连接TDengine，很多依赖库本地访问不了导致的报错。我们可以改用TestFul的方式去连接。

下面是官方教程中创建 Sink Connector 实例用到的配置文件

{"name": "TDengineSinkConnector","config": {"connection.database": "power","connection.password": "taosdata","connection.url": "jdbc:TAOS://127.0.0.1:6030","connection.user": "root","connector.class": "com.taosdata.kafka.connect.sink.TDengineSinkConnector","data.precision": "ns","db.schemaless": "line","key.converter": "org.apache.kafka.connect.storage.StringConverter","tasks.max": "1","topics": "meters","value.converter": "org.apache.kafka.connect.storage.StringConverter","name": "TDengineSinkConnector","errors.tolerance": "all","errors.deadletterqueue.topic.name": "dead_letter_topic","errors.deadletterqueue.topic.replication.factor": "1",    },"tasks": [],"type": "sink"
}

阅读源码后，我们将其中的"connection.url": "jdbc:TAOS://127.0.0.1:6030"修改为"connection.url": "jdbc:TAOS-RS://127.0.0.1:6041?user=root&password=taosdata"。

解释一下：

• jdbc:TAOS:代表使用原生连接，jdbc:TAOS-RS代表使用REST连接。
• 加上?user=root&password=taosdata是因为我发现改成REST连接后，connection.user配置项不生效，只好在url上直接拼接了。