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