Unverified changefeed updates are still saved into Coordinator memory despite not persisted into backend

[kennytm]

What did you do?

1. Prepare

tiup playground-ng:nightly v8.5.5 --db 1 --kv 1 --pd 1 --ticdc 2 --ticdc.binpath ~/.tiup/components/cdc/v8.5.6-release.1/cdc --ticdc.config ./ticdc-newarch.toml --without-monitor

tiup cdc:v8.5.6-release.1 cli changefeed create -c c --sink-uri 'blackhole:'
tiup cdc:v8.5.6-release.1 cli changefeed pause -c c
tiup cdc:v8.5.6-release.1 cli changefeed query -c c | jq .sink_uri
# "blackhole:"

2. Update changefeed to an invalid Sink

echo 'y' | tiup cdc:v8.5.6-release.1 cli changefeed update -c c --sink-uri 'whitehole:'
# Error: [CDC:ErrSinkURIInvalid]sink uri invalid 'whitehole:': [CDC:ErrSinkURIInvalid]sink uri invalid 'whitehole:'

3. Check the Sink URI again

tiup cdc:v8.5.6-release.1 cli changefeed query -c c | jq .sink_uri
# "whitehole:" (<-- unexpected!)

4. Move coordinator and check again

curl -X POST 'http://127.0.0.1:8300/api/v2/owner/resign'
# (move the coordinator to the other CDC)
tiup cdc:v8.5.6-release.1 cli changefeed query -c c | jq .sink_uri
# "blackhole:"

What did you expect to see?

If changefeed update failed, the in-memory changefeed info should not be modified. That is, step 3 should still print "blackhole:".

What did you see instead?

^

Versions of the cluster

v8.5.6-release.1
v8.5.5

[kennytm]

_{(internal reference TICKET-8212)}

The problem is that oldCfInfo here, obtained through (*coordinator).GetChangefeed → (*Controller).GetChangefeed, is a shared pointer directly taken out of the changefeedDB.

ticdc/api/v2/changefeed.go

Line 900 in 95374f0

oldCfInfo, status, err := co.GetChangefeed(c, changefeedDisplayName)

So all assignment here will take effect immediately despite checks later may pronounce the changes are invalid

ticdc/api/v2/changefeed.go

Lines 924 to 941 in 95374f0

	if updateCfConfig.TargetTs != 0 {
	if updateCfConfig.TargetTs <= oldCfInfo.StartTs {
	_ = c.Error(errors.ErrChangefeedUpdateRefused.GenWithStack(
	"can not update target_ts:%d less than start_ts:%d",
	updateCfConfig.TargetTs, oldCfInfo.StartTs))
	return
	}
	oldCfInfo.TargetTs = updateCfConfig.TargetTs
	}
	if updateCfConfig.ReplicaConfig != nil {
	configUpdated = true
	oldCfInfo.Config = updateCfConfig.ReplicaConfig.ToInternalReplicaConfig()
	}
	if updateCfConfig.SinkURI != "" {
	sinkURIUpdated = true
	oldCfInfo.SinkURI = updateCfConfig.SinkURI
	}
	if updateCfConfig.StartTs != 0 {

If the changes are invalid, (*coordinator).UpdateChangefeed at the end of the function won't be called, leading to inconsistency between the in-memory and backend record of the changefeed info.