title: Importing json data into postgres using kafka jdbc sink connector
body: <p>I'm new to kafka.
  My requirement is to load json data from kafka topic to postgres table.</p>
  <p>Below is the table structure</p>
  <pre><code>CREATE TABLE dup_emp (
      emp_id integer PRIMARY KEY,
      emp_name text,
      emp_salary integer 
  );
  </code></pre>
  <p>Was producing messages in json format on to topic manually (i.e., from producer console).
  Example, <code>{&quot;emp_id&quot;:1,&quot;emp_name&quot;:&quot;bheem&quot;,&quot;emp_salary&quot;:2000}</code></p>
  <p>And my connector configuration as follows:-</p>
  <pre><code>curl -X PUT http://localhost:8083/connectors/load_test/config \
  -H &quot;Content-Type: application/json&quot; \
  -d '{
   &quot;connector.class&quot;: &quot;io.confluent.connect.jdbc.JdbcSinkConnector&quot;,
   &quot;connection.url&quot;:&quot;jdbc:postgresql://localhost:5432/somedb&quot;,
   &quot;connection.user&quot;:&quot;user&quot;,
   &quot;connection.password&quot;:&quot;passwd&quot;,
   &quot;key.converter&quot;:&quot;org.apache.kafka.connect.json.JsonConverter&quot;,
   &quot;value.converter&quot;:&quot;org.apache.kafka.connect.json.JsonConverter&quot;,
   &quot;key.converter.schemas.enable&quot;:&quot;false&quot;,
   &quot;value.converter.schemas.enable&quot;:&quot;false&quot;,
   &quot;tasks.max&quot; : &quot;1&quot;,
   &quot;topics&quot;:&quot;dup_emp&quot;,
   &quot;table.name.format&quot;:&quot;dup_emp&quot;,
   &quot;insert.mode&quot;:&quot;insert&quot;,
   &quot;quote.sql.identifiers&quot;:&quot;never&quot;
  }'
  </code></pre>
  <p>And was ending up with the following error:-</p>
  <p><strong>Caused by: org.apache.kafka.connect.errors.ConnectException: Sink connector 'load_test' 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='dup_emp',partition=0,offset=0,timestamp=1633066307312) with a HashMap value and null value schema.</strong></p>
  <p>As per my knowledge, when we use source connector, it will store schema in schema registry and used by  sink connector for validation. But, now as i'm not using source connector, as I was producing data in json format manually, how can I load data into postgres.</p>
  <p>Can someone, help me out in understanding the error in better way and suggest me a proper configuration to load data into postgres.</p>