# Kafka connect - JDBC sink connectivity with KSQL **URL:** \ **Category:** ksqlDB\ **Created:** [7 February 2023 08:50 UTC](https://forum.confluent.io/t/kafka-connect-jdbc-sink-connectivity-with-ksql/7174 "2023-02-07T08:50:29Z")\ **Posts on this page:** 2\ **Page:** 1
**Author:** ![Dhruv](https://avatars.discourse-cdn.com/v4/letter/d/7c8e57/32.png) [@Dhruv](https://forum.confluent.io/u/Dhruv)\ **Post date:** [7 February 2023 08:50 UTC](https://forum.confluent.io/t/kafka-connect-jdbc-sink-connectivity-with-ksql/7174/1 "2023-02-07T08:50:29Z")
Hello Team, I am having an issue with JDBC sink connector to consume topic created by KSQL. below options I have tried to make it work: 1. with key and without key 2. with schema registry and with schema manually created 3. with AVRO and with JSON two types of Errors I am facing, with scenario 3 error looks like below ![error_byte](https://us1.discourse-cdn.com/flex019/uploads/confluentcommunity/original/2X/8/838248a50ccc953d385ac8b138c06c7be0e74f7e.png) with scenario 1 & 2 error says .DataException: JsonConverter with schemas.enable requires “schema” and “payload” fields and may not contain additional fields I followed few articles and videos simillar to my issue but none of them worked reference articles: 1. [🎥 ksqlDB and the Kafka Connect JDBC Sink - Kafka Connect / Self-Managed Connectors - Confluent Community](https://forum.confluent.io/t/ksqldb-and-the-kafka-connect-jdbc-sink/187) 2. [Can not write stream into jdbc sink unless key is set · Issue #3487 · confluentinc/ksql · GitHub](https://github.com/confluentinc/ksql/issues/3487) My configuration and topic given below, which I am trying to sink in Oracle database: 1. scenario without key: { “name”: “destination-connector-simple”, “config”: { “connector.class”: “io.confluent.connect.jdbc.JdbcSinkConnector”, “topics”: “MY\_STREAM1”, “tasks.max”: “1”, “connection.url”: “jdbc:oracle:thin:@oracle21:1521/orclpdb1”, “connection.user”: “c\_\_sinkuser”, “connection.password”: “sinkpw”, “table.name.format”: “kafka\_customers”, “auto.create”: “true”, “key.ignore”:“true”, “pk.mode”: “none”, “value.converter.schemas.enable”: “false”, “key.converter.schemas.enable”: “false”, “key.converter”: “org.apache.kafka.connect.storage.StringConverter” } } 2. scenario with key: { “name”: “oracle-sink”, “config”: { “connector.class”: “io.confluent.connect.jdbc.JdbcSinkConnector”, “tasks.max”: “1”, “topics”: “MY\_EMPLOYEE”, “table.name.format”: “kafka\_customers”, “connection.url”: “jdbc:oracle:thin:@oracle21:1521/orclpdb1”, “connection.user”: “c\_\_sinkuser”, “connection.password”: “sinkpw”, “auto.create”:true, “auto.evolve”:true, “pk.fields”: “ID”, “insert.mode”:“upsert”, “delete.enabled”:true, “delete.retention.ms”:100, “pk.mode”: “record\_key”, “key.converter”: “io.confluent.connect.json.JsonSchemaConverter”, “key.converter.schema.registry.url”: “h t tp 😕 / schema-registry :8081”, “value.converter”: “io.confluent.connect.json.JsonSchemaConverter”, “value.converter.schema.registry.url”: “htt p : / /schema-registry :8081” } } topic to be consumed in sink (Any one of them would be okay): 1. without key print ‘MY\_STREAM1’ from beginning; Key format: ¯\_(ツ)\_/¯ - no data processed Value format: JSON or KAFKA\_STRING rowtime: 2023/02/05 18:16:16.553 Z, key: , value: {“L\_EID”:“101”,“NAME”:“Dhruv”,“LNAME”:“S”,“L\_ADD\_ID”:“201”}, partition: 0 rowtime: 2023/02/05 18:16:16.554 Z, key: , value: {“L\_EID”:“102”,“NAME”:“Dhruv1”,“LNAME”:“S1”,“L\_ADD\_ID”:“202”}, partition: 0 2. topic with key: ksql\> print ‘MY\_EMPLOYEE’ from beginning; Key format: JSON or KAFKA\_STRING Value format: JSON or KAFKA\_STRING rowtime: 2023/02/05 18:16:16.553 Z, key: 101, value: {“EID”:“101”,“NAME”:“Dhruv”,“LNAME”:“S”,“ADD\_ID”:“201”}, partition: 0 rowtime: 2023/02/05 18:16:16.554 Z, key: 102, value: {“EID”:“102”,“NAME”:“Dhruv1”,“LNAME”:“S1”,“ADD\_ID”:“202”}, partition: 0 3. Topic with schema (manually created) ksql\> print ‘E\_SCHEMA’ from beginning; Key format: ¯\_(ツ)\_/¯ - no data processed Value format: JSON or KAFKA\_STRING rowtime: 2023/02/06 20:01:25.824 Z, key: , value: {“SCHEMA”:{“TYPE”:“struct”,“FIELDS”:[{“TYPE”:“int32”,“OPTIONAL”:false,“FIELD”:“L\_EID”},{“TYPE”:“int32”,“OPTIONAL”:false,“FIELD”:“NAME”},{“TYPE”:“int32”,“OPTIONAL”:false,“FIELD”:“LAME”},{“TYPE”:“int32”,“OPTIONAL”:false,“FIELD”:“L\_ADD\_ID”}],“OPTIONAL”:false,“NAME”:“”},“PAYLOAD”:{“L\_EID”:“201”,“NAME”:“Vishuddha”,“LNAME”:“Sh”,“L\_ADD\_ID”:“401”}}, partition: 0 4. Topic with Avro: ksql\> print ‘MY\_STREAM\_AVRO’ from beginning; Key format: ¯\_(ツ)\_/¯ - no data processed Value format: AVRO or KAFKA\_STRING rowtime: 2023/02/05 18:16:16.553 Z, key: , value: {“L\_EID”: “101”, “NAME”: “Dhruv”, “LNAME”: “S”, “L\_ADD\_ID”: “201”}, partition: 0 rowtime: 2023/02/05 18:16:16.554 Z, key: , value: {“L\_EID”: “102”, “NAME”: “Dhruv1”, “LNAME”: “S1”, “L\_ADD\_ID”: “202”}, partition: 0 rowtime: 2023/02/05 18:16:16.553 Z, key: , value: {“L\_EID”: “101”, “NAME”: “Dhruv”, “LNAME”: “S”, “L\_ADD\_ID”: “201”}, partition: 0 rowtime: 2023/02/05 18:16:16.554 Z, key: , value: {“L\_EID”: “102”, “NAME”: “Dhruv1”, “LNAME”: “S1”, “L\_ADD\_ID”: “202”}, partition: 0 could you please help me complete my POC in time. ---
**Author:** ![system](https://us1.discourse-cdn.com/flex019/uploads/confluentcommunity/original/1X/c49438c90c9df282e9996fdf6971be890c71b65a.svg) [@system](https://forum.confluent.io/u/system)\ **Post date:** [9 March 2023 08:50 UTC](https://forum.confluent.io/t/kafka-connect-jdbc-sink-connectivity-with-ksql/7174/2 "2023-03-09T08:50:52Z")
This topic was automatically closed after 30 days. New replies are no longer allowed.