IN CLUSTER cluster_name FROM item_name INTO kafka_sink_connection KEY ( key_column , ) NOT ENFORCED HEADERS headers_column FORMAT KEY FORMAT sink_format_spec VALUE FORMAT sink_format_spec ENVELOPE DEBEZIUM UPSERT WITH with_options