Kafka JDBC connector producing events validating schema value and schema ID.
So, I have a topic with JSON Schema Value and JSON Schema Key. My connector needs to produce events that validate the schemas, i.e., magic type, schema ID, and the rest of the message. Does anyone know if this is possible with single transformation or custom transformation? Below is my connector; it has a custom SMT, but I haven’t achieved my goal. Has anyone encountered this before?
{
“connector.class”: “io.confluent.connect.jdbc.JdbcSourceConnector”,
“errors.log.enable”: true,
“errors.log.include.messages”: true,
“tasks.max”: “1”,
“connection.url”: “${CONNECT_CONNECTION}”,
“connection.user”: “${CONNECT_USER}”,
“connection.password”: “${CONNECT_PASSWORD}”,
“connection.attempts”: “999”,
“connection.backoff.ms”: “300000”,
“table.types”: “VIEW”,
“table.whitelist”: “vw_onthefly”,
“mode”: “timestamp”,
“timestamp.column.name”: “DT_UPDATED”,
“poll.interval.ms”: “${CONNECT_INTERVAL}”,
“batch.max.rows”: “${CONNECT_FETCHSIZE}”,
“fetch.size”: “${CONNECT_FETCHSIZE}”,
“topic.prefix”: “op.my-topic”,
“db.timezone”: “Europe/London”,
“errors.retry.delay.max.ms”: “5000”,
“errors.retry.timeout”: “15000”,
“errors.tolerance”: “none”,
“key.converter”: “org.apache.kafka.connect.json.JsonConverter”,
“key.converter.schemas.enable”: “false”,
“value.converter”: “org.apache.kafka.connect.json.JsonConverter”,
“value.converter.schemas.enable”: “false”,
“auto.register.schemas”: “false”,
“use.latest.version”: “true”,
“transforms”: “CustomTransform,ReplaceTopic”,
“transforms.CustomTransform.type”: “kafka.connect.MY_SMT”,
“transforms.ReplaceTopic.type”: “org.apache.kafka.connect.transforms.RegexRouter”,
“transforms.ReplaceTopic.regex”: “(.*)”,
“transforms.ReplaceTopic.replacement”: “eop.my-topic”
}