Estoy tratando de usar Kafka Connect para sincronizar datos de una base de datos DB2 antigua a una base de datos Postgres usando los conectores JDBC Source and Sink. Funciona bien, pero solo si soy muy estricto con el caso que uso para los nombres de las tablas.

Por ejemplo, tengo una tabla en DB2 llamada ACCIÓN y también existe en Postgres con las mismas columnas, etc. La única diferencia es que en DB2 es ACTION en mayúsculas y en Postgres es action en minúsculas.

Aquí hay un archivo receptor que funciona:

 { "name": "jdbc_sink_pg_action", "config": { "_comment": "The JDBC connector class", "connector.class": "io.confluent.connect.jdbc.JdbcSinkConnector", "_comment": "How to serialise the value of keys ", "key.converter": "org.apache.kafka.connect.json.JsonConverter", "_comment": "As above, but for the value of the message. Note that these key/value serialisation settings can be set globally for Connect and thus omitted for individual connector configs to make them shorter and clearer", "value.converter": "org.apache.kafka.connect.json.JsonConverter", "_comment": " --- JDBC-specific configuration below here --- ", "_comment": "JDBC connection URL.", "connection.url": "jdbc:postgresql://localhost:5435/postgres", "connection.user": "postgres", "connection.password": "*****", "topics": "ACTION", "table.name.format": "action", "_comment": "The insertion mode to use", "insert.mode": "upsert", "_comment": "The primary key mode", "pk.mode": "record_value", "_comment": "List of comma-separated primary key field names. The runtime interpretation of this config depends on the pk.mode", "pk.fields": "ACTION_ID", "quote.sql.identifiers": "never" } }

Esto está bien, pero no es muy flexible. Por ejemplo, tengo muchas otras tablas y me gustaría sincronizarlas también, pero no quiero crear un archivo conector para todas y cada una de las tablas. Así que trato de usar:

"table.name.format": "${topic}",

Cuando hago esto, aparece el siguiente error en los registros cuando intento cargar mi conector receptor:

Causado por: org.apache.kafka.connect.errors.ConnectException: Falta la tabla "ACCIÓN" y la creación automática está deshabilitada

Así que me parece que "quote.sql.identifiers": "never" en realidad no funciona; de lo contrario, la consulta que está haciendo el conector receptor no estaría entre comillas y permitiría cualquier caso (se convertiría a menor).

¿Por qué esto no funciona? Obtengo los mismos resultados si solo uso ACTION como table.name.format.