Multiple PKey sets of multiple tables support for `pk.fields`

[CC-Hsu]

Hi, Team,

I try to setup a single JDBC Sink connector task with multiple target tables without success, and I find that this goal currently might not be possible according to the discussion in Issue#743.

I feel that one reason I encounter is the pk.fields setting format is not easy to specify different primary columns for different tables.

However, I find that in Debezium there is a setting called message.key.columns, which can be used to specify different "new primary key fields" for different source tables.

Debezium message.key.columns is designed to accept different primary key fields setting for different tables as follows.

"message.key.columns": "<fully-qualified_tableName1>:_<keyColumnA>_,<keyColumnB>;<fully-qualified_tableName2>:_<keyColumnC>_,<keyColumnD>"

Is it possible to improve JDBC Sink Connector pk.fields setting to support similar config format?

What I imagine looks like the below excerpt.

"connector.class": "io.confluent.connect.jdbc.JdbcSinkConnector",
"topics": "<table1_topic>,<table2_topic>",
"table.name.format": "<fully-qualified_tableName1>,<fully-qualified_tableName2>",
"pk.fields": "<fully-qualified_tableName1>:_<keyColumnA>_,<keyColumnB>;<fully-qualified_tableName2>:_<keyColumnC>_,<keyColumnD>"

It would be helpful for users who want to setup multiple target tables inside a sink connector task.

Best Regards.