Kafka consumer panic

[fubinzh]

What did you do?

1. topo Upstream TiDB -> cdc -> kafka -> Kafka consumer -> Target tidb
2. Br restore a backup to upstream and target TiDB
3. Enable row checksum on upstream and downstream tidb
4. Create a Kafka changefeed

sink-uri: "sink_uri": "kafka://downstream-kafka.cdc-kafka-simple-workloads-tps-7587577-1-639:9092/cdc_kafka_simple_cpekm5jsucktnl0c7avg?replication-factor=3\u0026partition-num=4\u0026protocol=simple\u0026encoding-format=avro" 
# changefeed config ( for both cdc changefeed and  Kafka consumer)
[integrity]
                integrity-check-level = "correctness"
                corruption-handle-level = "error"
                [sink.kafka-config.large-message-handle]
                large-message-handle-option = "claim-check"
                claim-check-storage-uri = "s3://tmp/test-infra-redolog"
                large-message-handle-compression = "snappy"
                [sink]
                dispatchers = [
                        {matcher = ["*.*"], partition = "index-value"}
                ]

5. run workload to upstream tidb
6. run Kafka consumer to consumer the message received in Kafka

What did you expect to see?

Kafka Consumer should not panic

What did you see instead?

Consumer panic:

[2024/06/03 04:52:37.977 +00:00] [INFO] [db_helper.go:173] ["sink uri is configured"] [dsn="root:******@tcp(target-tidb.cdc-kafka-simple-workloads-tps-7587577-1-639:4000)/?interpolateParams=true&multiStatements=true&allow_auto_random_explicit_insert=1&charset=utf8mb4&foreign_key_checks
[2024/06/03 04:52:37.982 +00:00] [INFO] [mysql_ddl_sink.go:107] ["MySQL DDL sink is created"] [namespace=default] [changefeed=kafka-consumer]
[2024/06/03 04:56:39.248 +00:00] [INFO] [decoder.go:329] ["table info stored"] [schema=uds2] [table=index_Data23] [version=450203511471996937]
[2024/06/03 04:56:39.248 +00:00] [INFO] [decoder.go:329] ["table info stored"] [schema=uds2] [table=index_Data23] [version=450203511471996937]
[2024/06/03 04:56:39.249 +00:00] [INFO] [decoder.go:329] ["table info stored"] [schema=uds2] [table=index_Data23] [version=450203511471996937]
[2024/06/03 04:56:39.250 +00:00] [INFO] [decoder.go:329] ["table info stored"] [schema=uds2] [table=index_Data23] [version=450203511471996937]
[2024/06/03 04:56:40.148 +00:00] [INFO] [decoder.go:329] ["table info stored"] [schema=uds2] [table=index_Data20] [version=450203511471996937]
[2024/06/03 04:56:40.148 +00:00] [INFO] [decoder.go:329] ["table info stored"] [schema=uds2] [table=index_Data20] [version=450203511471996937]
[2024/06/03 04:56:40.148 +00:00] [INFO] [decoder.go:329] ["table info stored"] [schema=uds2] [table=index_Data20] [version=450203511471996937]
[2024/06/03 04:56:40.149 +00:00] [INFO] [decoder.go:329] ["table info stored"] [schema=uds2] [table=index_Data20] [version=450203511471996937]
[2024/06/03 04:56:40.150 +00:00] [INFO] [decoder.go:329] ["table info stored"] [schema=uds2] [table=index_Data12] [version=450203511471996937]
[2024/06/03 04:56:40.150 +00:00] [INFO] [decoder.go:329] ["table info stored"] [schema=uds2] [table=index_Data12] [version=450203511471996937]
[2024/06/03 04:56:40.151 +00:00] [INFO] [decoder.go:329] ["table info stored"] [schema=uds2] [table=index_Data16] [version=450203511471996937]
[2024/06/03 04:56:40.151 +00:00] [INFO] [decoder.go:329] ["table info stored"] [schema=uds2] [table=index_Data16] [version=450203511471996937]
[2024/06/03 04:56:40.152 +00:00] [INFO] [decoder.go:329] ["table info stored"] [schema=uds2] [table=index_Data12] [version=450203511471996937]
[2024/06/03 04:56:40.152 +00:00] [INFO] [decoder.go:329] ["table info stored"] [schema=uds2] [table=index_Data12] [version=450203511471996937]
[2024/06/03 04:56:40.153 +00:00] [INFO] [decoder.go:329] ["table info stored"] [schema=uds2] [table=Data20] [version=450203511471996937]
[2024/06/03 04:56:40.153 +00:00] [PANIC] [writer.go:361] ["RowChangedEvent dispatched to wrong partition"] [partition=2] [expected=1] [partitionNum=4] [offset=284] [row="{\"StartTs\":0,\"CommitTs\":450203640958025785,\"RowID\":0,\"PhysicalTableID\":2575,\"TableInfo\":{\"id\":2575,\"nam
~

Versions of the cluster

Kafka consumer:
[release-version=v8.2.0-master-dirty] [git-hash=b52aa5812e7d424028b10a2ff1a8797c2ae61fa1]

CDC:
[root@upstream-ticdc-0 /]# /cdc version
Release Version: v7.5.2
Git Commit Hash: 3625884
Git Branch: HEAD

[fubinzh]

/severity major

[fubinzh]

/assign @3AceShowHand