Hello All, I need some help with getting flink kafka connect to produce correct data when data is of complex object type

CREATE TABLE IF NOT EXISTS SOURCE_TABLE
(
    source      VARCHAR NOT NULL,
    id          VARCHAR NOT NULL,
    contentJson VARCHAR NOT NULL
) WITH (
    'connector' = 'kafka',
    'topic-pattern' = '${source_topic_pattern}',
    'properties.bootstrap.servers' = '${source_kafka_servers}',
    'properties.group.id' = '${kafka_consumer_group_id}'
);

contentJson example

"contentJson": "{\"field1\":[{\"identityId\":\"21ade8be05f548a983acf68a8ab5ff73\",\"IdsAfter\":[\"seg-id-1\"]}]}"

if i have this sink defined as

CREATE TABLE IF NOT EXISTS SINK_TABLE
(
    field1      VARCHAR NOT NULL,
) WITH (
    'connector' = 'kafka',
    'topic-pattern' = 'SINK_TABLE',
    'properties.bootstrap.servers' = '${source_kafka_servers}',
    'properties.group.id' = '${kafka_consumer_group_id}'
);

here the field1 is a string, json is enclosed with in quotes

"[{\"identityId\":\"21ade8be05f548a983acf68a8ab5ff73\"}]"

i can define the field as Array of ROW I get what I need but I need to write a UDF that converts the json into RowData using GenericRowData.of() method if my sink is

CREATE TABLE IF NOT EXISTS SINK_TABLE
(
    field1      ARRAY<ROW<identityId STRING>>
) WITH (
    'connector' = 'kafka',
    'topic-pattern' = 'SINK_TABLE',
    'properties.bootstrap.servers' = '${source_kafka_servers}',
    'properties.group.id' = '${kafka_consumer_group_id}'
);

output has data like this

[{"identityId":"21ade8be05f548a983acf68a8ab5ff73"}]

How do i do this without using udf to convert this to row data. Better do I get the required output without defining that as a complex object in the sink i have tried

field1 RAW

i get an error.