Kafka-connect-jdbc: JDBC sink fails to Postgres database

[kccheung]

I encounter the following error when I sink a topic to my postgres database:

org.apache.kafka.connect.errors.ConnectException: Exiting WorkerSinkTask due to unrecoverable exception.
at org.apache.kafka.connect.runtime.WorkerSinkTask.deliverMessages(WorkerSinkTask.java:546)
at org.apache.kafka.connect.runtime.WorkerSinkTask.poll(WorkerSinkTask.java:302)
at org.apache.kafka.connect.runtime.WorkerSinkTask.iteration(WorkerSinkTask.java:205)
at org.apache.kafka.connect.runtime.WorkerSinkTask.execute(WorkerSinkTask.java:173)
at org.apache.kafka.connect.runtime.WorkerTask.doRun(WorkerTask.java:170)
at org.apache.kafka.connect.runtime.WorkerTask.run(WorkerTask.java:214)
at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
at java.util.concurrent.FutureTask.run(FutureTask.java:266)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
at java.lang.Thread.run(Thread.java:745)
used by: java.lang.ClassCastException: java.util.HashMap cannot be cast to org.apache.kafka.connect.data.Struct
at io.confluent.connect.jdbc.sink.PreparedStatementBinder.bindRecord(PreparedStatementBinder.java:61)
at io.confluent.connect.jdbc.sink.BufferedRecords.flush(BufferedRecords.java:116)
at io.confluent.connect.jdbc.sink.JdbcDbWriter.write(JdbcDbWriter.java:69)
at io.confluent.connect.jdbc.sink.JdbcSinkTask.put(JdbcSinkTask.java:69)
at org.apache.kafka.connect.runtime.WorkerSinkTask.deliverMessages(WorkerSinkTask.java:524)
... 10 more

My sink connector config:

{
    "name": "pg_sink_S_COUNTS_VA_INTERMEDIATE",
    "config": {
        "topics": "S_COUNTS_VA_INTERMEDIATE",
        "tasks.max": 1,
        "connector.class": "io.confluent.connect.jdbc.JdbcSinkConnector",
        "key.converter": "org.apache.kafka.connect.storage.StringConverter",
        "key.ignore": "true",
        "schema.ignore": "true",
        "schemas.enable": "false",
        "value.converter": "org.apache.kafka.connect.json.JsonConverter",
        "value.converter.schemas.enable": "false",
        "type.name": "kafka-connect",
        "connection.url": "jdbc:postgresql://*****.com:5433/kafka-sink",
        "connection.user": "****",
        "connection.password": "*****",
        "insert.mode": "upsert",
        "auto.create": true,
        "auto.evolve": true,
        "pk.mode": "kafka",
        "pk.fields": "__connect_topic,__connect_partition,__connect_offset",
        "batch.size": 30
    }
}

ENVIRONMENT variables of my kafka-connect container:

      CONNECT_KEY_CONVERTER: "org.apache.kafka.connect.storage.StringConverter"
      CONNECT_VALUE_CONVERTER: "org.apache.kafka.connect.json.JsonConverter"
      CONNECT_INTERNAL_KEY_CONVERTER: "org.apache.kafka.connect.json.JsonConverter"
      CONNECT_INTERNAL_VALUE_CONVERTER: "org.apache.kafka.connect.json.JsonConverter"
      CONNECT_REST_ADVERTISED_HOST_NAME: kafka-connect
      CONNECT_REST_PORT: 8083
      CONNECT_LOG4J_ROOT_LOGLEVEL: INFO
      CONNECT_PLUGIN_PATH: /usr/share/java

I have a separate producer which can produce JSON data to Kafka without schema defined.

[rmoff]

"value.converter": "org.apache.kafka.connect.json.JsonConverter",
"value.converter.schemas.enable": "false",

The JDBC Sink requires a schema to your data. I'm not sure why this is triggering the error you're seeing, but you definitely to include a schema, either as part of your JSON or using Avro. See https://www.confluent.io/blog/kafka-connect-deep-dive-converters-serialization-explained

[rhauch]

We'll keep this open so that we catch this earlier and give a much better exception message.

[krishkir]

Facing the same issue with mongo as source and mysql as sink.Request your help..
"value.converter": "org.apache.kafka.connect.json.JsonConverter",
"value.converter.schemas.enable": "false"

[fabiotc]

And if I have a schemaless origin topic, is it possible to create a separated Avro file for the topic data and use it to be able to have a well-defined schema?

[OneCricketeer]

@krishkir As explained, JDBC sink connector requires schemas to be enabled

@fabiotc You'd have to write a custom Connect SMT or other processor that could parse your topic and return a new record with a schema applied

[fabiotc]

Thanks @cricket007

[rmoff]

Some notes on this: https://rmoff.net/2020/01/22/kafka-connect-classcastexception/

[yousufdev]

@cricket007 Thanks for the suggestion on writing a custom SMT. @fabiotc here is a SMT I wrote for appending schema to a record https://github.com/yousufdev/kafka-connect-append-schema, hope it helps.

[fabiotc]

thanks @rmoff
@yousufdev thanks for this! And using your example, Is it possible to append a schema into a complex and nested JSON structure?

[yousufdev]

@fabiotc yes you can append schema into a complex structure, just pass it in the 'schema' property of the SMT.