I have three tables , I want to merge the all three tables and send the data to elastic search, currently i am using SQL-CLI in apache flink.

CREATE TABLE customers_inv (
     `id` STRING,
     `first_name` STRING,
     `last_name` STRING,
     `email` VARCHAR,
     PRIMARY KEY (`id`) NOT ENFORCED
   ) WITH (
  'connector' = 'kafka',
  'topic' = 'dbserver1.inventory.customers',
  'properties.bootstrap.servers' = 'kafka:9092',
  'properties.group.id' = 'eventsGroup=customersinv',
  'value.debezium-json.schema-include' = 'true',
  'scan.startup.mode' = 'earliest-offset',
  'value.format' = 'debezium-json'
   );
   
CREATE TABLE products_inv (
     `id` STRING,
     `name` VARCHAR,
     `description` VARCHAR,
     `weight` FLOAT,
     PRIMARY KEY (`id`) NOT ENFORCED
   ) WITH (
  'connector' = 'kafka',
  'topic' = 'dbserver1.inventory.products',
  'properties.bootstrap.servers' = 'kafka:9092',
  'properties.group.id' = 'eventsGroup=productsinv',
  'value.debezium-json.schema-include' = 'true',
  'scan.startup.mode' = 'earliest-offset',
  'value.format' = 'debezium-json'
   );

CREATE TABLE orders_inv (
     `id` STRING,
     `order_date` STRING,
     `purchaser` STRING,
     `quantity` INT,
     `product_id` STRING,
     PRIMARY KEY (`id`) NOT ENFORCED
   ) WITH (
  'connector' = 'kafka',
  'topic' = 'dbserver1.inventory.orders',
  'properties.bootstrap.servers' = 'kafka:9092',
  'properties.group.id' = 'eventsGroup=ordersinv',
  'value.debezium-json.schema-include' = 'true',
  'scan.startup.mode' = 'earliest-offset',
  'value.format' = 'debezium-json'
   );

How can i further proceed and send the data to elasticsearch I tried to create new table and insert data but didn't worked.

CREATE TABLE users_sink (
     `name` STRING
   ) WITH (
  'connector' = 'kafka',
  'topic' = 'sink-topic-json',
  'properties.bootstrap.servers' = 'kafka:9092',
  'properties.group.id' = 'SinkGroup',
   'value.format' = 'json'
   );
   
INSERT INTO users_sink   
SELECT first_name from customers_inv;

I am getting this error and could'nt get much information about it

[ERROR] Could not execute SQL statement. Reason: 
org.apache.flink.table.api.TableException: Table sink 'default_catalog.default_database.users_sink' doesn't support consuming update and delete changes which is produced by node TableSourceScan(table=[[def
ault_catalog, default_database, customers_inv]], fields=[id, first_name, last_name, email])

Any other approches also will be helpful. Thanks in Advance

your input table will contain flink row with +U and -U kind, i.e. changelog, either you can have a window tvf sql between the source and sink or you need to change sink to flink upsert kafka

Okay, Let me give a try by changing it

source should be upsert kafka in case you have a compacted topic

Okay and does upsert kafka(for sink) supports debezium-json so that i cound send it to the elasticsearch?