Ignore updating missing fields with Confluent JDBC Connector

Viewed 465

Say I have a Postgres database table with the fields "id", "flavor" and "item". I also have a Kafka topic with two messages (let's ignore the kafka-key and assume the ID is in the value for now. Schema-definition is also omitted):

{"id": 1, "flavor": "chocolate"} 

{"id": 1, "item": "cookie"}

Now I'd like to use the Confluent JDBC (Sink) Connector for persisting the kafka messages in UPSERT-mode, hoping to get to the following end result in the database:

id | flavor    | item
----------------------
1  | chocolate | cookie

Ẁhat I did get, however, is was this:

id | flavor    | item
----------------------
1  | null      | cookie

I assume that's because the second message uses an UPDATE-statement where it infers the "null" values for fields that weren't provided, and writes those null-values over my actual data.

Is there a way to get to my desired result by changing the configuration of either the Confluent JDBC Connector or PostgreSQL 12? Failing that, is there another reasonably supported postgreSQL compatible connector out there that can do this?


Here's my connector configuration (connection details obviously redacted):

{
    "name": "sink-jdbc-upsertstest",
        "config": {
            "connector.class": "io.confluent.connect.jdbc.JdbcSinkConnector",
            "tasks.max": "1",
            "topics": "TEST-upserts",           
                        
            "connection.url": "jdbc:postgresql://host:port/database",
            "connection.user": "user",
            "connection.password": "password",
            
            "dialect.name": "ExtendedPostgreSqlDatabaseDialect",
            "table.name.format": "upsert-test",
            
            "batch.size": "100",
            "insert.mode": "upsert",
            "auto.evolve": "true",
            "auto.create": "true",
            "pk.mode": "record_value",
            "pk.fields": "id",
                        
            "key.converter": "org.apache.kafka.connect.storage.StringConverter",
            
            "value.converter": "org.apache.kafka.connect.json.JsonConverter",
            "value.converter.schemas.enable": "true",
            
            "errors.deadletterqueue.topic.name": "dlq_upserts",
            "errors.tolerance": "all",          
            "errors.deadletterqueue.topic.replication.factor": "1",
            "errors.deadletterqueue.context.headers.enable": "true"
        }
}
0 Answers
Related