Kafka Sink Connector Unsupported Source Data Type for Array Fields

Viewed 1211

Current Kafka Sink Connector have some array fields in the schema file

When I setup my sink connector here is the config

{"config": {
            "tasks.max:"2",
            "connector.class":"io.confluent.connect.jdbc.JdbcSinkConnector",
            "connection.url":"'"$URL"'",
            "topics": "topic_scheduled, topic_stream",
            "transforms":"ExtractField",
            "transforms.extractField.type":"org.apache.kafka.connect.transforms.ExtractField$Key",
            "transforms.extractField.field":"session_field",
            "errors.tolerance": "all",
            "errors.log.enable": true,
            "errors.log.include.messages": true
           }
}

schema file

{
 "name":"session_field",
 "type": ["null,string],
"default":null
},
{
  "name":"session_kind",
  "type": "string",
  "description":"map LinkedSession>SessionKind of session source"
}
{
"name": "session_type",
"type": [
  "null",
  {
    "type": "array",
    "items":"SessionType file"
  }
],
"default: null
}

DB Schema:

CREATE TABLE topic_one 
   row_id serial not null,
   session_field varchar(25),
   session_kind varchar(25),
   session_type varchar(25)
CONSTRAINT table_name_pkey PRIMARY KEY(row_id)

Sink connector status I get this error message

org.apache.kafka.connect.errors.ConnectException:
 Unsupported source data type:
ARRAY\n\tat io.confluent.connect.jdbc.dialect.GenericDatabaseDialect.bindField
(GenericDatabaseDialect.java:1436)\n\tat
io.confluent.connect.jdbc.sink.PreparedStatementBinder.bindField(PreparedStatementBinder.java:141)
\n\tat io.confluent.connect.jdbc.sink.PreparedStatementBinder.bindNonKeyFields
(PreparedStatementBinder.java:135)\n\tat
io.confluent.connect.jdbc.sink.PreparedStatementBinder.bindRecord(PreparedStatementBinder.java:71)\n
\tat io.confluent.connect.jdbc.sink.BufferedRecords.flush(BufferedRecords.java:139)\n\tat
io.confluent.connect.jdbc.sink.JdbcDbWriter.write(JdbcDbWriter.java:72)\n\tat
io.confluent.connect.jdbc.sink.JdbcSinkTask.put(JdbcSinkTask.java:74)\n\tat
org.apache.kafka.connect.runtime.WorkerSinkTask.deliverMessages(WorkerSinkTask.java:539)\n\tat
10more\n"

Based off the Kafka documentation regex transform fields documentation

0 Answers
Related