Hello all, I have setup of ◦ flink (1.20.1), ◦ flink-cdc (3.5.0), ◦ starrocks (4.0.0), ◦ mysql(8.0). • I'm using below pipeline yaml for data streaming. the problem is that there is one table that is related to add/delete cart at a time it has more data (more row count) in starrocks then mysql. I have check there is no data duplication of data in starrocks but delete is not getting performed what can be the problem? how can i resolve this? • I have tried the drop the table and restarting the pipeline. for some time it works fine. but after some time when traffic is high then starrocks data count getting more then mysql. even when traffic is almost none in non working hours its still dose not get back to normal

source:
  type: mysql
  hostname: 10.3.4.168
  name: ms_mysql_prod_sales_us_ca
  port: 3306
  username: flink_replication_admin
  password: Uq^ZXXXXXXX
  tables: |
    ms.sales_flat_quote_item,
    ms_canada.sales_flat_quote_item
  server-id: 8000-8500
  server-time-zone: UTC
  scan.newly-added-table.enabled: true
  schema-change.enabled: true

sink:
  type: starrocks
  name: StarRocks Sink
  jdbc-url: jdbc:<mysql://10.231.XX.XX:9030>
  load-url: 10.231.XX.XX:8030
  username: root
  password: dpBjIXXXXXX
  sink.buffer-flush.interval-ms: 2000
  sink.buffer-flush.max-bytes: 67108864
  table.create.properties.replication_num: 1
  sink.at-least-once.use-transaction-stream-load: false
  sink.io.thread-count: 6
  sink.label-prefix: flink_sync_custom

route:
  - source-table: ms.\.*
    sink-table: ms.mage_us_<>
    replace-symbol: <>
    description: route all tables in ms to ms
  - source-table: ms_canada.\.*
    sink-table: ms.mage_ca_<>
    replace-symbol: <>
    description: route all tables in ms_canada to ms
pipeline:
  parallelism: 4
  name: Sync MySQL Sales US/CA DWH Tables to StarRocks
  execution.checkpointing.interval: 30000
  execution.checkpointing.max-concurrent-checkpoints: 1
  execution.checkpointing.mode: EXACTLY_ONCE
  execution.checkpointing.timeout: 300000
  execution.checkpointing.min-pause: 5000
  state.backend.changelog.enabled: true