Kafka-connect-jdbc: kafka jdbc sink connector to mysql problem

[qiutianhc]

version：kafka2.0
mysql to kafka(producer)：debezium
kafka to mysql(consumer)：jdbc sink connector

the parameters set:

key.converter.schemas.enable=true
value.converter.schemas.enable=true

the kafka data is:

{
  "schema": {
    "type": "struct",
    "fields": [
      {
        "type": "struct",
        "fields": [
          {
            "type": "int32",
            "optional": false,
            "field": "id"
          },
          {
            "type": "string",
            "optional": true,
            "field": "name"
          }
        ],
        "optional": true,
        "name": "test.aaa.bbb.Value",
        "field": "before"
      },
      {
        "type": "struct",
        "fields": [
          {
            "type": "int32",
            "optional": false,
            "field": "id"
          },
          {
            "type": "string",
            "optional": true,
            "field": "name"
          }
        ],
        "optional": true,
        "name": "test.aaa.bbb.Value",
        "field": "after"
      },
      {
        "type": "struct",
        "fields": [
          {
            "type": "string",
            "optional": true,
            "field": "version"
          },
          {
            "type": "string",
            "optional": false,
            "field": "name"
          },
          {
            "type": "int64",
            "optional": false,
            "field": "server_id"
          },
          {
            "type": "int64",
            "optional": false,
            "field": "ts_sec"
          },
          {
            "type": "string",
            "optional": true,
            "field": "gtid"
          },
          {
            "type": "string",
            "optional": false,
            "field": "file"
          },
          {
            "type": "int64",
            "optional": false,
            "field": "pos"
          },
          {
            "type": "int32",
            "optional": false,
            "field": "row"
          },
          {
            "type": "boolean",
            "optional": true,
            "default": false,
            "field": "snapshot"
          },
          {
            "type": "int64",
            "optional": true,
            "field": "thread"
          },
          {
            "type": "string",
            "optional": true,
            "field": "db"
          },
          {
            "type": "string",
            "optional": true,
            "field": "table"
          },
          {
            "type": "string",
            "optional": true,
            "field": "query"
          }
        ],
        "optional": false,
        "name": "io.debezium.connector.mysql.Source",
        "field": "source"
      },
      {
        "type": "string",
        "optional": false,
        "field": "op"
      },
      {
        "type": "int64",
        "optional": true,
        "field": "ts_ms"
      }
    ],
    "optional": false,
    "name": "test.aaa.bbb.Envelope"
  },
  "payload": {
    "before": null,
    "after": {
      "id": 1,
      "name": "a"
    },
    "source": {
      "version": "0.8.3.Final",
      "name": "test",
      "server_id": 0,
      "ts_sec": 0,
      "gtid": null,
      "file": "mysql-bin.000006",
      "pos": 485,
      "row": 0,
      "snapshot": true,
      "thread": null,
      "db": "aaa",
      "table": "bbb",
      "query": null
    },
    "op": "c",
    "ts_ms": 1546945519731
  }
}

the sink connector run error:

Caused by: org.apache.kafka.connect.errors.ConnectException: test.aaa.bbb.Value (STRUCT) type doesn't have a mapping to the SQL database column type

[rmoff]

AFAIK you can't write nested structures with the JDBC Sink. You need to use a Single Message Transform (which Debezium provides) to flatten the data

e.g.

            [existing connector config 
              …
             ]
            "transforms": "unwrap",
            "transforms.unwrap.type": "io.debezium.transforms.UnwrapFromEnvelope"

[holgerbrandl]

@rmoff Would this transform also be supported when using io.confluent.connect.jdbc.JdbcSinkConnector or do would we need to use debezium in order to use it?

UnwrapFromEnvolope is deprecated, it has been replaced with ExtractNewRecordState. However even after reading its docs, it's still unclear to me if it would transform a struct into a json-payload (which could be handled structurally e.g. in Postgres)?

[jinlunjie]

nested structures

What do I need to do if I want to keep both before and after payload. ExtractNewRecordState only gives me the new state

[rmoff]

Use the Flatten SMT:

'transforms'          = 'flatten',
'transforms.flatten.type'= 'org.apache.kafka.connect.transforms.Flatten$Value'

[jinlunjie]

flatten will flat all the nested all the way till the end, where I just want a "before" blob and an "after" blob

[rmoff]

Use a stream processor such as ksqlDB or Kafka Streams.

[jinlunjie]

Hi rmoff,

im having issue for type "Array" I have a json schema field of ARRAY[STRING], im getting error of
Caused by: org.apache.kafka.connect.errors.ConnectException: null (ARRAY) type doesn't have a mapping to the SQL database column type
any workarounds?

[katty0924]

hi jinlunjie，may i ask you that question is figured out? i also wanna try to user the mysql to kafka(producer)：debezium
kafka to mysql(consumer)：jdbc sink connector, but it seems like that jdbc sink connector cann't parse the data from debezium

[jinlunjie]

I ended up forking their JDBC sink and converts incoming json into JSONB in PG.

[rmoff]

@jinlunjie that sounds interesting, have you considered a PR back into the original project?

[tahseen447]

@jinlunjie can you give me the location for your forked repo and is it okay if i try it? its pretty much what i want. nested object to be saved as jsonb in postgres