如何在 JDBC Kafka 源连接器中设置动态模式名​​称

Question

我正在使用 SAP HANA 源连接器 "connector.class":"com.sap.kafka.connect.source.hana.HANASourceConnector" 将数据导入 kafka 主题。

我想为连接器创建的 avro 模式设置服装 name/namespace。因此，我使用 SetSchemaMetadata 函数来转换模式名称。

这是我的连接器配置示例：

 "key.converter.schemas.enable": true,
 "key.converter": "io.confluent.connect.avro.AvroConverter",
 "key.converter.schema.registry.url": "http://xx.xx.xx/schema/",
 "key.converter.enhanced.avro.schema.support": true,
 "value.converter": "io.confluent.connect.avro.AvroConverter",
 "value.converter.schema.registry.url": "http://xx.xx.xx/schema/",
 "value.converter.enhanced.avro.schema.support": true,


 "transforms":"SetSchemaMetadata",
 "transforms.SetSchemaMetadata.type":"org.apache.kafka.connect.transforms.SetSchemaMetadata$Value",
 "transforms.SetSchemaMetadata.schema.name":"com.company.kafka.avro.table",
    
 "table1.table.name": "\"database\".\"schema.table1\"",
 "table2.table.name": "\"database\".\"schema.table2\"",

这将导致两个架构（table1 值和 table2 值）都将具有命名空间“com.company.kafka.avro”和名称“table”

有没有办法让架构名称与主题名称相同？类似于："transforms.SetSchemaMetadata.schema.name":"com.company.kafka.avro.${topic}",

或者是否可以为每个主题和模式手动指定它？

Answer 1

据我所知，没有直接的方法可以做到这一点。而是为每个表创建一个单独的连接器。