# JDBC Sink connector using DB Sequence for Primary key **URL:** \ **Category:** Self-Managed Connectors\ **Created:** [30 March 2021 11:01 UTC](https://forum.confluent.io/t/jdbc-sink-connector-using-db-sequence-for-primary-key/1341 "2021-03-30T11:01:29Z")\ **Posts on this page:** 3\ **Page:** 1
**Author:** ![sumitpratap999](https://avatars.discourse-cdn.com/v4/letter/s/97f17d/32.png) [@sumitpratap999](https://forum.confluent.io/u/sumitpratap999)\ **Post date:** [30 March 2021 11:01 UTC](https://forum.confluent.io/t/jdbc-sink-connector-using-db-sequence-for-primary-key/1341/1 "2021-03-30T11:01:29Z")
Hi All, I have one requirement to insert new records in data base using DB sequence to insert primary key(user\_id) in the table. I am using Confluent JDBC sink connector to produce data from Kafka topic. Except primary key, rest all values are coming from AVRO. I believe need to use “pk.mode”:“none”, but how to define sequence for the primary key while insert? Example table ```sql CREATE TABLE emp_msg_in ( user_id serial PRIMARY KEY, username VARCHAR ( 50 ) UNIQUE NOT NULL, password VARCHAR ( 50 ) NOT NULL, email VARCHAR ( 255 ) UNIQUE NOT NULL, created_on TIMESTAMP NOT NULL, last_login TIMESTAMP ); ``` Connector: ```json curl -X "POST" "http://localhost:18083/connectors/" -H "Content-Type: application/json" -d '{"name":"emp-decoder-msg-sink-db-connector","config":{ "connector.class":"io.confluent.connect.jdbc.JdbcSinkConnector", "input.data.format":"AVRO", "dialect.name":"PostgreSqlDatabaseDialect", "connection.url":"jdbc:postgresql://postgres:5432/postgres?", "connection.port":"5432","connection.user":"postgres", "connection.password":"password", "table.name.format":"emp_msg_in", "topics":"lxa-dev-us-emp-decoder-sink", "insert.mode":"INSERT", "db.timezone":"UTC", "auto.evolve":"true", "pk.mode":"none", "tasks.max":"1", "key.converter":"org.apache.kafka.connect.storage.StringConverter", "value.converter":"io.confluent.connect.avro.AvroConverter", "value.converter.schemas.enable":"true", "key.converter.schema.registry.url":"http://schema-registry:8081", "value.converter.schema.registry.url":"http://schema-registry:8081", "transforms":"t1,t2,t3", "transforms.t1.type":"org.apache.kafka.connect.transforms.Flatten$Value", "transforms.t2.type":"org.apache.kafka.connect.transforms.ReplaceField$Value", "transforms.t2.renames":"empHeader.messageTypeID:EMP_MSG_TYPE_ID,empHeader.owningScac:LOCO_OWNING_CUSTOMER,empHeader.flags:PROCESSED,empHeader.locoScac:LOCOMOTIVE_INITIAL,empHeader.roadNumber:ROAD_NUMBER,empHeader.source:EMP_SRC_ADDRESS,empHeader.destination:EMP_DEST_ADDRESS,empHeader.roadNumber:EMP_MESSAGE_NUMBER,empHeader.crc:MSG_CRC,empHeader.dataLength:MSG_SIZE", "transforms.t3.type":"org.apache.kafka.connect.transforms.ReplaceField$Value", "transforms.t3.whitelist":"messageTypeID,owningScac,flags,locoScac,roadNumber,source,destination,roadNumber,crc,dataLength"}}' ``` Schema used ```auto { "type": "record", "name": "EMP", "namespace": "com.wabtec.lxa.protocol.spec", "fields": [ { "name": "empHeader", "type": { "type": "record", "name": "EMPHeader", "fields": [ { "name": "protocolVersion", "type": "int" }, { "name": "messageTypeID", "type": "int" }, { "name": "messageVersion", "type": "int" }, { "name": "flags", "type": "int" }, { "name": "dataLength", "type": "int" }, { "name": "messageNumber", "type": "long" }, { "name": "messageTimeStamp", "type": "long" }, { "name": "dateTimeStamp", "type": { "type": "string", "avro.java.string": "String" }, "default": "" }, { "name": "variableHeaderSize", "type": "int" }, { "name": "timeToLive", "type": "long" }, { "name": "routingQos", "type": "int" }, { "name": "source", "type": { "type": "string", "avro.java.string": "String" } }, { "name": "destination", "type": { "type": "string", "avro.java.string": "String" } }, { "name": "crc", "type": "long" }, { "name": "owningScac", "type": { "type": "string", "avro.java.string": "String" }, "default": "" }, { "name": "locoScac", "type": { "type": "string", "avro.java.string": "String" }, "default": "" }, { "name": "roadNumber", "type": "int", "default": 0 } ] } }, { "name": "empBody", "type": { "type": "record", "name": "EMPBody", "fields": [ { "name": "payload", "type": { "type": "string", "avro.java.string": "String" } } ] } } ] } ``` Thanks, Sumit ---
**Author:** ![rmoff](https://sea1.discourse-cdn.com/flex019/user_avatar/forum.confluent.io/rmoff/32/3_2.png) [@rmoff](https://forum.confluent.io/u/rmoff)\ **Post date:** [30 March 2021 13:08 UTC](https://forum.confluent.io/t/jdbc-sink-connector-using-db-sequence-for-primary-key/1341/2 "2021-03-30T13:08:47Z")
Your table DDL doesn’t match your schema so I’m not _quite_ sure what you’re trying to do currently. However, in principle if you omit a field from the schema sent by the JDBC Sink to the target database it will rely on the default value, or that provided by the sequence. So in short; ensure that you drop the `user_id` field (using the [`ReplaceField` Single Message Transform](https://rmoff.net/2020/12/21/twelve-days-of-smt-day-10-replacefield/#_dropping_fields_in_a_sink_connector)) from the payload that you send, and postgres should use the next sequence value for the inserted record. ---
**Author:** ![sumitpratap999](https://avatars.discourse-cdn.com/v4/letter/s/97f17d/32.png) [@sumitpratap999](https://forum.confluent.io/u/sumitpratap999)\ **Post date:** [30 March 2021 15:13 UTC](https://forum.confluent.io/t/jdbc-sink-connector-using-db-sequence-for-primary-key/1341/3 "2021-03-30T15:13:14Z")
Thanks for help, it seems PK got auto generated even though I didn’t pass it from configuration.