Kafka-connect-jdbc: Support for topic prefix filter when defining table name

[daniellavoie]

Problem Description

Currently, the Sink Connector can use a placeholder in the table name format. This means that the table name will be composed from the entire topic name. For a table calledcustomers, the only option is to have a topic explicitly named customers. Table name format allows to mutate the destination name but not the source topic name.

Kafka being generally used in a multi-tenant environment, I might be interested in prefixing the topic names while the target table name is stripped from that prefix.

Example : topic : database1.cdc.customers and target table name : customers. This would allow handling table name conflict if Kafka is used to sink tables from different data sources to different databases.

Proposal

Add a filter configuration to the table name format. For example : topic.name-prefix set to database1.cdc. This configuration would be used to filter the prefix from the target table name.

[junzebao]

Why not just define table.name.format to be what you want? in your example just configure it to be "customers"?

[daniellavoie]

Great recommendation, however, it means one connector instance per table. With a topic.name-prefix, it would still be possible to support multiple tables with a single connector instance.

[junzebao]

ah yes, good point! It might be too much trouble to create one connector for one topic. I just found out some useful transformations that might give something working without what you're asking for. Like you could use the InsertField to inject the topic name into the key or value and then use the ExtractTopic to change the record topic. It sounds much of a trouble though ;) (Caveat: the ExtractTopic transformation is only provided by Confluent, not Apache Kafka)

I do see the need for the issue you brought up though and I think we'll probably face similar issues too. Hope they'll consider it, but at least we have some hacky "workarounds".

[hishamjaved]

This is very much needed feature to join multiple topics to merge the data from different source topics of different tenants into one sink table. Option should be generic to cater all the cases.

You can also use RegexRouter in JDBC Sink Connector as described below.

{
    "name": "connector_sink_common",
    "config": {
        "connector.class": "io.confluent.connect.jdbc.JdbcSinkConnector",
        "tasks.max": "1",       
        "connection.url": "jdbc:mysql://dev.sink.com:3306/sink_common?verifyServerCertificate=false&nullCatalogMeansCurrent=true",
        "connection.user": "root",
        "connection.password": "********",
                "topics.regex": "^ps_\\w+_sink.*",
        "auto.create": "true",
                 "auto.evolve": "true",
        "insert.mode": "upsert",
                "transforms": "route, unwrap",
                "transforms.route.type": "org.apache.kafka.connect.transforms.RegexRouter",
                "transforms.route.regex": "([^.]+)\\.([^.]+)",
                "transforms.route.replacement": "$2",
        "transforms.unwrap.type": "io.debezium.transforms.ExtractNewRecordState",
                "transforms.unwrap.drop.tombstones": "false",
                "delete.enabled": "true",
        "pk.mode": "record_key"
    }
}

[orasik]

I faced the same issue, the solution is to use dropPrefix.

Example, to save data from topics hello.dbo.table1,hello.dbo.table2 to table1 and table2 in the database, use the following config:

    tasks.max: 1
    topics: hello.dbo.table1, hello.dbo.table2
    connection.url: jdbc:sqlserver://server:port;database=dbname;user=dbuser
    connection.user: dbuser
    connection.password: dbpass
    transforms: dropPrefix,unwrap
    transforms.dropPrefix.type: org.apache.kafka.connect.transforms.RegexRouter
    transforms.dropPrefix.regex: hello\.dbo\.(.*)
    transforms.dropPrefix.replacement: $1
    transforms.unwrap.type: io.debezium.transforms.ExtractNewRecordState
    transforms.unwrap.drop.tombstones: false
    auto.create: true
    value.converter: org.apache.kafka.connect.json.JsonConverter
    value.converter.schemas.enable: true
    insert.mode: upsert
    delete.enabled: true
    pk.mode: record_key

[sauljabin]

I faced the same issue, the solution is to use dropPrefix.

Example, to save data from topics hello.dbo.table1,hello.dbo.table2 to table1 and table2 in the database, use the following config:

    tasks.max: 1
    topics: hello.dbo.table1, hello.dbo.table2
    connection.url: jdbc:sqlserver://server:port;database=dbname;user=dbuser
    connection.user: dbuser
    connection.password: dbpass
    transforms: dropPrefix,unwrap
    transforms.dropPrefix.type: org.apache.kafka.connect.transforms.RegexRouter
    transforms.dropPrefix.regex: hello\.dbo\.(.*)
    transforms.dropPrefix.replacement: $1
    transforms.unwrap.type: io.debezium.transforms.ExtractNewRecordState
    transforms.unwrap.drop.tombstones: false
    auto.create: true
    value.converter: org.apache.kafka.connect.json.JsonConverter
    value.converter.schemas.enable: true
    insert.mode: upsert
    delete.enabled: true
    pk.mode: record_key

It works perfectly, thanks!, I do not understand why the unwrap