ElasticsearchSinkConnector object mapping can't be changed from nested to non-nested

I'm streaming my data from kafka topic into elasticsearch. But it's throwing this error from connector {\"type\":\"illegal_argument_exception\",\"reason\":\"object mapping [search_data] can't be changed from nested to non-nested\"}

But when I take the message from topic and add document manually using elasticsearch api, it is working fine.

Does kafka-connect-elasticsearch not support nested object type?

Please help me reply on this as I am stuck here for days.

Elasticsearch version: 7.6.2

Kafka Connect Image: confluentinc/cp-kafka-connect:5.4.2

Below is my configuration for connector.

{
    "connector.class": "io.confluent.connect.elasticsearch.ElasticsearchSinkConnector",
    "tasks.max": "1",
    "topics": "es_sink_products",
    "key.ignore": "false",
    "schema.ignore": "true",
    "connection.url": "localhost:9200",
    "type.name": "kafka-connect",
    "name": "product-elasticsearch-sink",
    "key.converter": "org.apache.kafka.connect.storage.StringConverter"
}

Elasticsearch Schema

{
  "mappings": {
    "properties": {
      "search_result_data": {
        "properties": {
          "product_id": {"type": "long"},
          "product_name": {"type": "text"},
        }
      },
      "search_data":{
        "type": "nested",
        "include_in_parent": false,
        "properties": {
          "product_id": {"type": "long"},
          "full_text": {
            "type": "text",
          },
        }
      }
    }
  }
}

Sample message from topic es_sink_products

{
    "search_data": {
        "product_id": 1,
        "full_text": "Product 1"
    },
    "search_result_data": {
        "product_id": 1,
        "product_name": "Product Name 1"
    }
}

This is the full error from connector "org.apache.kafka.connect.errors.ConnectException: Exiting WorkerSinkTask due to unrecoverable exception.\n\tat org.apache.kafka.connect.runtime.WorkerSinkTask.deliverMessages(WorkerSinkTask.java:561)\n\tat org.apache.kafka.connect.runtime.WorkerSinkTask.poll(WorkerSinkTask.java:322)\n\tat org.apache.kafka.connect.runtime.WorkerSinkTask.iteration(WorkerSinkTask.java:224)\n\tat org.apache.kafka.connect.runtime.WorkerSinkTask.execute(WorkerSinkTask.java:192)\n\tat org.apache.kafka.connect.runtime.WorkerTask.doRun(WorkerTask.java:177)\n\tat org.apache.kafka.connect.runtime.WorkerTask.run(WorkerTask.java:227)\n\tat java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)\n\tat java.util.concurrent.FutureTask.run(FutureTask.java:266)\n\tat java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)\n\tat java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)\n\tat java.lang.Thread.run(Thread.java:748)\nCaused by: org.apache.kafka.connect.errors.ConnectException: Bulk request failed: [{\"type\":\"illegal_argument_exception\",\"reason\":\"object mapping [search_data] can't be changed from nested to non-nested\"}]\n\tat io.confluent.connect.elasticsearch.bulk.BulkProcessor$BulkTask.handleMalformedDoc(BulkProcessor.java:479)\n\tat io.confluent.connect.elasticsearch.bulk.BulkProcessor$BulkTask.execute(BulkProcessor.java:433)\n\tat io.confluent.connect.elasticsearch.bulk.BulkProcessor$BulkTask.call(BulkProcessor.java:389)\n\tat io.confluent.connect.elasticsearch.bulk.BulkProcessor$BulkTask.call(BulkProcessor.java:375)\n\tat java.util.concurrent.FutureTask.run(FutureTask.java:266)\n\tat java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)\n\tat java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)\n\tat java.lang.Thread.run(Thread.java:748)\n\tat io.confluent.connect.elasticsearch.bulk.BulkProcessor$BulkProcessorThread.run(BulkProcessor.java:370)\n"