# Kafka JDBC Source Connector- UPDATE

**URL:** <https://forum.confluent.io/t/kafka-jdbc-source-connector-update/10998>\
**Category:** Kafka Connect\
**Created:** [3 July 2024 23:33 UTC](https://forum.confluent.io/t/kafka-jdbc-source-connector-update/10998 "2024-07-03T23:33:45Z")\
**Posts on this page:** 9\
**Page:** 1

<div class="post-metadata">

**Author:** ![Anil](https://avatars.discourse-cdn.com/v4/letter/a/da6949/32.png) [@Anil](https://forum.confluent.io/u/Anil)\
**Post date:** [3 July 2024 23:33 UTC](https://forum.confluent.io/t/kafka-jdbc-source-connector-update/10998/1 "2024-07-03T23:33:45Z")

</div>

Hi,

I am new to Kafka JDBC Connect, working on Sales\_Order database. while working in insert through query its working fine. But I need to update some of records the same time , but this is not working.

I have taken Oracle DB for this.

for insertion, I have used below file.

{  
name= my-jdbc-connector2  
config= {  
name=kafkasource1  
connector.class=io.confluent.connect.jdbc.JdbcSourceConnector  
name=oracle-kafka  
connection.url=jdbc:oracle:thin:@localhost:1521/orcl  
connection.user=kafka  
connection.password=check  
task.max=5  
topic.prefix= kafkaop1  
mode= timestamp+incrementing  
query=SELECT APPLICATION\_NUMBER,CUSTOMER\_ID,PURCHASE\_ORDER\_NO,MATERIAL\_NO,ORDER\_TYPE,SUPPLIER,UNIT,SALESDATE,ADDRESS,MOBILENO,EMAIL\_ID FROM SALESORDER  
timestamp.column.name=SALESDATE  
incrementing.column.name=APPLICATION\_NUMBER  
poll.interval.ms=1000  
numeric.mapping=best\_fit  
errors.tolerance=all  
key.converter=org.apache.kafka.connect.storage.StringConverter  
value.converter=org.apache.kafka.connect.json.JsonConverter  
key.converter.schemas.enable=false  
value.converter.schemas.enable=true  
dialect.name=OracleDatabaseDialect  
}  
}

for insertion , select statement is working, where as I needs to work on same properties file for insertion, update and deletion.

Please , let me know for update I needs to change any query.

can I write two queries in same properties file.

to workable, I need to update any config changes in properties file.

its not sending any data to topic…

So, help to understand, is kafka jdbc source connector will work for this??

---

<div class="post-metadata">

**Author:** ![Anil](https://avatars.discourse-cdn.com/v4/letter/a/da6949/32.png) [@Anil](https://forum.confluent.io/u/Anil)\
**Post date:** [4 July 2024 09:37 UTC](https://forum.confluent.io/t/kafka-jdbc-source-connector-update/10998/2 "2024-07-04T09:37:32Z")

</div>

Hi,

Please, help me to understand how can we achieve, update events through kafka source connector. I have stuck up with this issue.

I needs to do both insert and update options with single query.

where as Insertition is working fine, Need to check with update entry. I am not getting any update. So, please help here.

I am using only Kafka jdbc source connector.instead of this, am i need to user any other connectors.

Thanks and Regards,  
A.

---

<div class="post-metadata">

**Author:** ![mmuehlbeyer](https://sea1.discourse-cdn.com/flex019/user_avatar/forum.confluent.io/mmuehlbeyer/32/1088_2.png) [@mmuehlbeyer](https://forum.confluent.io/u/mmuehlbeyer)\
**Post date:** [4 July 2024 10:48 UTC](https://forum.confluent.io/t/kafka-jdbc-source-connector-update/10998/3 "2024-07-04T10:48:38Z")

</div>

hey @Anil

have a look at the “mode” setting.

[https://docs.confluent.io/kafka-connectors/jdbc/current/source-connector/source\_config\_options.html#mode](https://docs.confluent.io/kafka-connectors/jdbc/current/source-connector/source_config_options.html#mode)

> **[JDBC Source Connector for Confluent Platform | Confluent Documentation](https://docs.confluent.io/kafka-connectors/jdbc/current/source-connector/overview.html#incremental-query-modes)**
>
> Get started with the JDBC Source connector for Confluent Platform.

just to have you on the same page some deep dive how it works.

> **[Kafka Connect Deep Dive – JDBC Source Connector | Confluent](https://www.confluent.io/blog/kafka-connect-deep-dive-jdbc-source-connector/)**
>
> The JDBC source connector for Kafka Connect enables you to pull data (source) from a database into Apache Kafka®, and to push data (sink) from a Kafka topic to a database. Almost all relational databases provide a JDBC driver, including Oracle,...

---

<div class="post-metadata">

**Author:** ![Anil](https://avatars.discourse-cdn.com/v4/letter/a/da6949/32.png) [@Anil](https://forum.confluent.io/u/Anil)\
**Post date:** [5 July 2024 06:44 UTC](https://forum.confluent.io/t/kafka-jdbc-source-connector-update/10998/4 "2024-07-05T06:44:53Z")

</div>

Hi mmuehlbeyer,

Thanks for sharing Information.

I need some information about update option.

I have taken Timestamp + incrementing mode. I am looking for updated rows only, New rows insertion is working fine.

* * *

mode= timestamp+incrementing

query=SELECT SALES\_ORDER, CAST(SO\_ITEM AS NUMBER(5,0)) AS SO\_ITEM, MATERIAL, CAST(QUANTITY AS NUMBER(5,2)) AS QUANTITY, UNIT\_OF\_MEASURE, REQUIRED\_DATE, SOLD\_TO\_PARTY\_NAME,SHIP\_TO\_PARTY\_NAME, SALES\_ORGANIZATION, DISTRIBUTION\_CHANNEL, PURCHASE\_ORDER\_NUMBER, PLANT,DESTINATION\_CITY, ORDER\_CREATION\_DATE, MODE\_OF\_TRANSPORT, CAST(STATUS\_FLAG AS NUMERIC(1,0)) AS STATUS\_FLAG, ERROR\_DESCRIPTION, DATE\_CREATED,DATE\_MODIFIED, PI\_PROCESS\_STATUS, CAST(SO\_OVERDELIVERY\_TOL AS NUMBER(5,0)) AS SO\_OVERDELIVERY\_TOL, CAST(SO\_UNDERDELIVERY\_TOL AS NUMBER(5,0)) AS SO\_UNDERDELIVERY\_TOL, ROUTE,ROUTE\_DESCRIPTION,UNLOADING\_POINT,RECEIVING\_POINT,ROUTE\_TREE,PROCESS\_PATH,MATERIAL\_TREE, YIELD\_STRING,RELEASE\_DATE, PROPOSED\_DELIVERY\_DATE, COMMITTED\_DATE FROM VJNR\_MES\_SO\_HEADER

timestamp.column.name=REQUIRED\_DATE  
incrementing.column.name=SO\_ITEM

* * *

We are updating rows as below.

UPDATE SOPOOB.VJNR\_MES\_SO\_HEADER

SET SALES\_ORDER=‘pqrs’

## WHERE DATE\_MODIFIED = TO\_DATE(‘23/04/24’, ‘DD/MM/YY’)

UPDATE SOPOOB.VJNR\_MES\_SO\_HEADER

SET SALES\_ORDER=‘pqrs’

WHERE SO\_ITEM=24;

we have not installed any trigger. Just config properties only taken through jdbc source connector.

I can able to see in Oracle, it was updating properly, but in topic I cant able to see any data is coming.

while insertion, I can able to see below commit log.

[2024-07-05 12:12:25,960] INFO [oracle-kafka|task-0|offsets] WorkerSourceTask{id=oracle-kafka-0} Committing offsets for 1 acknowledged messages (org.apache.kafka.connect.runtime.WorkerSourceTask:233)

and data is coming in topic..

while updating the record, I am unable to see any commit log is coming.  
Kindly, help me to understand more about Update query..

For insertion, I have used, select query, for update , am I need to write any query.

are we needs to create trigger for update/insert , please, let me know.

Kindly, help me to understand, am I missing anything..here.

Thanks and Regards,  
Anilkumar. S

---

<div class="post-metadata">

**Author:** ![Anil](https://avatars.discourse-cdn.com/v4/letter/a/da6949/32.png) [@Anil](https://forum.confluent.io/u/Anil)\
**Post date:** [5 July 2024 06:48 UTC](https://forum.confluent.io/t/kafka-jdbc-source-connector-update/10998/5 "2024-07-05T06:48:51Z")

</div>

my Table

CREATE TABLE “SOPOOB”.“VJNR\_MES\_SO\_HEADER”  
( “SALES\_ORDER” VARCHAR2(30),  
“SO\_ITEM” NUMBER(10,0),  
“MATERIAL” VARCHAR2(30),  
“QUANTITY” NUMBER(38,3),  
“UNIT\_OF\_MEASURE” VARCHAR2(30),  
“REQUIRED\_DATE” DATE,  
“SOLD\_TO\_PARTY\_NAME” VARCHAR2(40),  
“SHIP\_TO\_PARTY\_NAME” VARCHAR2(40),  
“SALES\_ORGANIZATION” VARCHAR2(40),  
“DISTRIBUTION\_CHANNEL” VARCHAR2(30),  
“PURCHASE\_ORDER\_NUMBER” VARCHAR2(35),  
“PLANT” VARCHAR2(30),  
“DESTINATION\_CITY” VARCHAR2(30),  
“ORDER\_CREATION\_DATE” DATE,  
“MODE\_OF\_TRANSPORT” VARCHAR2(40),  
“STATUS\_FLAG” NUMBER(38,0) NOT NULL ENABLE,  
“ERROR\_DESCRIPTION” VARCHAR2(32),  
“DATE\_CREATED” DATE,  
“DATE\_MODIFIED” DATE,  
“PI\_PROCESS\_STATUS” VARCHAR2(1),  
“SO\_OVERDELIVERY\_TOL” NUMBER,  
“SO\_UNDERDELIVERY\_TOL” NUMBER,  
“ROUTE” VARCHAR2(6),  
“ROUTE\_DESCRIPTION” VARCHAR2(50),  
“UNLOADING\_POINT” VARCHAR2(50),  
“RECEIVING\_POINT” VARCHAR2(50),  
“ROUTE\_TREE” VARCHAR2(300) DEFAULT ‘NA’ NOT NULL ENABLE,  
“PROCESS\_PATH” VARCHAR2(300) DEFAULT ‘NA’ NOT NULL ENABLE,  
“MATERIAL\_TREE” VARCHAR2(300) DEFAULT ‘NA’ NOT NULL ENABLE,  
“YIELD\_STRING” VARCHAR2(300),  
“RELEASE\_DATE” DATE,  
“PROPOSED\_DELIVERY\_DATE” DATE,  
“COMMITTED\_DATE” DATE  
)

Source connector

name=kafkasource  
connector.class=io.confluent.connect.jdbc.JdbcSourceConnector  
name=oracle-kafka  
connection.url=jdbc:oracle:thin:@localhot:1521/\<\>  
connection.user=XXXXX  
connection.password=XXXXXXXX  
task.max=1  
topic.prefix=orakafkapost8  
mode= timestamp+incrementing

query=SELECT SALES\_ORDER, CAST(SO\_ITEM AS NUMBER(5,0)) AS SO\_ITEM, MATERIAL, CAST(QUANTITY AS NUMBER(5,2)) AS QUANTITY, UNIT\_OF\_MEASURE, REQUIRED\_DATE, SOLD\_TO\_PARTY\_NAME,SHIP\_TO\_PARTY\_NAME, SALES\_ORGANIZATION, DISTRIBUTION\_CHANNEL, PURCHASE\_ORDER\_NUMBER, PLANT,DESTINATION\_CITY, ORDER\_CREATION\_DATE, MODE\_OF\_TRANSPORT, CAST(STATUS\_FLAG AS NUMERIC(1,0)) AS STATUS\_FLAG, ERROR\_DESCRIPTION, DATE\_CREATED,DATE\_MODIFIED, PI\_PROCESS\_STATUS, CAST(SO\_OVERDELIVERY\_TOL AS NUMBER(5,0)) AS SO\_OVERDELIVERY\_TOL, CAST(SO\_UNDERDELIVERY\_TOL AS NUMBER(5,0)) AS SO\_UNDERDELIVERY\_TOL, ROUTE,ROUTE\_DESCRIPTION,UNLOADING\_POINT,RECEIVING\_POINT,ROUTE\_TREE,PROCESS\_PATH,MATERIAL\_TREE, YIELD\_STRING,RELEASE\_DATE, PROPOSED\_DELIVERY\_DATE, COMMITTED\_DATE FROM VJNR\_MES\_SO\_HEADER

timestamp.column.name=REQUIRED\_DATE  
incrementing.column.name=SO\_ITEM

validate.non.null=true

poll.interval.ms=1000  
numeric.mapping=best\_fit  
errors.tolerance=all  
key.converter=org.apache.kafka.connect.storage.StringConverter  
value.converter=org.apache.kafka.connect.json.JsonConverter  
key.converter.schemas.enable=false  
value.converter.schemas.enable=false  
dialect.name=OracleDatabaseDialect

numeric.precision.mapping = true

transforms= REQUIRED\_DATE\_CONVERTER,ORDER\_CREATION\_DATE\_CONVERTER,DATE\_CREATED\_CONVERTER,DATE\_MODIFIED\_CONVERTER,RELEASE\_DATE\_CONVERTER,PROPOSED\_DELIVERY\_DATE\_CONVERTER, COMMITTED\_DATE\_CONVERTER  
transforms.REQUIRED\_DATE\_CONVERTER.type= org.apache.kafka.connect.transforms.TimestampConverter$Value  
transforms.REQUIRED\_DATE\_CONVERTER.format= yyyy-MM-dd  
transforms.REQUIRED\_DATE\_CONVERTER.target.type= string  
transforms.REQUIRED\_DATE\_CONVERTER.field= REQUIRED\_DATE

transforms.ORDER\_CREATION\_DATE\_CONVERTER.type= org.apache.kafka.connect.transforms.TimestampConverter$Value  
transforms.ORDER\_CREATION\_DATE\_CONVERTER.format= yyyy-MM-dd  
transforms.ORDER\_CREATION\_DATE\_CONVERTER.target.type= string  
transforms.ORDER\_CREATION\_DATE\_CONVERTER.field= ORDER\_CREATION\_DATE

transforms.DATE\_CREATED\_CONVERTER.type= org.apache.kafka.connect.transforms.TimestampConverter$Value  
transforms.DATE\_CREATED\_CONVERTER.format= yyyy-MM-dd  
transforms.DATE\_CREATED\_CONVERTER.target.type= string  
transforms.DATE\_CREATED\_CONVERTER.field= DATE\_CREATED

transforms.DATE\_MODIFIED\_CONVERTER.type= org.apache.kafka.connect.transforms.TimestampConverter$Value  
transforms.DATE\_MODIFIED\_CONVERTER.format= yyyy-MM-dd  
transforms.DATE\_MODIFIED\_CONVERTER.target.type= string  
transforms.DATE\_MODIFIED\_CONVERTER.field= DATE\_MODIFIED

transforms.RELEASE\_DATE\_CONVERTER.type= org.apache.kafka.connect.transforms.TimestampConverter$Value  
transforms.RELEASE\_DATE\_CONVERTER.format= yyyy-MM-dd  
transforms.RELEASE\_DATE\_CONVERTER.target.type= string  
transforms.RELEASE\_DATE\_CONVERTER.field= RELEASE\_DATE

transforms.PROPOSED\_DELIVERY\_DATE\_CONVERTER.type= org.apache.kafka.connect.transforms.TimestampConverter$Value  
transforms.PROPOSED\_DELIVERY\_DATE\_CONVERTER.format= yyyy-MM-dd  
transforms.PROPOSED\_DELIVERY\_DATE\_CONVERTER.target.type= string  
transforms.PROPOSED\_DELIVERY\_DATE\_CONVERTER.field= PROPOSED\_DELIVERY\_DATE

transforms.COMMITTED\_DATE\_CONVERTER.type= org.apache.kafka.connect.transforms.TimestampConverter$Value  
transforms.COMMITTED\_DATE\_CONVERTER.format= yyyy-MM-dd  
transforms.COMMITTED\_DATE\_CONVERTER.target.type= string  
transforms.COMMITTED\_DATE\_CONVERTER.field= COMMITTED\_DATE

---

<div class="post-metadata">

**Author:** ![mmuehlbeyer](https://sea1.discourse-cdn.com/flex019/user_avatar/forum.confluent.io/mmuehlbeyer/32/1088_2.png) [@mmuehlbeyer](https://forum.confluent.io/u/mmuehlbeyer)\
**Post date:** [5 July 2024 06:53 UTC](https://forum.confluent.io/t/kafka-jdbc-source-connector-update/10998/6 "2024-07-05T06:53:59Z")

</div>

hey,

thanks for the update.  
have look at for what you would like to do :

> **[JDBC Source Connector for Confluent Platform | Confluent Documentation](https://docs.confluent.io/kafka-connectors/jdbc/current/source-connector/overview.html#specifying-a-where-clause-with-query-modes)**
>
> Get started with the JDBC Source connector for Confluent Platform.

is the query needed? might it be enough to run the connector without the query config?

best,  
michael

---

<div class="post-metadata">

**Author:** ![Anil](https://avatars.discourse-cdn.com/v4/letter/a/da6949/32.png) [@Anil](https://forum.confluent.io/u/Anil)\
**Post date:** [5 July 2024 11:30 UTC](https://forum.confluent.io/t/kafka-jdbc-source-connector-update/10998/7 "2024-07-05T11:30:04Z")

</div>

Hi Michael,

for oracle table,

CREATE TABLE “KAFKA”.“EMP”  
( “EMPNO” NUMBER(5,0),  
“ENAME” VARCHAR2(15),  
“JOININGDATE” DATE,  
“RELEASEDATE” DATE,  
PRIMARY KEY (“EMPNO”)

for Postgres,

CREATE TABLE public.emp (

empno numeric(5) NOT NULL,

ename varchar(15) NULL,

joiningdate timestamp NULL,

releasedate timestamp NULL,

CONSTRAINT emp\_pkey PRIMARY KEY (empno)

);

JDBC Source Connector:- Oracle

{  
name= my-jdbc-connector  
config= {  
name=kafkasouece  
connector.class=io.confluent.connect.jdbc.JdbcSourceConnector  
name=oracle-kafka  
connection.url=jdbc:oracle:thin:@localhost:1521/orcl  
connection.user=kafka  
connection.password=check  
task.max=1  
topic.prefix= opkcheck125  
mode= timestamp+incrementing  
query=SELECT EMPNO,ENAME,JOININGDATE,RELEASEDATE FROM EMP  
timestamp.column.name=RELEASEDATE  
incrementing.column.name=EMPNO  
poll.interval.ms=1000  
numeric.mapping=best\_fit  
errors.tolerance=all

key.converter=org.apache.kafka.connect.storage.StringConverter  
value.converter=org.apache.kafka.connect.json.JsonConverter  
key.converter.schemas.enable=false  
value.converter.schemas.enable=true

dialect.name=OracleDatabaseDialect  
}  
}

JDBC Sink Connector - Postgres

{  
name=postgres-jdbc-sink-connector  
config={  
name= Kafka-sink-postgres  
connector.class=io.confluent.connect.jdbc.JdbcSinkConnector  
task.max=1  
topics=opkcheck125  
//JDBC Sink Connectors specific properties

connection.url=jdbc:postgresql://localhost:5432/postgres  
connection.user=postgres  
connection.password=admin

insert.mode=upsert

table.types=TABLE  
table.name.format=public.emp

auto.create= true  
auto.evolve= true

pk.mode=record\_value  
pk.fields=empno

transforms = RenameField  
transforms.RenameField.type = org.apache.kafka.connect.transforms.ReplaceField$Value  
transforms.RenameField.renames = EMPNO:empno,ENAME:ename,JOININGDATE:joiningdate,RELEASEDATE:releasedate

key.converter=org.apache.kafka.connect.storage.StringConverter  
value.converter=org.apache.kafka.connect.json.JsonConverter  
key.converter.schemas.enable=false  
value.converter.schemas.enable=true

dialect.name=PostgreSqlDatabaseDialect

db.timezone=Asia/Kolkata  
}  
}

I am updating at oracle side,

INSERT INTO emp values(11,‘venkat’,TO\_DATE(‘23/04/22’, ‘DD/MM/YY’),TO\_DATE(‘23/04/24’, ‘DD/MM/YY’));

SELECT \* FROM emp;

UPDATE emp

SET ENAME=‘Rajesh’

WHERE EMPNO=11;

Still, facing issue for update, I have taken timestamp+incremental, So, am i doing any wrong…please help to check and update me.

any action needs to take.

---

<div class="post-metadata">

**Author:** ![mmuehlbeyer](https://sea1.discourse-cdn.com/flex019/user_avatar/forum.confluent.io/mmuehlbeyer/32/1088_2.png) [@mmuehlbeyer](https://forum.confluent.io/u/mmuehlbeyer)\
**Post date:** [6 July 2024 15:08 UTC](https://forum.confluent.io/t/kafka-jdbc-source-connector-update/10998/8 "2024-07-06T15:08:11Z")

</div>

what happens if you leave the query setting empty?

---

<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:** [5 August 2024 15:09 UTC](https://forum.confluent.io/t/kafka-jdbc-source-connector-update/10998/9 "2024-08-05T15:09:01Z")

</div>

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