anyone know how to make deduplication work for kafka source and sink connector? I am getting following error

2023-08-29 13:48:41,701 [] ERROR ExceptionHandler                      [] - An exception occurred at startup:
org.apache.flink.table.api.TableException: StreamPhysicalDeduplicate doesn't support consuming update and delete changes which is produced by node TableSourceScan(table=[[default_catalog, default_database, watermark=[-(_timestamp, 30000:INTERVAL SECOND)], idletimeout=[300000]]], fields=[id, time_created, cid, bid, _message_type, _uuid, _writer_schema_id, _timestamp, _timestamp_microsecond])

The source and sink table looks like this source table

DefaultCatalogTable{schema=(
  `id` INT NOT NULL,
  `time_created` INT,
  `cid` INT,
  `bid` INT,
  `_message_type` STRING,
  `_uuid` STRING,
  `_writer_schema_id` INT,
  `_timestamp` TIMESTAMP(3),
  `_timestamp_microsecond` BIGINT,
  WATERMARK FOR `_timestamp` AS [_timestamp - INTERVAL '30' SECOND],
  CONSTRAINT `PK_id` PRIMARY KEY (`id`) NOT ENFORCED
)

sink table

DefaultCatalogTable{schema=(
  `id` INT NOT NULL,
  `bid` INT,
  `cid` INT,
  `_message_type` VARCHAR(7) NOT NULL,
  CONSTRAINT `PK_id` PRIMARY KEY (`id`) NOT ENFORCED
)

Query is as follows

SELECT id, bid, cid, _message_type
  FROM (
    SELECT *,
      ROW_NUMBER() OVER (PARTITION BY id, bid, cid, _message_type ORDER BY _timestamp DESC) AS row_num
    FROM (
      SELECT
        id, bid, cid, _timestamp,
        CASE WHEN _message_type = 'delete' THEN 'delete' ELSE 'refresh' END AS _message_type
      FROM ad_bid
    )
  )
  WHERE row_num = 1

From the stack trace, it appears that you are using an upsert Kafka source that includes a changelog stream. However, using row_number for deduplication is only applicable to append-only streams. If you need to deduplicate on a CDC stream, you can set

table.exec.source.cdc-events-duplicate

to

true

. For more information, you can refer to the documentation at https://nightlies.apache.org/flink/flink-docs-master/docs/dev/table/config/#table-exec-source-cdc-events-duplicate.

thanks 😄