title: How to disable JSON schema in Kafka Source Connector (e.g. Debezium)
body: <p>I followed Debezium tutorial (<a href="https://github.com/debezium/debezium-examples/tree/master/tutorial#using-postgres" rel="nofollow noreferrer">https://github.com/debezium/debezium-examples/tree/master/tutorial#using-postgres</a>) and all received CDC data from Postgres are sent to Kafka topic in JSON format with schema - <strong>how to get rid of schema?</strong></p>
  <p>Here is config of connector (launched in Docker container)</p>
  <pre><code>{
      &quot;name&quot;: &quot;inventory-connector&quot;,
      &quot;config&quot;: {
          &quot;connector.class&quot;: &quot;io.debezium.connector.postgresql.PostgresConnector&quot;,
          &quot;tasks.max&quot;: &quot;1&quot;,
          &quot;key.converter.schemas.enable&quot;: &quot;false&quot;,
          &quot;value.converter.schemas.enable&quot;: &quot;false&quot;,
          &quot;database.hostname&quot;: &quot;postgres&quot;,
          &quot;database.port&quot;: &quot;5432&quot;,
          &quot;database.user&quot;: &quot;postgres&quot;,
          &quot;database.password&quot;: &quot;postgres&quot;,
          &quot;database.dbname&quot; : &quot;postgres&quot;,
          &quot;database.server.name&quot;: &quot;dbserver1&quot;,
          &quot;schema.include&quot;: &quot;inventory&quot;
      }
  }
  </code></pre>
  <p>The JSON schema is still in message.
  I managed to get rid of it only when launched Docker container with following environment variables:</p>
  <pre><code> - CONNECT_KEY_CONVERTER_SCHEMAS_ENABLE=false
   - CONNECT_VALUE_CONVERTER_SCHEMAS_ENABLE=false
  </code></pre>
  <p><strong>Why I cannot achieve exactly the same from connector configuration?</strong></p>
  <p>Example of Kafka message with schema:</p>
  <pre><code>{&quot;schema&quot;:{&quot;type&quot;:&quot;struct&quot;,&quot;fields&quot;:[{&quot;type&quot;:&quot;int32&quot;,&quot;optional&quot;:false,&quot;field&quot;:&quot;id&quot;}],&quot;optional&quot;:false,&quot;name&quot;:&quot;dbserver1.inventory.customers.Key&quot;},&quot;payload&quot;:{&quot;id&quot;:1001}}    {&quot;schema&quot;:{&quot;type&quot;:&quot;struct&quot;,&quot;fields&quot;:[{&quot;type&quot;:&quot;struct&quot;,&quot;fields&quot;:[{&quot;type&quot;:&quot;int32&quot;,&quot;optional&quot;:false,&quot;field&quot;:&quot;id&quot;},{&quot;type&quot;:&quot;string&quot;,&quot;optional&quot;:false,&quot;field&quot;:&quot;first_name&quot;},{&quot;type&quot;:&quot;string&quot;,&quot;optional&quot;:false,&quot;field&quot;:&quot;last_name&quot;},{&quot;type&quot;:&quot;string&quot;,&quot;optional&quot;:false,&quot;field&quot;:&quot;email&quot;}],&quot;optional&quot;:true,&quot;name&quot;:&quot;dbserver1.inventory.customers.Value&quot;,&quot;field&quot;:&quot;before&quot;},{&quot;type&quot;:&quot;struct&quot;,&quot;fields&quot;:[{&quot;type&quot;:&quot;int32&quot;,&quot;optional&quot;:false,&quot;field&quot;:&quot;id&quot;},{&quot;type&quot;:&quot;string&quot;,&quot;optional&quot;:false,&quot;field&quot;:&quot;first_name&quot;},{&quot;type&quot;:&quot;string&quot;,&quot;optional&quot;:false,&quot;field&quot;:&quot;last_name&quot;},{&quot;type&quot;:&quot;string&quot;,&quot;optional&quot;:false,&quot;field&quot;:&quot;email&quot;}],&quot;optional&quot;:true,&quot;name&quot;:&quot;dbserver1.inventory.customers.Value&quot;,&quot;field&quot;:&quot;after&quot;},{&quot;type&quot;:&quot;struct&quot;,&quot;fields&quot;:[{&quot;type&quot;:&quot;string&quot;,&quot;optional&quot;:false,&quot;field&quot;:&quot;version&quot;},{&quot;type&quot;:&quot;string&quot;,&quot;optional&quot;:false,&quot;field&quot;:&quot;connector&quot;},{&quot;type&quot;:&quot;string&quot;,&quot;optional&quot;:false,&quot;field&quot;:&quot;name&quot;},{&quot;type&quot;:&quot;int64&quot;,&quot;optional&quot;:false,&quot;field&quot;:&quot;ts_ms&quot;},{&quot;type&quot;:&quot;string&quot;,&quot;optional&quot;:true,&quot;name&quot;:&quot;io.debezium.data.Enum&quot;,&quot;version&quot;:1,&quot;parameters&quot;:{&quot;allowed&quot;:&quot;true,last,false&quot;},&quot;default&quot;:&quot;false&quot;,&quot;field&quot;:&quot;snapshot&quot;},{&quot;type&quot;:&quot;string&quot;,&quot;optional&quot;:false,&quot;field&quot;:&quot;db&quot;},{&quot;type&quot;:&quot;string&quot;,&quot;optional&quot;:false,&quot;field&quot;:&quot;schema&quot;},{&quot;type&quot;:&quot;string&quot;,&quot;optional&quot;:false,&quot;field&quot;:&quot;table&quot;},{&quot;type&quot;:&quot;int64&quot;,&quot;optional&quot;:true,&quot;field&quot;:&quot;txId&quot;},{&quot;type&quot;:&quot;int64&quot;,&quot;optional&quot;:true,&quot;field&quot;:&quot;lsn&quot;},{&quot;type&quot;:&quot;int64&quot;,&quot;optional&quot;:true,&quot;field&quot;:&quot;xmin&quot;}],&quot;optional&quot;:false,&quot;name&quot;:&quot;io.debezium.connector.postgresql.Source&quot;,&quot;field&quot;:&quot;source&quot;},{&quot;type&quot;:&quot;string&quot;,&quot;optional&quot;:false,&quot;field&quot;:&quot;op&quot;},{&quot;type&quot;:&quot;int64&quot;,&quot;optional&quot;:true,&quot;field&quot;:&quot;ts_ms&quot;},{&quot;type&quot;:&quot;struct&quot;,&quot;fields&quot;:[{&quot;type&quot;:&quot;string&quot;,&quot;optional&quot;:false,&quot;field&quot;:&quot;id&quot;},{&quot;type&quot;:&quot;int64&quot;,&quot;optional&quot;:false,&quot;field&quot;:&quot;total_order&quot;},{&quot;type&quot;:&quot;int64&quot;,&quot;optional&quot;:false,&quot;field&quot;:&quot;data_collection_order&quot;}],&quot;optional&quot;:true,&quot;field&quot;:&quot;transaction&quot;}],&quot;optional&quot;:false,&quot;name&quot;:&quot;dbserver1.inventory.customers.Envelope&quot;},&quot;payload&quot;:{&quot;before&quot;:null,&quot;after&quot;:{&quot;id&quot;:1001,&quot;first_name&quot;:&quot;Sally&quot;,&quot;last_name&quot;:&quot;Thomas&quot;,&quot;email&quot;:&quot;sally.thomas@acme.com&quot;},&quot;source&quot;:{&quot;version&quot;:&quot;1.4.1.Final&quot;,&quot;connector&quot;:&quot;postgresql&quot;,&quot;name&quot;:&quot;dbserver1&quot;,&quot;ts_ms&quot;:1611918971029,&quot;snapshot&quot;:&quot;true&quot;,&quot;db&quot;:&quot;postgres&quot;,&quot;schema&quot;:&quot;inventory&quot;,&quot;table&quot;:&quot;customers&quot;,&quot;txId&quot;:602,&quot;lsn&quot;:34078720,&quot;xmin&quot;:null},&quot;op&quot;:&quot;r&quot;,&quot;ts_ms&quot;:1611918971032,&quot;transaction&quot;:null}}
  </code></pre>
  <p>Example (<strong>desired</strong> by me) w/o schema:</p>
  <pre><code>{&quot;id&quot;:1001} {&quot;before&quot;:null,&quot;after&quot;:{&quot;id&quot;:1001,&quot;first_name&quot;:&quot;Sally&quot;,&quot;last_name&quot;:&quot;Thomas&quot;,&quot;email&quot;:&quot;sally.thomas@acme.com&quot;},&quot;source&quot;:{&quot;version&quot;:&quot;1.4.1.Final&quot;,&quot;connector&quot;:&quot;postgresql&quot;,&quot;name&quot;:&quot;dbserver1&quot;,&quot;ts_ms&quot;:1611920304594,&quot;snapshot&quot;:&quot;true&quot;,&quot;db&quot;:&quot;postgres&quot;,&quot;schema&quot;:&quot;inventory&quot;,&quot;table&quot;:&quot;customers&quot;,&quot;txId&quot;:597,&quot;lsn&quot;:33809448,&quot;xmin&quot;:null},&quot;op&quot;:&quot;r&quot;,&quot;ts_ms&quot;:1611920304596,&quot;transaction&quot;:null}
  </code></pre>
  <p>Debezium container is run with following command:</p>
  <pre><code>docker run -it --name connect -p 8083:8083 -e GROUP_ID=1 -e CONFIG_STORAGE_TOPIC=my_connect_configs -e OFFSET_STORAGE_TOPIC=my_connect_offsets -e STATUS_STORAGE_TOPIC=my_connect_statuses -e CONNECT_KEY_CONVERTER_SCHEMAS_ENABLE=false -e CONNECT_VALUE_CONVERTER_SCHEMAS_ENABLE=false --link zookeeper:zookeeper --link kafka:kafka --link mysql:mysql debezium/connect:1.3
  </code></pre>
  <p>or as docker-compose</p>
  <pre><code>  connect:
      image: debezium/connect:${DEBEZIUM_VERSION}
      ports:
       - 8083:8083
      links:
       - kafka
       - postgres
      environment:
       - BOOTSTRAP_SERVERS=kafka:9092
       - GROUP_ID=1
       - CONFIG_STORAGE_TOPIC=my_connect_configs
       - OFFSET_STORAGE_TOPIC=my_connect_offsets
       - STATUS_STORAGE_TOPIC=my_connect_statuses
       - CONNECT_KEY_CONVERTER_SCHEMAS_ENABLE=false
       - CONNECT_VALUE_CONVERTER_SCHEMAS_ENABLE=false
  </code></pre>
  <p><code>CONNECT_KEY_CONVERTER_SCHEMAS_ENABLE=false</code> and <code>CONNECT_VALUE_CONVERTER_SCHEMAS_ENABLE=false</code> was added later by me, but without them I cannot get rid of schema.</p>
  <p><code>connect</code> docker container (Kafka connectors servers cluster - if I understood it correctly) is started without any connector.
  I create it manually.</p>
  <p>LOGs from docker-compose for connect when connector created</p>
  <pre><code>connect_1    | 2021-01-29 18:04:57,395 INFO   ||  JsonConverterConfig values: 
  connect_1    |  converter.type = key
  connect_1    |  decimal.format = BASE64
  connect_1    |  schemas.cache.size = 1000
  connect_1    |  schemas.enable = true
  connect_1    |    [org.apache.kafka.connect.json.JsonConverterConfig]
  connect_1    | 2021-01-29 18:04:57,396 INFO   ||  Set up the key converter class org.apache.kafka.connect.json.JsonConverter for task inventory-connector-0 using the worker config   [org.apache.kafka.connect.runtime.Worker]
  connect_1    | 2021-01-29 18:04:57,396 INFO   ||  JsonConverterConfig values: 
  connect_1    |  converter.type = value
  connect_1    |  decimal.format = BASE64
  connect_1    |  schemas.cache.size = 1000
  connect_1    |  schemas.enable = true
  connect_1    |    [org.apache.kafka.connect.json.JsonConverterConfig]
  ...
  connect_1    | 2021-01-29 18:04:57,458 INFO   ||  Starting PostgresConnectorTask with configuration:   [io.debezium.connector.common.BaseSourceTask]
  connect_1    | 2021-01-29 18:04:57,460 INFO   ||     key.converter.schemas.enable = false   [io.debezium.connector.common.BaseSourceTask]
  connect_1    | 2021-01-29 18:04:57,460 INFO   ||     value.converter.schemas.enable = false   [io.debezium.connector.common.BaseSourceTask]
  </code></pre>
  <p>Here are get connector command output:</p>
  <pre><code>$ curl -i http://localhost:8083/connectors/inventory-connector
  
  {&quot;name&quot;:&quot;inventory-connector&quot;,&quot;config&quot;:{&quot;connector.class&quot;:&quot;io.debezium.connector.postgresql.PostgresConnector&quot;,**&quot;key.converter.schemas.enable&quot;:&quot;false&quot;**,&quot;database.user&quot;:&quot;postgres&quot;,&quot;database.dbname&quot;:&quot;postgres&quot;,&quot;tasks.max&quot;:&quot;1&quot;,&quot;database.hostname&quot;:&quot;postgres&quot;,&quot;database.password&quot;:&quot;postgres&quot;,**&quot;value.converter.schemas.enable&quot;:&quot;false&quot;**,&quot;name&quot;:&quot;inventory-connector&quot;,&quot;database.server.name&quot;:&quot;dbserver1&quot;,&quot;database.port&quot;:&quot;5432&quot;,&quot;schema.include&quot;:&quot;inventory&quot;},&quot;tasks&quot;:[{&quot;connector&quot;:&quot;inventory-connector&quot;,&quot;task&quot;:0}],&quot;type&quot;:&quot;source&quot;}
  </code></pre>