英文:
Kafka connect oracle-cdc autogenerates schemas for every topic with the same name "ConnectDefault". Can it be customized per topic?
问题
I'm using Kafka Connect with the oracle cdc (https://docs.confluent.io/kafka-connectors/oracle-cdc/current/overview.html) source connector to do CDC on some database tables. It creates a topic per table and autogenerates an Avro schema form each one. The problem is all schemas have the same namespace and name:
"name": "ConnectDefault",
"namespace": "io.confluent.connect.avro",
"type": "record"
Can this be customized?
I've already tried setting different configuration properties in the connector config but none of the fixed the problem:
"value.converter.schema.name.strategy": "io.confluent.kafka.serializers.subject.TopicNameStrategy",
"value.converter.schema.registry.subject": "my.namespace.myname",
"transforms":"AddNamespace",
"transforms.AddNamespace.type":"org.apache.kafka.connect.transforms.SetSchemaMetadata$Value",
"transforms.AddNamespace.timestamp.record.namespace":"example",
...
英文:
I'm using Kafka Connect with the oracle cdc (https://docs.confluent.io/kafka-connectors/oracle-cdc/current/overview.html) source connector to do CDC on some database tables. It creates a topic per table and autogenerates an Avro schema form each one. The problem is all schemas have the same namespace and name:
"name": "ConnectDefault",
"namespace": "io.confluent.connect.avro",
"type": "record"
Can this be customized?
I've already tried setting different configuration properties in the connector config but none of the fixed the problem:
"value.converter.schema.name.strategy": "io.confluent.kafka.serializers.subject.TopicNameStrategy",
"value.converter.schema.registry.subject": "my.namespace.myname",
"transforms":"AddNamespace",
"transforms.AddNamespace.type":"org.apache.kafka.connect.transforms.SetSchemaMetadata$Value",
"transforms.AddNamespace.timestamp.record.namespace":"example",
...
答案1
得分: 1
接受的答案建议使用转换。对于这个特定组件,您必须考虑以下内容:
改变模式的SMT仅限于使用谓词的表主题,如SMT示例文档中所示。
使用谓词和转换,代码如下:
"predicates": "idDepartment",
"predicates.idDepartment.pattern": "depcdc",
"predicates.idDepartment.type": "org.apache.kafka.connect.transforms.predicates.TopicNameMatches",
"transforms": "SetSchemaDepKey, SetSchemaDepValue",
"transforms.SetSchemaDepKey.type": "org.apache.kafka.connect.transforms.SetSchemaMetadata$Key",
"transforms.SetSchemaDepKey.schema.name": "com.codependent.DepKey",
"transforms.SetSchemaDepKey.predicate": "idDepartment",
"transforms.SetSchemaDepValue.type": "org.apache.kafka.connect.transforms.SetSchemaMetadata$Value",
"transforms.SetSchemaDepValue.schema.name": "com.codependent.DepValue",
"transforms.SetSchemaDepValue.predicate": "idDepartment"
请注意,这是您提供的代码的翻译部分。
英文:
The accepted answer suggests using a transform. For this particular component you have to consider this:
> The SMTs that alter schema are restricted to the table topics using
> predicates as demonstrated in the SMT Examples documentation.
Using predicates and transform the code looks like this:
"predicates": "idDepartment",
"predicates.idDepartment.pattern": "depcdc",
"predicates.idDepartment.type": "org.apache.kafka.connect.transforms.predicates.TopicNameMatches",
"transforms": "SetSchemaDepKey, SetSchemaDepValue",
"transforms.SetSchemaDepKey.type": "org.apache.kafka.connect.transforms.SetSchemaMetadata$Key",
"transforms.SetSchemaDepKey.schema.name": "com.codependent.DepKey",
"transforms.SetSchemaDepKey.predicate": "idDepartment",
"transforms.SetSchemaDepValue.type": "org.apache.kafka.connect.transforms.SetSchemaMetadata$Value",
"transforms.SetSchemaDepValue.schema.name": "com.codependent.DepValue",
"transforms.SetSchemaDepValue.predicate": "idDepartment",
答案2
得分: 0
更改主题的正确属性名称是 value.converter.
+ value.schema.name.strategy
(您缺少了value
部分)。
schema.registry.subject
不是有效的配置;请使用变换进行设置。
但这不会更改命名空间。对于命名空间,timestamp.record.namespace
也不是正确的属性,正确的是 schema.name
。
您提供的链接是配置详细信息的来源。
英文:
The correct property name for changing the subject is value.converter.
+ value.schema.name.strategy
(you are missing the value
part)
schema.registry.subject
is not a valid config; use the transform for that.
But that will not change the namespace. For that, timestamp.record.namespace
is not the correct property, either. It is schema.name
https://docs.confluent.io/platform/current/connect/transforms/setschemametadata.html#properties
通过集体智慧和协作来改善编程学习和解决问题的方式。致力于成为全球开发者共同参与的知识库,让每个人都能够通过互相帮助和分享经验来进步。
评论