Kafka connector消费debezium oracle增量数据疑问

托马斯_X · 2023年10月9日 09:29

sr版本3.1.0
kafka版本3.5.1
debezium oracle connector 2.4

想问下有没有具体的代码实现 starrocks Kafka connector 消费kafka里debezium解析的oracle增量数据？
我根据官网的步骤尝试了下，报错：

{"error_code":400,"message":"Connector configuration is invalid and contains the following 2 error(s):\nInvalid value io.confluent.connect.json.JsonSchemaConverter for configuration key.converter: Class io.confluent.connect.json.JsonSchemaConverter could not be found.\nInvalid value io.confluent.connect.json.JsonSchemaConverter for configuration value.converter: Class io.confluent.connect.json.JsonSchemaConverter could not be found.\nYou can also find the above list of errors at the endpoint `/connector-plugins/{connectorType}/config/validate`"}

kafka connector配置如下：

curl -i http://10.133.1.1:8083/connectors -H "Content-Type: application/json" -X POST -d '{  "name":"starrocks-kafka-connector",  "config":{    "connector.class":"com.starrocks.connector.kafka.StarRocksSinkConnector",    "topics":"test1.t1",	"starrocks.topic2table.map":"test1.t1:t1",    "starrocks.http.url":"10.133.1.1:8030","starrocks.username":"root",    "starrocks.password":"xxxx","starrocks.database.name":"ems",    "key.converter":"io.confluent.connect.json.JsonSchemaConverter",    "value.converter":"io.confluent.connect.json.JsonSchemaConverter",	"transforms":"addfield,unwrap",	"transforms.addfield.type":"com.starrocks.connector.kafka.transforms.AddOpFieldForDebeziumRecord",	"transforms.unwrap.type":"io.debezium.transforms.ExtractNewRecordState","transforms.unwrap.drop.tombstones":"true",	"transforms.unwrap.delete.handling.mode":"rewrite"  }}'

jingdan · 2023年10月9日 12:39

参考文档，看起来配置不太对，https://docs.starrocks.io/zh-cn/main/loading/Kafka-connector-starrocks

托马斯_X · 2023年10月10日 00:55

前面的配置参考的这里，还是说只用后面就行了？

jingdan · 2023年10月10日 03:25

需要加下transforms的配置

托马斯_X · 2023年10月10日 03:37

上面的命令有，但还是报错

{
	"name": "starrocks-kafka-connector",
	"connector": {
		"state": "RUNNING",
		"worker_id": "10.133.1.1:8083"
	},
	"tasks": [
		{
			"id": 0,
			"state": "FAILED",
			"worker_id": "10.133.1.1:8083",
			"trace": "org.apache.kafka.connect.errors.ConnectException: Exiting WorkerSinkTask due to unrecoverable exception.\n\tat org.apache.kafka.connect.runtime.WorkerSinkTask.deliverMessages(WorkerSinkTask.java:618)\n\tat org.apache.kafka.connect.runtime.WorkerSinkTask.poll(WorkerSinkTask.java:336)\n\tat org.apache.kafka.connect.runtime.WorkerSinkTask.iteration(WorkerSinkTask.java:237)\n\tat org.apache.kafka.connect.runtime.WorkerSinkTask.execute(WorkerSinkTask.java:206)\n\tat org.apache.kafka.connect.runtime.WorkerTask.doRun(WorkerTask.java:204)\n\tat org.apache.kafka.connect.runtime.WorkerTask.run(WorkerTask.java:259)\n\tat org.apache.kafka.connect.runtime.isolation.Plugins.lambda$withClassLoader$1(Plugins.java:181)\n\tat java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:539)\n\tat java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264)\n\tat java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1136)\n\tat java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:635)\n\tat java.base/java.lang.Thread.run(Thread.java:833)\nCaused by: java.lang.NullPointerException: Cannot invoke \"java.lang.Throwable.getMessage()\" because the return value of \"com.starrocks.data.load.stream.v2.StreamLoadManagerV2.getException()\" is null\n\tat com.starrocks.connector.kafka.StarRocksSinkTask.put(StarRocksSinkTask.java:231)\n\tat org.apache.kafka.connect.runtime.WorkerSinkTask.deliverMessages(WorkerSinkTask.java:587)\n\t... 11 more\n"
		}
	],
	"type": "sink"
}

jingdan · 2023年10月12日 09:11

发下完整的kafka connector配置

托马斯_X · 2023年10月12日 09:14

完整的connector就上面的，通过curl方式添加

curl -i http://10.133.1.1:8083/connectors -H "Content-Type: application/json" -X POST -d '{  "name":"starrocks-kafka-connector",  "config":{    "connector.class":"com.starrocks.connector.kafka.StarRocksSinkConnector",    "topics":"test1.t1",	"starrocks.topic2table.map":"test1.t1:t1",    "starrocks.http.url":"10.133.1.1:8030","starrocks.username":"root",    "starrocks.password":"xxxx","starrocks.database.name":"ems",    "key.converter":"io.confluent.connect.json.JsonSchemaConverter",    "value.converter":"io.confluent.connect.json.JsonSchemaConverter",	"transforms":"addfield,unwrap",	"transforms.addfield.type":"com.starrocks.connector.kafka.transforms.AddOpFieldForDebeziumRecord",	"transforms.unwrap.type":"io.debezium.transforms.ExtractNewRecordState","transforms.unwrap.drop.tombstones":"true",	"transforms.unwrap.delete.handling.mode":"rewrite"  }}'

kafka connect用的standlone，配置默认