title: Kafka connect using json and schema registry
body: <p>I'm trying to store json messages from Kafka topic to postgreSQL table using JdbcSinkConnector from Kafka Connect. This feature should be supported in <a href="https://www.confluent.io/blog/confluent-platform-now-supports-protobuf-json-schema-custom-formats/?d_utk=27bdeaab-d3a3-4655-a8bc-bfe4f32b4b4d" rel="nofollow noreferrer">Confluent Platform from version 5.5</a></p>
  <p>On producer side I'm using following serializer:</p>
  <pre><code>value.serial'sizer=io.confluent.kafka.serializers.json.KafkaJsonSchemaSerializer
  </code></pre>
  <p>Schema registry config:</p>
  <pre><code>{
    &quot;$schema&quot;: &quot;http://json-schema.org/draft-07/schema#&quot;,
    &quot;$id&quot;: &quot;http://example.com/myURI.schema.json&quot;,
    &quot;title&quot;: &quot;value_greeting&quot;,
    &quot;description&quot;: &quot;Sample schema to help you get started.&quot;,
    &quot;type&quot;: &quot;object&quot;,
    &quot;properties&quot;: {
      
      &quot;msg&quot;: {
        &quot;type&quot;: &quot;string&quot;,
        &quot;description&quot;: &quot;The string type is used for strings of text.&quot;
      },
      &quot;name&quot;: {
        &quot;type&quot;: &quot;string&quot;,
        &quot;description&quot;: &quot;The string type is used for strings of text.&quot;
      }  
    }
  }
  </code></pre>
  <p>Connectors's config:</p>
  <pre><code>&quot;config&quot;: {
      &quot;connector.class&quot;: &quot;io.confluent.connect.jdbc.JdbcSinkConnector&quot;,
      &quot;tasks.max&quot;: &quot;1&quot;,
      &quot;key.converter&quot;: &quot;org.apache.kafka.connect.storage.StringConverter&quot;,
      &quot;value.converter&quot;: &quot;io.confluent.connect.json.JsonSchemaConverter&quot;,
      &quot;topics&quot;: &quot;greeting&quot;,
      &quot;connection.url&quot;: &quot;jdbc:postgresql://postgres:5432/postgres&quot;,
      &quot;connection.user&quot;: &quot;postgres&quot;,
      &quot;connection.password&quot;: &quot;postgres&quot;,
      &quot;value.converter.schema.registry.url&quot;: &quot;http://schema-registry:8081&quot;
    }
  }
  </code></pre>
  <p>Unfortunately, the connector fails with the error:</p>
  <pre><code>ERROR WorkerSinkTask{id=GREE-0} Task threw an uncaught and unrecoverable exception (org.apache.kafka.connect.runtime.WorkerTask)
  org.apache.kafka.connect.errors.ConnectException: Tolerance exceeded in error handler
          at org.apache.kafka.connect.runtime.errors.RetryWithToleranceOperator.execAndHandleError(RetryWithToleranceOperator.java:178)
          at org.apache.kafka.connect.runtime.errors.RetryWithToleranceOperator.execute(RetryWithToleranceOperator.java:104)
          at org.apache.kafka.connect.runtime.WorkerSinkTask.convertAndTransformRecord(WorkerSinkTask.java:492)
          at org.apache.kafka.connect.runtime.WorkerSinkTask.convertMessages(WorkerSinkTask.java:469)
          at org.apache.kafka.connect.runtime.WorkerSinkTask.poll(WorkerSinkTask.java:325)
          at org.apache.kafka.connect.runtime.WorkerSinkTask.iteration(WorkerSinkTask.java:228)
          at org.apache.kafka.connect.runtime.WorkerSinkTask.execute(WorkerSinkTask.java:196)
          at org.apache.kafka.connect.runtime.WorkerTask.doRun(WorkerTask.java:184)
          at org.apache.kafka.connect.runtime.WorkerTask.run(WorkerTask.java:234)
          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:1149)
          at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
          at java.lang.Thread.run(Thread.java:748)
  Caused by: org.apache.kafka.connect.errors.DataException: Converting byte[] to Kafka Connect data failed due to serialization error:
          at io.confluent.connect.json.JsonSchemaConverter.toConnectData(JsonSchemaConverter.java:111)
          at org.apache.kafka.connect.storage.Converter.toConnectData(Converter.java:87)
          at org.apache.kafka.connect.runtime.WorkerSinkTask.lambda$convertAndTransformRecord$2(WorkerSinkTask.java:492)
          at org.apache.kafka.connect.runtime.errors.RetryWithToleranceOperator.execAndRetry(RetryWithToleranceOperator.java:128)
          at org.apache.kafka.connect.runtime.errors.RetryWithToleranceOperator.execAndHandleError(RetryWithToleranceOperator.java:162)
          ... 13 more
  Caused by: org.apache.kafka.common.errors.SerializationException: Error deserializing JSON message for id -1
  Caused by: org.apache.kafka.common.errors.SerializationException: Unknown magic byte!
  </code></pre>
  <p>What am i doing wrong?</p>
  <p>Thanks a lot!</p>