unstable integration test random_drop_message

[wk989898]

What did you do?

run integration test with redo

What did you expect to see?

No response

What did you see instead?

https://do.pingcap.net/jenkins/blue/organizations/jenkins/pingcap%2Fticdc%2Fpull_cdc_mysql_integration_heavy/detail/pull_cdc_mysql_integration_heavy/6091/pipeline/485
data inconsistency

Versions of the cluster

Upstream TiDB cluster version (execute SELECT tidb_version(); in a MySQL client):

(paste TiDB cluster version here)

Upstream TiKV version (execute tikv-server --version):

(paste TiKV version here)

TiCDC version (execute cdc version):

(paste TiCDC version here)

[wk989898]

Analyze

cdc2.log

[2025/12/30 02:19:27.139 +08:00] [DEBUG] [basic_dispatcher.go:406] ["dispatcher receive all event"] [dispatcher=83769698967099914411017462095718873739] [mode=0]
 [eventType=DMLEvent] [event="DMLEvent{Version: 1, DispatcherID: 83769698967099914411017462095718873739, Seq: 145, PhysicalTableID: 131, 
 StartTs: 463216930385559599, CommitTs: 463216930385559600, Table: test.test_table_8, Checksum: [], Length: 59, Size: 8067, Rows: 
Update: PreRow: 10, arithmetic_10_1767032355_266,Row: 10, arithmetic_10_1767032356_267;

[2025/12/30 02:19:31.141 +08:00] [DEBUG] [basic_dispatcher.go:406] ["dispatcher receive all event"] [dispatcher=83769698967099914411017462095718873739] [mode=0] [eventType=DMLEvent]
 [event="DMLEvent{Version: 1, DispatcherID: 83769698967099914411017462095718873739, Seq: 147, PhysicalTableID: 131, 
StartTs: 463216931800875019, CommitTs: 463216931800875020, Table: test.test_table_8, Checksum: [], Length: 61, Size: 8334, Rows:
 Update: PreRow: 10, arithmetic_10_1767032356_267,Row: 10, arithmetic_10_1767032359_272;

[2025/12/30 02:19:32.033 +08:00] [DEBUG] [maintainer.go:712] ["can advance checkpointTs"] [changefeedID=default/1601687249592459948517439283040421030530] 
[newCheckpointTs=463216932482973723] [newResolvedTs=463216932482973723] [minCheckpointTsForScheduler=18446744073709551615] [minCheckpointTsForBarrier=18446744073709551615]
[2025/12/30 02:19:32.033 +08:00] [DEBUG] [redo_dispatcher.go:115] ["update redo meta"] [resolvedTs=463216932797022215] [checkpointTs=463216932482973723]

cdc1-restart-3.log

[2025/12/30 02:19:29.433 +08:00] [INFO] [dispatcher_manager.go:293] ["dispatcher manager initialized"] 
[changefeedID=default/1601687249592459948517439283040421030530] [maintainerID=35f71d6e-b906-4abc-bf1b-ff3a8f88445f] 
[startTs=463216875833393192] [sinkQuota=536870912] [redoQuota=536870912] [redoEnable=true] [outputRawChangeEvent=false] [filterConfig="filterConfig:<rules:\"*.*\" > "]

[2025/12/30 02:19:34.455 +08:00] [INFO] [event_broker.go:1126] ["reset dispatcher"] [changefeedID=default/1601687249592459948517439283040421030530] [dispatcherID=83769698967099914411017462095718873739] [tableID=131] 
[span="tableID: 131, startKey: 7480000000000000ff835f720000000000fa, endKey: 7480000000000000ff835f730000000000fa, keyspaceID: 0"] 
[originStartTs=463216930385559569] [newStartTs=463216930385559569] [newEpoch=1] [resetTime=331.038µs]

[2025/12/30 02:19:35.466 +08:00] [DEBUG] [basic_dispatcher.go:406] ["dispatcher receive all event"] [dispatcher=83769698967099914411017462095718873739] [mode=0] [eventType=DMLEvent] 
[event="DMLEvent{Version: 1, DispatcherID: 83769698967099914411017462095718873739, Seq: 2, PhysicalTableID: 131, 
StartTs: 463216930385559599, CommitTs: 463216930385559600,
 Table: test.test_table_8, Checksum: [], Length: 59, Size: 8067, Rows:
 Update: PreRow: 10, arithmetic_10_1767032355_266,Row: 10, arithmetic_10_1767032356_267;

cdc1 restarts with the smaller start-ts than checkpointTs, which causes cdc1 to write duplicate data to the downstream.
If the node of the maintainer is not restarted, other nodes always use the startCheckpointTs to initialize.

ticdc/maintainer/maintainer.go

Lines 1045 to 1053 in 53e43bb

	msg := &heartbeatpb.MaintainerBootstrapRequest{
	ChangefeedID: m.changefeedID.ToPB(),
	Config: cfgBytes,
	StartTs: m.startCheckpointTs,
	TableTriggerEventDispatcherId: nil,
	TableTriggerRedoDispatcherId: nil,
	IsNewChangefeed: false,
	KeyspaceId: m.info.KeyspaceID,
	}