Hello folks, I’m running a simple query on top of a table that is created with an ‘upsert-kafka’ connector. Here is the query:

SELECT event_id, event_type, data_source_type FROM event_changes

Here event_id, event_type, data_source_type are the primary key fields in the table. When run in the sql client shell, this query is giving proper results. When run as part of the job throws below error:

Exception in thread "main" org.apache.flink.table.api.TableException: Table sink '*anonymous_datastream_sink$1*' doesn't support consuming update and delete changes which is produced by node ChangelogNormalize(key=[event_id, event_type, data_source_type])

This is observed only when I run using tableEnv.sqlQuery() method, when I tried with tableEnv.executeSql() method, the query is running fine and I’m able to see the TableResult type. I am trying to create a DataStream out of the result so using .sqlQuery() method. Any inputs on this is appreciated. Thanks in advance.