title: Why record cannot get deleted using JDBC Sink Connector in Kafka Connect
body: <p><strong>my sink properties :</strong></p>
  
  <pre><code>{
    "name": "jdbc-oracle",
    "config": {
      "connector.class": "io.confluent.connect.jdbc.JdbcSinkConnector",
      "tasks.max": "1",
      "topics": "orders",
      "connection.url": "jdbc:oracle:thin:@10.1.2.3:1071/orac",
      "connection.user": "ersin",
      "connection.password": "ersin!",
      "auto.create": "true",
      "delete.enabled": "true",
      "pk.mode": "record_key",
      "pk.fields": "id",
      "insert.mode": "upsert",
      "plugin.path": "/home/ersin/confluent-5.4.1/share/java/",
      "name": "jdbc-oracle"
    },
    "tasks": [
      {
        "connector": "jdbc-oracle",
        "task": 0
      }
    ],
    "type": "sink"
  }
  </code></pre>
  
  <p><strong>my connect-avro-distributed.properties</strong>   :</p>
  
  <pre><code>bootstrap.servers=10.0.0.0:9092
  
  group.id=connect-cluster
  
  key.converter=io.confluent.connect.avro.AvroConverter
  key.converter.schema.registry.url=http://10.0.0.0:8081
  value.converter=io.confluent.connect.avro.AvroConverter
  value.converter.schema.registry.url=http://10.0.0.0:8081
  
  config.storage.topic=connect-configs
  offset.storage.topic=connect-offsets
  status.storage.topic=connect-statuses
  
  config.storage.replication.factor=1
  offset.storage.replication.factor=1
  status.storage.replication.factor=1
  
  internal.key.converter=org.apache.kafka.connect.json.JsonConverter
  internal.value.converter=org.apache.kafka.connect.json.JsonConverter
  internal.key.converter.schemas.enable=false
  internal.value.converter.schemas.enable=false
  </code></pre>
  
  <p><strong>I send data like this:</strong></p>
  
  <pre><code>./bin/kafka-avro-console-producer \
    --broker-list 10.0.0.0:9092 --topic orders \
    --property parse.key="true" \
    --property key.schema='{"type":"record","name":"key_schema","fields":[{"name":"id","type":"int"}]}' \
    --property key.separator="$" \
    --property value.schema='{"type":"record","name":"myrecord","fields":[{"name":"id","type":["null","int"],"default": null},{"name":"product","type": ["null","string"],"default": null}, {"name":"quantity", "type":  ["null","int"],"default": null}, {"name":"price","type":  ["null","int"],"default": null}]}' \
    --property schema.registry.url=http://10.0.0.0:8081
  </code></pre>
  
  <p><strong>I can insert or update data on oracle like this</strong></p>
  
  <pre><code>{"id":3}${"id": {"int":2}, "product": {"string":"Yağız Gülbahar"}, "quantity": {"int":1071}, "price": {"int":1453}}
  </code></pre>
  
  <p><strong>but when i put this to delete record on oracle it cant delete data just update columns as null</strong></p>
  
  <pre><code>{"id":2}${"id": null, "product": null , "quantity": null , "price": null }
  </code></pre>
  
  <p>how can I solve this?</p>
  
  <p>thanks in advance</p>