cdc changefeed stucks after rolling restart kafka

[fubinzh]

What did you do?

1. TiDB cluster with 3 CDC node, downstream Kafka cluster with 3 nodes.
2. There are 2 kafka changefeed running, big-table for a big table 50TB+, 4k-table for 4k small tables.

[root@tc-ticdc-0 log]# /cdc cli changefeed --server http://127.0.0.1:8301 query -c big-table
{
  "upstream_id": 7234390566115419974,
  "namespace": "default",
  "id": "big-table",
  "sink_uri": "kafka://downstream-kafka.cdc-kafka-big-cluster-tps-1712340-1-428:9092/kafka-bigcluster?protocol=avro\u0026partition-num=3\u0026replication-factor=3\u0026enable-tidb-extension=true\u0026avro-decimal-handling-mode
=string\u0026avro-bigint-unsigned-handling-mode=string",
  "config": {
    "memory_quota": 1017986918,
    "case_sensitive": true,
    "enable_old_value": true,
    "force_replicate": false,
    "ignore_ineligible_table": false,
    "check_gc_safe_point": true,
    "enable_sync_point": false,
    "bdr_mode": false,
    "sync_point_interval": 600000000000,
    "sync_point_retention": 86400000000000,
    "filter": {
      "rules": [
        "bigcluster.*"
      ],
      "event_filters": null
    },
    "mounter": {
      "worker_num": 16
    },
    "sink": {
      "protocol": "avro",
      "schema_registry": "http://10.200.23.41:8081",
      "csv": {
        "delimiter": ",",
        "quote": "\"",
        "null": "\\N",
        "include_commit_ts": false
      },
      "column_selectors": null,
      "transaction_atomicity": "",
      "encoder_concurrency": 16,
      "terminator": "\r\n",
      "date_separator": "none",
      "enable_partition_separator": true,
      "file_index_digit": 0,
      "enable_kafka_sink_v2": false,
      "only_output_updated_columns": null
    },
    "consistent": {
      "level": "none",
      "max_log_size": 64,
      "flush_interval": 2000,
      "storage": "",
      "use_file_backend": false
    },
"scheduler": {
      "enable_table_across_nodes": true,
      "region_threshold": 50000,
      "write_key_threshold": 10000
    },
    "integrity": {
      "integrity_check_level": "correctness",
      "corruption_handle_level": "warn"
    }
  },
  "create_time": "2023-05-19 09:31:36.388",
  "start_ts": 441752185454985498,
  "resolved_ts": 441804940470124612,
  "target_ts": 0,
  "checkpoint_tso": 441803299776364635,
  "checkpoint_time": "2023-05-29 15:37:19.601",
  "state": "normal",
  "error": null,
  "error_history": null,
  "creator_version": "v7.1.0",
  "task_status": [
    {
      "capture_id": "1edf6b1c-10bb-40a2-8c8f-15acce5dbc66",
      "table_ids": [
        100
      ],
      "table_operations": null
    },
    {
      "capture_id": "f0c725ef-686a-4a08-843c-1636af8ba66d",
      "table_ids": [
        100
      ],
      "table_operations": null
    },
    {
      "capture_id": "18cdfaf1-00ec-405f-b653-036679a4ab06",
      "table_ids": [
        100
      ],
      "table_operations": null
    }
  ]
}

4. Rolling restart kafka

What did you expect to see?

Changefeed should not stuck after kafka rolling upgrade.

What did you see instead?

big-table changefeed stucks, 4k-tables looks OK.

Versions of the cluster

TiCDC version (execute cdc version):

[root@tc-ticdc-0 log]# /cdc version
Release Version: v7.1.0
Git Commit Hash: a2158c3691a5c758606f8db5e68ef21ac822f3fc
Git Branch: heads/refs/tags/v7.1.0
UTC Build Time: 2023-05-26 12:59:07
Go Version: go version go1.20.3 linux/amd64
Failpoint Build: false

[nongfushanquan]

/remove-label affects-7.1
/label affects-7.1

[nongfushanquan]

/label affects-7.1