# Jdbc sink connector to load data from 2 different topics to 2 different table

**URL:** <https://forum.confluent.io/t/jdbc-sink-connector-to-load-data-from-2-different-topics-to-2-different-table/5143>\
**Category:** Kafka Connect\
**Created:** [25 May 2022 13:09 UTC](https://forum.confluent.io/t/jdbc-sink-connector-to-load-data-from-2-different-topics-to-2-different-table/5143 "2022-05-25T13:09:04Z")\
**Posts on this page:** 5\
**Page:** 1

<div class="post-metadata">

**Author:** ![sangeetha](https://sea1.discourse-cdn.com/flex019/user_avatar/forum.confluent.io/sangeetha/32/1935_2.png) [@sangeetha](https://forum.confluent.io/u/sangeetha)\
**Post date:** [25 May 2022 13:09 UTC](https://forum.confluent.io/t/jdbc-sink-connector-to-load-data-from-2-different-topics-to-2-different-table/5143/1 "2022-05-25T13:09:04Z")

</div>

I have 2 different kafka topics (task,member) which does not have  
similar column names. Using JDBC sink connector I have to write the data from kafka topics (task,member) to mysql tables(test\_task,test\_member) using UPSERT mode

**data in task topics:(jsondata)**  
{“orderid”: 1,“name”:“sangeetha”,“products”:“Mobile”,“modeofpayment”:“UPI”}

**data in membertopic:(jsondata)**  
{“pzinskey”: “1”,“membername”:“sangeetha”,“type”:“Gold”}

**Below is my sink connector configuration:**  
curl -X PUT -H “Content-Type: application/json” [http://localhost:8083/connectors/multipletopics/config](http://localhost:8083/connectors/multipletopics/config) -d ‘{  
“name” : “multipletopics”,  
“connector.class” : “io.confluent.connect.jdbc.JdbcSinkConnector”,  
“connection.url”: “jdbc:mysql://172.19.0.2:3306/sample”,  
“connection.user”: “user”,  
“connection.password”: “\*\*\*\*\*”,  
“topics”: “task,member”,  
“auto.create”:“true”,  
“auto.evolve”:“true”,  
“insert.mode”:“upsert”,  
“pk.mode”:“record\_value”,  
“pk.fields”:“orderid,pzinskey”,  
“table.name.format”:“test\_${topic}”,  
“value.converter” : “io.confluent.connect.json.JsonSchemaConverter”,  
“value.converter.schema.registry.url”: “[http://schema-registry:8081](http://schema-registry:8081/)”,  
“tasks.max”: “1”  
}’

**but the above configuration is not working… getting below error**  
Caused by: org.apache.kafka.connect.errors.ConnectException: PK mode for table ‘test\_task’ is RECORD\_VALUE with configured PK fields [orderid, pzinskey], but record value schema does not contain field: pzinskey

Is there a way to write data from 2 different topics to 2 different table using UPSERT mode via jdbc sink connector

---

<div class="post-metadata">

**Author:** ![OneCricketeer](https://sea1.discourse-cdn.com/flex019/user_avatar/forum.confluent.io/onecricketeer/32/1551_2.png) [@OneCricketeer](https://forum.confluent.io/u/OneCricketeer)\
**Post date:** [26 May 2022 11:55 UTC](https://forum.confluent.io/t/jdbc-sink-connector-to-load-data-from-2-different-topics-to-2-different-table/5143/2 "2022-05-26T11:55:42Z")

</div>

> record value schema does not contain field: pzinskey

You’ve shown the records, but please show the JSONSchema you’ve stored in the registry

Alternatively, neither of your records contain both of these fields, as the error says, so you must use two different connector configurations, one for each topic with their own PK.

---

<div class="post-metadata">

**Author:** ![sangeetha](https://sea1.discourse-cdn.com/flex019/user_avatar/forum.confluent.io/sangeetha/32/1935_2.png) [@sangeetha](https://forum.confluent.io/u/sangeetha)\
**Post date:** [26 May 2022 12:15 UTC](https://forum.confluent.io/t/jdbc-sink-connector-to-load-data-from-2-different-topics-to-2-different-table/5143/3 "2022-05-26T12:15:07Z")

</div>

**Json Schema for task topic:**  
{“type”:“object”,“properties”:{“orderid”:{“type”:“number”},“name”:{“type”:“string”},“products”:{“type”:“string”},“modeofpayment”:{“type”:“string”}}}

**json schema for member topic:**  
{“type”:“object”,“properties”:{“pzinskey”:{“type”:“number”},“membername”:{“type”:“string”},“type”:{“type”:“string”}}}

---

<div class="post-metadata">

**Author:** ![OneCricketeer](https://sea1.discourse-cdn.com/flex019/user_avatar/forum.confluent.io/onecricketeer/32/1551_2.png) [@OneCricketeer](https://forum.confluent.io/u/OneCricketeer)\
**Post date:** [26 May 2022 13:38 UTC](https://forum.confluent.io/t/jdbc-sink-connector-to-load-data-from-2-different-topics-to-2-different-table/5143/4 "2022-05-26T13:38:31Z")

</div>

As mentioned, every `pk.fields` value needs to exist in every one of the `topics`, so the solution here would be to create distinct connector configurations.

---

<div class="post-metadata">

**Author:** ![system](https://us1.discourse-cdn.com/flex019/uploads/confluentcommunity/original/1X/c49438c90c9df282e9996fdf6971be890c71b65a.svg) [@system](https://forum.confluent.io/u/system)\
**Post date:** [25 June 2022 13:38 UTC](https://forum.confluent.io/t/jdbc-sink-connector-to-load-data-from-2-different-topics-to-2-different-table/5143/5 "2022-06-25T13:38:44Z")

</div>

This topic was automatically closed 30 days after the last reply. New replies are no longer allowed.