DataStax Cassandra Sink Connector - Ingesting data from a Kafka topic according to conditions

I am trying to ingest data from a Kafka topic to a Cassandra table in real-time. For this, I'm using DataStax Cassandra Sink connector. The format of the events in the topic will be JSON. I'm able to directly map JSON fields of the events in the topic to the table, but that's not exactly what I need.

My scenario is like this: There is one topic that needs to be mapped to more than one Cassandra tables. The events in the topic should be able to go to the tables according to some conditions. Let's say there are events with different sort of headers A, B, and C. Events with header A needs to go to Cassandra table A, header B events to table B and header C to table C. In the connector configuration JSON file, am I able to do this?

I've made some attempts for a single topic and single table and tried to ingest the events with a specific condition on "ID" field.

My current configuration file for the connector:

{
  "name": "cassandra-json-sink",
  "config": {
    "connector.class": "com.datastax.oss.kafka.sink.CassandraSinkConnector",
    "tasks.max": "1",
    "topics": "json_test_topic",
    "contactPoints": "cassandra",
    "loadBalancing.localDc": "datacenter1",
    "port": 9042,
    "auth.username": "cassandra",
    "auth.password": "cassandra",
    "topic.json_test_topic.kconnect_json.customer.mapping": "id=key, name=value.name, lname=value.lname, adress=value.adress",
    "key.converter": "org.apache.kafka.connect.storage.StringConverter",
    "value.converter": "org.apache.kafka.connect.storage.StringConverter",
    "key.converter.schemas.enable": false,
    "value.converter.schemas.enable": false
  }
}

Details of the Cassandra table:

USE kconnect_json;
CREATE TABLE customer (id TEXT PRIMARY KEY, name TEXT, lname TEXT, adress TEXT);

Example messages I've published to my topic "json_test_topic":

abc:{"name":"john", "lname":"doe", "adress":"WY"}
efg:{"name":"wanda", "lname":"hill", "adress":"CA"}

In this simple attempt I want to ingest events with "name" field = "john" to my Cassandra table. I've changed the previous configuration file for the connector to this:

{
  "name": "cassandra-json-sink",
  "config": {
    "connector.class": "com.datastax.oss.kafka.sink.CassandraSinkConnector",
    "tasks.max": "1",
    "topics": "json_test_topic",
    "contactPoints": "cassandra",
    "loadBalancing.localDc": "datacenter1",
    "port": 9042,
    "auth.username": "cassandra",
    "auth.password": "cassandra",
    "topic.json_test_topic.kconnect_json.customer.mapping": "id=key, name=value.name, lname=value.lname, adress=value.adress",
    "topic.json_test_topic.kconnect_json.customer.query": "INSERT INTO kconnect_json.customer(id, name, lname, adress) SELECT :id, :name, :lname, :adress FROM topic.json_test_topic WHERE :name = 'john';",
    "topic.json_test_topic.kconnect_json.musteri.deletesEnabled": false,
    "key.converter": "org.apache.kafka.connect.storage.StringConverter",
    "value.converter": "org.apache.kafka.connect.storage.StringConverter",
    "key.converter.schemas.enable": false,
    "value.converter.schemas.enable": false
  }
}

While doing this, I've tried to utilize this doc and customized the examples a bit: https://docs.datastax.com/en/kafka/doc/kafka/kafkaCqlQuery.html

But this customization is not working, as the query on the Cassandra side is not valid. I was not able to find INSERT statements on Cassandra with inner SELECT statements in them, so this might be an issue, but I don't know the alternative.

Would be glad if anyone can help.