如何将单消息转换与 Kafka Connect JDBC 源连接器和多个表一起使用？

Question

我想在使用 Kafka Connect 源 JDBC 连接器导入表时设置消息键。

当定义了多个表以从 JDBC 连接器读取时，如何将 Kafka Connect/Source 中的单消息转换 (SMT) 定位到正确的字段？SMT 需要一个列名，当有多个表时，该列名可能会有所不同。

我没有看到基于表名或类似名称过滤 SMT 定义的方法。下面的代码示例工作正常，因为它只有一张表。

但是如果你有不同的表，比如 User, Order, Product 怎么办？

"table.whitelist" : "User"
"transforms":"createKey,extract",
"transforms.createKey.type":"org.apache.kafka.connect.transforms.ValueToKey",
                "transforms.createKey.fields":"user_id",
"transforms.extract.type":"org.apache.kafka.connect.transforms.ExtractField\$Key",
"transforms.extract.field":"user_id",

当具有该配置的工作任务遇到没有该 user_id 字段的表时，它会崩溃并保持 FAILED 状态

org.apache.kafka.connect.errors.ConnectException: 
Tolerance exceeded in error handler
at org.apache.kafka.connect.runtime.errors.RetryWithToleranceOperator.execAndHandleError(RetryWithToleranceOperator.java:178)
at org.apache.kafka.connect.runtime.errors.RetryWithToleranceOperator.execute(RetryWithToleranceOperator.java:104)
at org.apache.kafka.connect.runtime.TransformationChain.apply(TransformationChain.java:50)
at org.apache.kafka.connect.runtime.WorkerSourceTask.sendRecords(WorkerSourceTask.java:293)
at org.apache.kafka.connect.runtime.WorkerSourceTask.execute(WorkerSourceTask.java:229)
at org.apache.kafka.connect.runtime.WorkerTask.doRun(WorkerTask.java:175)
at org.apache.kafka.connect.runtime.WorkerTask.run(WorkerTask.java:219)
at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
at java.util.concurrent.FutureTask.run(FutureTask.java:266)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
at java.lang.Thread.run(Thread.java:748)\nCaused by: java.lang.NullPointerException
at org.apache.kafka.connect.transforms.ValueToKey.applyWithSchema(ValueToKey.java:85)
at org.apache.kafka.connect.transforms.ValueToKey.apply(ValueToKey.java:65)
at org.apache.kafka.connect.runtime.TransformationChain.lambda$apply$0(TransformationChain.java:50)
at org.apache.kafka.connect.runtime.errors.RetryWithToleranceOperator.execAndRetry(RetryWithToleranceOperator.java:128)
at org.apache.kafka.connect.runtime.errors.RetryWithToleranceOperator.execAndHandleError(RetryWithToleranceOperator.java:162)
... 11 more

这是合理的，因为不可能通过表或目标光学来定义，或者是吗？我希望能够将转换限制为给定的表格或主题，例如

transforms.<topic-name>.createKey.type

我错过了什么还是连接限制？

Answer 1

不可能将 SMT 仅应用于特定主题，因为这是连接器级别的配置，这意味着它应用于每个处理过的消息。

我建议您为每个主题创建不同的连接器，以便您可以仅将 SMT 应用于主题的一个子集。