I was able to successfully connect Debezium CDC + Kafka over postgres. Also I was able to pickup debezium kafka topic over simple table books(serial pk, string title, string author) by JDBC Kafka connect back to postgresql db. Inserts, Updates, Deletes working fine over this simple table. But what to do, if I want to have one kafka connect topic for multiple postgres tables (debezium topics)? I need to transer data 1:1. Example below uses the same db, in real it will be written in other db, so table names would be consistent.
Below is the postgres config properties file
name=sink-postgres
connector.class=io.confluent.connect.jdbc.JdbcSinkConnector
value.converter=org.apache.kafka.connect.json.JsonConverter
value.converter.schemas.enable=true
topics=pokladna.public.books
connection.url=jdbc:postgresql://localhost:5434/postgres
connection.user=postgres
connection.password=postgres
dialect.name=PostgreSqlDatabaseDialect
table.name.format=books_kafka
transforms=unwrap
transforms.unwrap.type=io.debezium.transforms.ExtractNewRecordState
transforms.unwrap.drop.tombstones=false
auto.create=true
auto.evolve=true
insert.mode=upsert
delete.enabled=true
pk.fields=id
pk.mode=record_key
I have 2 questions:
- Do you know, how to set pk.mode and pk.fields to pickup changes in table with composite primary key which is under constrait?
Update and a answer: Simple composite key is fine (eg. PRIMARY KEY(question_id, tag_id)), just add to pk.fields column names. Also key under Constraint works fine (eg. CONSTRAINT tag_constraint RIMARY KEY(question_id, tag_id)) - you just cannost use the same database for source and target because different constraint names.
- How can I extract automatically primary column names from kafka message?
Bellow I attached output from debezium topic. I was thinking to use SMT extract field by name, but it will work for simple message asi it is shown in the example. In the case below, I guess, I have to flatten the message and extract Payload.after.id. But what I should do if the name of the primary key is different in each table?
bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --to
pokladna.public.books --property print.key=true --property key.separator=::::
{"schema":{"type":"struct","fields":[{"type":"int32","optional":false,"field":"id"}],"optional":false,
"name":"pokladna.public.books.Key"},"payload":{"id":8}}
::::
{"schema":{"type":"struct","fields":[{"type":"struct","fields":[{"type":"int32" ,"optional":false,"field":"id"},{"type":"string","optional":false,"field":"title"},{"type":"string","optional":true," field":"primary_author"}],"optional":true,"name":"pokladna.public.books.Value","field":"before"},{"type":"struct","fields":[{"type":"int32","optional":false,"field":"id"},{"type":"string","optional":false,"field":"title"},{"type":"string" ,"optional":true,"field":"primary_author"}],"optional":true,"name":"pokladna.public.books.Value",
"field":"after"},{"type":"struct","fields":[{"type":"string","optional":false,"field":"version"},{"type":"string","optional":false,"field":"connector"},{"type":"string","optional":false,"field":"name"},{"type":"int64","optional":false,"field":"ts_ms"},{"type":"string" ,"optional":true,"name":"io.debezium.data.Enum","version":1,
"parameters":{"allowed":"true,last,false"},"default":"false","field":"snapshot"},{"type":"string","optional":false,"field":"db"},{"type":"string","optional":false,
"field":"schema"} ,{"type":"string","optional":false,"field":"table"},{"type":"int64","optional":true,"field":"txId"},
{"type":"int64","optional":true,"field":"lsn"},{"type":"int64","optional":true,"field":"xmin"}],"optional":false,
"name":"io.debezium.connector.postgresql.Source","field":"source"},{"type":"string","optional":false,"field":"op"},{"type":"int64","optional":true,"field":"ts_ms"},{"type":"struct","fields":"type":"string","optional":false,"field":"id"},{"type":"int64", "optional":false,"field":"total_order"},{"type":"int64","optional":false,"field":"data_collection_order"}],"optional" :true,"field":"transaction"}],"optional":false,"name":"pokladna.public.books.Envelope"},
"payload":{"before":null,"after":{"id":8,"title":"jabadyba","primary_author":"jajajaja"},"source":{"version":"1.4.0-SNAPSHOT","connector":"postgresql" ,"name":"pokladna","ts_ms":1603718235441,"snapshot":"false","db":"postgres","schema":"public",
"table":"books","txId":716,"lsn":30457288,"xmin":null},"op":"c","ts_ms":1603718235812,"transaction":null}}