Kafka Connect without schema, only JSON

Question

I want to use JDBC sink connector with JSON and without schema. They write (source):

If you need to use JSON without Schema Registry for Connect data, you can use the JsonConverter supported with Kafka. The example below shows the JsonConverter key and value properties that are added to the configuration:

key.converter=org.apache.kafka.connect.json.JsonConverter value.converter=org.apache.kafka.connect.json.JsonConverter key.converter.schemas.enable=false value.converter.schemas.enable=false

When the properties key.converter.schemas.enable and value.converter.schemas.enable are set to true, the key or value is not treated as plain JSON, but rather as a composite JSON object containing both an internal schema and the data. When these are enabled for a source connector, both the schema and data are in the composite JSON object. When these are enabled for a sink connector, the schema and data are extracted from the composite JSON object. Note that this implementation never uses Schema Registry.

When the properties key.converter.schemas.enable and value.converter.schemas.enable are set to false (the default), only the data is passed along, without the schema. This reduces the payload overhead for applications that do not need a schema.

I configured connector:

{
  "name": "noschemajustjson",
  "config": {
    "key.converter.schemas.enable": "false",
    "value.converter.schemas.enable": "false",
    "schemas.enable": "false",
    "name": "noschemajustjson",
    "connector.class": "io.confluent.connect.jdbc.JdbcSinkConnector",
    "key.converter": "org.apache.kafka.connect.json.JsonConverter",
    "value.converter": "org.apache.kafka.connect.json.JsonConverter",
    "config.action.reload": "restart",
    "errors.log.enable": "true",
    "errors.log.include.messages": "true",
    "topics": "testconnect2",
    "connection.url": "jdbc:postgresql://postgres:5432/postgres",
    "connection.user": "postgres",
    "connection.password": "********",
    "dialect.name": "PostgreSqlDatabaseDialect",
    "table.name.format": "utp",
    "auto.create": "false",
    "auto.evolve": "false"
  }
}

But I still get error:

Caused by: org.apache.kafka.connect.errors.ConnectException: Sink connector 'noschemajustjson2' is configured with 'delete.enabled=false' and 'pk.mode=none' and therefore requires records with a non-null Struct value and non-null Struct schema, but found record at (topic='testconnect2',partition=0,offset=0,timestamp=1626416739697) with a HashMap value and null value schema.

So what should I do to force Connect work without schema at all (only plain JSON)?

user_350613 · Answer

I want to use JDBC sink connector with JSON and without schema

You cannot do this - the JDBC Sink connector streams to a relational database, and relational databases have schemas :-D The JDBC Sink connector therefore requires a schema to be present for the data.

Depending on where your data is coming from you have different options.

If it's ingested from Kafka Connect, use a converter that supports schemas (Avro, Protobuf, JSON Schema)
If it's produced by an application that you have control over, get that application to serialise that data with a schema (Avro, Protobuf, JSON Schema)
If it's coming from somewhere you don't have control over then you'll need to pre-process the topic to add an explicit schema and write it to a new topic that is then consumed by the JDBC Sink connector.

References & resources:

Kafka Connect without schema, only JSON

1 Answers