syncer(dm): discard checkpoint snapshot when syncer reset (pingcap#8378…

…) (pingcap#8558)

close pingcap#8374

dm/syncer/checkpoint.go

@@ -268,6 +268,10 @@ type CheckPoint interface {
 	// Snapshot make a snapshot of current checkpoint. If returns nil, it means nothing has changed since last call.
 	Snapshot(isSyncFlush bool) *SnapshotInfo
 
+	// DiscardPendingSnapshots discards all pending snapshots. It's used when we create a snapshot but are unable to
+	// call FlushPointsExcept() to flush the snapshot due to some error.
+	DiscardPendingSnapshots()
+
 	// FlushPointsExcept flushes the global checkpoint and tables'
 	// checkpoints except exceptTables, it also flushes SQLs with Args providing
 	// by extraSQLs and extraArgs. Currently extraSQLs contain shard meta only.
@@ -369,7 +373,7 @@ type RemoteCheckPoint struct {
 	needFlushSafeModeExitPoint atomic.Bool
 
 	logCtx *tcontext.Context
-	// these fields are used for async flush checkpoint
+
 	snapshots   []*remoteCheckpointSnapshot
 	snapshotSeq int
 }
@@ -456,6 +460,14 @@ func (cp *RemoteCheckPoint) Snapshot(isSyncFlush bool) *SnapshotInfo {
 	}
 }
 
+// DiscardPendingSnapshots discard all pending snapshots.
+func (cp *RemoteCheckPoint) DiscardPendingSnapshots() {
+	cp.Lock()
+	defer cp.Unlock()
+
+	cp.snapshots = nil
+}
+
 // Init implements CheckPoint.Init.
 func (cp *RemoteCheckPoint) Init(tctx *tcontext.Context) (err error) {
 	var db *conn.BaseDB
@@ -700,7 +712,7 @@ func (cp *RemoteCheckPoint) FlushPointsExcept(
 	cp.Lock()
 
 	if len(cp.snapshots) == 0 || cp.snapshots[0].id != snapshotID {
-		cp.logCtx.Logger.DPanic("snapshot not found", zap.Int("id", snapshotID))
+		return errors.Errorf("snapshot %d not found", snapshotID)
 	}
 	snapshotCp := cp.snapshots[0]
 	cp.snapshots = cp.snapshots[1:]

dm/syncer/checkpoint_test.go

@@ -25,7 +25,6 @@ import (
 	"github.com/DATA-DOG/go-sqlmock"
 	"github.com/go-mysql-org/go-mysql/mysql"
 	. "github.com/pingcap/check"
-	"github.com/pingcap/errors"
 	"github.com/pingcap/log"
 	tidbddl "github.com/pingcap/tidb/ddl"
 	"github.com/pingcap/tidb/parser/ast"
@@ -170,8 +169,11 @@ func (s *testCheckpointSuite) testGlobalCheckPoint(c *C, cp CheckPoint) {
 	s.mock.ExpectBegin()
 	s.mock.ExpectExec("(162)?"+flushCheckPointSQL).WithArgs(cpid, "", "", pos1.Name, pos1.Pos, "", "", 0, "", "null", true).WillReturnResult(sqlmock.NewResult(0, 1))
 	s.mock.ExpectCommit()
-	err = cp.FlushPointsExcept(tctx, cp.Snapshot(true).id, nil, nil, nil)
-	c.Log(errors.ErrorStack(err))
+	// Create a new snapshot, and discard it, then create a new snapshot again.
+	cp.Snapshot(true)
+	cp.DiscardPendingSnapshots()
+	snap := cp.Snapshot(true)
+	err = cp.FlushPointsExcept(tctx, snap.id, nil, nil, nil)
 	c.Assert(err, IsNil)
 	c.Assert(cp.GlobalPoint().Position, Equals, pos1)
 	c.Assert(cp.FlushedGlobalPoint().Position, Equals, pos1)
@@ -268,7 +270,7 @@ func (s *testCheckpointSuite) testGlobalCheckPoint(c *C, cp CheckPoint) {
 
 	// should flush because checkpoint hasn't been updated before (cp.globalPointCheckOrSaveTime.IsZero() == true).
 	snapshot := cp.Snapshot(true)
-	c.Assert(snapshot.id, Equals, 3)
+	c.Assert(snapshot.id, Equals, 4)
 
 	s.mock.ExpectQuery(loadCheckPointSQL).WillReturnRows(sqlmock.NewRows(nil))
 	err = cp.Load(tctx)

dm/syncer/syncer.go

@@ -617,6 +617,7 @@ func (s *Syncer) reset() {
 	}
 	// create new job chans
 	s.newJobChans()
+	s.checkpoint.DiscardPendingSnapshots()
 	s.checkpointFlushWorker = &checkpointFlushWorker{
 		input:              make(chan *checkpointFlushTask, 16),
 		cp:                 s.checkpoint,