# Trying to sync a table using "upsert" mode. It will only insert new rows, never update the existing ones **URL:** \ **Category:** Kafka Connect\ **Created:** [16 August 2021 08:14 UTC](https://forum.confluent.io/t/trying-to-sync-a-table-using-upsert-mode-it-will-only-insert-new-rows-never-update-the-existing-ones/2502 "2021-08-16T08:14:21Z")\ **Posts on this page:** 2\ **Page:** 1
**Author:** ![galiberti](https://avatars.discourse-cdn.com/v4/letter/g/22d042/32.png) [@galiberti](https://forum.confluent.io/u/galiberti)\ **Post date:** [16 August 2021 08:14 UTC](https://forum.confluent.io/t/trying-to-sync-a-table-using-upsert-mode-it-will-only-insert-new-rows-never-update-the-existing-ones/2502/1 "2021-08-16T08:14:21Z")
hello, i have these source and sync tables (mysql db): SOURCE _CREATE TABLE `f_event_source` (_ _`id_event` bigint(20) NOT NULL AUTO\_INCREMENT,_ _`event_value` varchar(20) DEFAULT NULL,_ _`category_description` varchar(20) DEFAULT NULL,_ _`tm_ins` datetime DEFAULT CURRENT\_TIMESTAMP,_ _`tm_upd` datetime DEFAULT CURRENT\_TIMESTAMP ON UPDATE CURRENT\_TIMESTAMP,_ _PRIMARY KEY (`id_event`),_ _KEY `idx_category` (`category_description`)_ _) ENGINE=InnoDB DEFAULT CHARSET=utf8;_ SYNC _CREATE TABLE `f_event_target` (_ _`id_event` bigint(20) NOT NULL AUTO\_INCREMENT,_ _`event_value` varchar(20) DEFAULT NULL,_ _`category_description` varchar(20) DEFAULT NULL,_ _`tm_ins` datetime DEFAULT CURRENT\_TIMESTAMP,_ _`tm_upd` datetime DEFAULT CURRENT\_TIMESTAMP ON UPDATE CURRENT\_TIMESTAMP,_ _PRIMARY KEY (`id_event`),_ _KEY `idx_category` (`category_description`)_ _) ENGINE=InnoDB DEFAULT CHARSET=utf8;_ my sink connector config: curl -X PUT [http://localhost:8083/connectors/sink-jdbc-mysql\_upsert/config](http://localhost:8083/connectors/sink-jdbc-mysql_upsert/config) -H “Content-Type: application/json” -d ‘{ “connector.class” : “io.confluent.connect.jdbc.JdbcSinkConnector”, “connection.url” : “jdbc:mysql://172.1.2.3:3306/db?serverTimezone=UTC”, “tasks.max” : “1”, “topics” : “f\_event\_source”, “transforms” : “unwrap”, “transforms.unwrap.type” : “io.debezium.transforms.ExtractNewRecordState”, “connection.user” : “user”, “connection.password” : “password”, “auto.create” : true, “insert.mode” : “upsert”, “pk.mode” : “record\_key”, “pk.fields” : “event\_value” }’ the KAFKA topic is: Struct{id\_event=17} {“schema”:{“type”:“struct”,“fields”:[{“type”:“struct”,“fields”:[{“type”:“int64”,“optional”:false,“field”:“id\_event”},{“type”:“string”,“optional”:true,“field”:“event\_value”},{“type”:“string”,“optional”:true,“field”:“category\_description”},{“type”:“int64”,“optional”:true,“name”:“io.debezium.time.Timestamp”,“version”:1,“default”:0,“field”:“tm\_ins”},{“type”:“int64”,“optional”:true,“name”:“io.debezium.time.Timestamp”,“version”:1,“default”:0,“field”:“tm\_upd”}],“optional”:true,“name”:“test.esperimento.f\_event\_source.Value”,“field”:“before”},{“type”:“struct”,“fields”:[{“type”:“int64”,“optional”:false,“field”:“id\_event”},{“type”:“string”,“optional”:true,“field”:“event\_value”},{“type”:“string”,“optional”:true,“field”:“category\_description”},{“type”:“int64”,“optional”:true,“name”:“io.debezium.time.Timestamp”,“version”:1,“default”:0,“field”:“tm\_ins”},{“type”:“int64”,“optional”:true,“name”:“io.debezium.time.Timestamp”,“version”:1,“default”:0,“field”:“tm\_upd”}],“optional”:true,“name”:“test.esperimento.f\_event\_source.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”:true,“field”:“table”},{“type”:“int64”,“optional”:false,“field”:“server\_id”},{“type”:“string”,“optional”:true,“field”:“gtid”},{“type”:“string”,“optional”:false,“field”:“file”},{“type”:“int64”,“optional”:false,“field”:“pos”},{“type”:“int32”,“optional”:false,“field”:“row”},{“type”:“int64”,“optional”:true,“field”:“thread”},{“type”:“string”,“optional”:true,“field”:“query”}],“optional”:false,“name”:“io.debezium.connector.mysql.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”:“test.esperimento.f\_event\_source.Envelope”},“payload”:{“before”:null,“after”:{“id\_event”:17,“event\_value”:“test143”,“category\_description”:“selectw4”,“tm\_ins”:1628593336000,“tm\_upd”:1628593336000},“source”:{“version”:“1.4.2.Final”,“connector”:“mysql”,“name”:“test”,“ts\_ms”:1628586136000,“snapshot”:“false”,“db”:“esperimento”,“table”:“f\_event\_source”,“server\_id”:190773,“gtid”:“d94d3a4b-6ba9-11eb-b997-02f847d0c6bf:163”,“file”:“mysql-bin.000062”,“pos”:15724,“row”:0,“thread”:35,“query”:null},“op”:“c”,“ts\_ms”:1628586136964,“transaction”:null}} with this configuration, when an existing row from the source is updated, a new row in the target table is created - this doesn’t update the existing one. I tried to use the field “id\_event” as record key, but in this case the sink won’t work - basically the sink connector tries to write the key from kafka record above “Struct{id\_event=17}” inside a bigint field and fails. What can i do to make it work? thx G. ---
**Author:** ![system](https://us1.discourse-cdn.com/flex019/uploads/confluentcommunity/original/1X/c49438c90c9df282e9996fdf6971be890c71b65a.svg) [@system](https://forum.confluent.io/u/system)\ **Post date:** [15 September 2021 08:14 UTC](https://forum.confluent.io/t/trying-to-sync-a-table-using-upsert-mode-it-will-only-insert-new-rows-never-update-the-existing-ones/2502/2 "2021-09-15T08:14:51Z")
This topic was automatically closed 30 days after the last reply. New replies are no longer allowed.