scheduler(ticdc): fix invlaid checkpoint when redo enabled (#9851)

cdc/owner/changefeed.go

@@ -427,10 +427,7 @@ LOOP2:
 	}
 
 	checkpointTs := c.state.Status.CheckpointTs
-	if c.resolvedTs == 0 {
-		c.resolvedTs = checkpointTs
-	}
-
+	c.resolvedTs = checkpointTs
 	minTableBarrierTs := c.state.Status.MinTableBarrierTs
 
 	failpoint.Inject("NewChangefeedNoRetryError", func() {
@@ -573,6 +570,7 @@ LOOP2:
 		return err
 	}
 	if c.redoMetaMgr.Enabled() {
+		c.resolvedTs = c.redoMetaMgr.GetFlushedMeta().ResolvedTs
 		c.wg.Add(1)
 		go func() {
 			defer c.wg.Done()
@@ -683,6 +681,7 @@ func (c *changefeed) releaseResources(ctx cdcContext.Context) {
 	c.barriers = nil
 	c.initialized = false
 	c.isReleased = true
+	c.resolvedTs = 0
 
 	log.Info("changefeed closed",
 		zap.String("namespace", c.id.Namespace),

cdc/processor/processor.go

@@ -166,6 +166,7 @@ func (p *processor) AddTable(
 			// table is `prepared`, and a `isPrepare = false` request indicate that old table should
 			// be stopped on original capture already, it's safe to start replicating data now.
 			if !isPrepare {
+<<<<<<< HEAD
 				if p.pullBasedSinking {
 					if p.redoDMLMgr.Enabled() {
 						// ResolvedTs is store in external storage when redo log is enabled, so we need to
@@ -177,6 +178,16 @@ func (p *processor) AddTable(
 					}
 				} else {
 					p.tables[tableID].Start(startTs)
+=======
+				redoStartTs := checkpoint.ResolvedTs
+				if p.redo.r.Enabled() {
+					// ResolvedTs is store in external storage when redo log is enabled, so we need to
+					// start table with ResolvedTs in redoDMLManager.
+					p.redo.r.StartTable(span, redoStartTs)
+				}
+				if err := p.sinkManager.r.StartTable(span, startTs, redoStartTs); err != nil {
+					return false, errors.Trace(err)
+>>>>>>> 3b8d55b1cd (scheduler(ticdc): fix invlaid checkpoint when redo enabled (#9851))
 				}
 			}
 			return true, nil

cdc/processor/sinkmanager/manager.go

@@ -831,7 +831,15 @@ func (m *SinkManager) AddTable(tableID model.TableID, startTs model.Ts, targetTs
 }
 
 // StartTable sets the table(TableSink) state to replicating.
+<<<<<<< HEAD
 func (m *SinkManager) StartTable(tableID model.TableID, startTs model.Ts) error {
+=======
+func (m *SinkManager) StartTable(
+	span tablepb.Span,
+	startTs model.Ts,
+	redoStartTs model.Ts,
+) error {
+>>>>>>> 3b8d55b1cd (scheduler(ticdc): fix invlaid checkpoint when redo enabled (#9851))
 	log.Info("Start table sink",
 		zap.String("namespace", m.changefeedID.Namespace),
 		zap.String("changefeed", m.changefeedID.ID),
@@ -857,8 +865,13 @@ func (m *SinkManager) StartTable(tableID model.TableID, startTs model.Ts) error
 	})
 	if m.redoDMLMgr != nil {
 		m.redoProgressHeap.push(&progress{
+<<<<<<< HEAD
 			tableID:           tableID,
 			nextLowerBoundPos: engine.Position{StartTs: 0, CommitTs: startTs + 1},
+=======
+			span:              span,
+			nextLowerBoundPos: engine.Position{StartTs: 0, CommitTs: redoStartTs + 1},
+>>>>>>> 3b8d55b1cd (scheduler(ticdc): fix invlaid checkpoint when redo enabled (#9851))
 			version:           tableSink.(*tableSinkWrapper).version,
 		})
 	}

cdc/processor/sinkmanager/manager_test.go

@@ -152,7 +152,11 @@ func TestAddTable(t *testing.T) {
 	require.True(t, ok)
 	require.NotNil(t, tableSink)
 	require.Equal(t, 0, manager.sinkProgressHeap.len(), "Not started table shout not in progress heap")
+<<<<<<< HEAD
 	err := manager.StartTable(tableID, 1)
+=======
+	err := manager.StartTable(span, 1, 1)
+>>>>>>> 3b8d55b1cd (scheduler(ticdc): fix invlaid checkpoint when redo enabled (#9851))
 	require.NoError(t, err)
 	require.Equal(t, uint64(0x7ffffffffffbffff), tableSink.(*tableSinkWrapper).replicateTs)
 
@@ -176,7 +180,11 @@ func TestRemoveTable(t *testing.T) {
 	tableSink, ok := manager.tableSinks.Load(tableID)
 	require.True(t, ok)
 	require.NotNil(t, tableSink)
+<<<<<<< HEAD
 	err := manager.StartTable(tableID, 0)
+=======
+	err := manager.StartTable(span, 0, 0)
+>>>>>>> 3b8d55b1cd (scheduler(ticdc): fix invlaid checkpoint when redo enabled (#9851))
 	require.NoError(t, err)
 	addTableAndAddEventsToSortEngine(t, e, tableID)
 	manager.UpdateBarrierTs(4, nil)
@@ -216,7 +224,11 @@ func TestGenerateTableSinkTaskWithBarrierTs(t *testing.T) {
 	manager.UpdateBarrierTs(4, nil)
 	manager.UpdateReceivedSorterResolvedTs(tableID, 5)
 	manager.schemaStorage.AdvanceResolvedTs(5)
+<<<<<<< HEAD
 	err := manager.StartTable(tableID, 0)
+=======
+	err := manager.StartTable(span, 0, 0)
+>>>>>>> 3b8d55b1cd (scheduler(ticdc): fix invlaid checkpoint when redo enabled (#9851))
 	require.NoError(t, err)
 
 	require.Eventually(t, func() bool {
@@ -244,7 +256,11 @@ func TestGenerateTableSinkTaskWithResolvedTs(t *testing.T) {
 	manager.UpdateBarrierTs(4, nil)
 	manager.UpdateReceivedSorterResolvedTs(tableID, 3)
 	manager.schemaStorage.AdvanceResolvedTs(4)
+<<<<<<< HEAD
 	err := manager.StartTable(tableID, 0)
+=======
+	err := manager.StartTable(span, 0, 0)
+>>>>>>> 3b8d55b1cd (scheduler(ticdc): fix invlaid checkpoint when redo enabled (#9851))
 	require.NoError(t, err)
 
 	require.Eventually(t, func() bool {
@@ -271,7 +287,11 @@ func TestGetTableStatsToReleaseMemQuota(t *testing.T) {
 	manager.UpdateBarrierTs(4, nil)
 	manager.UpdateReceivedSorterResolvedTs(tableID, 5)
 	manager.schemaStorage.AdvanceResolvedTs(5)
+<<<<<<< HEAD
 	err := manager.StartTable(tableID, 0)
+=======
+	err := manager.StartTable(span, 0, 0)
+>>>>>>> 3b8d55b1cd (scheduler(ticdc): fix invlaid checkpoint when redo enabled (#9851))
 	require.NoError(t, err)
 
 	require.Eventually(t, func() bool {
@@ -347,11 +367,21 @@ func TestSinkManagerRunWithErrors(t *testing.T) {
 		_ = failpoint.Disable("github.com/pingcap/tiflow/cdc/processor/sinkmanager/SinkWorkerTaskError")
 	}()
 
+<<<<<<< HEAD
 	source.AddTable(1)
 	manager.AddTable(1, 100, math.MaxUint64)
 	manager.StartTable(1, 100)
 	source.Add(1, model.NewResolvedPolymorphicEvent(0, 101))
 	manager.UpdateReceivedSorterResolvedTs(1, 101)
+=======
+	span := spanz.TableIDToComparableSpan(1)
+
+	source.AddTable(span, "test", 100)
+	manager.AddTable(span, 100, math.MaxUint64)
+	manager.StartTable(span, 100, 0)
+	source.Add(span, model.NewResolvedPolymorphicEvent(0, 101))
+	manager.UpdateReceivedSorterResolvedTs(span, 101)
+>>>>>>> 3b8d55b1cd (scheduler(ticdc): fix invlaid checkpoint when redo enabled (#9851))
 	manager.UpdateBarrierTs(101, nil)
 
 	timer := time.NewTimer(5 * time.Second)

cdc/scheduler/internal/v3/agent/main_test.go

@@ -0,0 +1,24 @@
+// Copyright 2022 PingCAP, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package agent
+
+import (
+	"testing"
+
+	"github.com/pingcap/tiflow/pkg/leakutil"
+)
+
+func TestMain(m *testing.M) {
+	leakutil.SetUpLeakTest(m)
+}

cdc/scheduler/internal/v3/compat/main_test.go

@@ -0,0 +1,24 @@
+// Copyright 2022 PingCAP, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package compat
+
+import (
+	"testing"
+
+	"github.com/pingcap/tiflow/pkg/leakutil"
+)
+
+func TestMain(m *testing.M) {
+	leakutil.SetUpLeakTest(m)
+}

cdc/scheduler/internal/v3/coordinator.go

@@ -313,7 +313,11 @@ func (c *coordinator) poll(
 	replications := c.replicationM.ReplicationSets()
 	runningTasks := c.replicationM.RunningTasks()
 	allTasks := c.schedulerM.Schedule(
+<<<<<<< HEAD
 		checkpointTs, currentTables, c.captureM.Captures, replications, runningTasks)
+=======
+		checkpointTs, currentSpans, c.captureM.Captures, replications, runningTasks, c.redoMetaManager)
+>>>>>>> 3b8d55b1cd (scheduler(ticdc): fix invlaid checkpoint when redo enabled (#9851))
 
 	// Handle generated schedule tasks.
 	msgs, err = c.replicationM.HandleTasks(allTasks)

cdc/scheduler/internal/v3/coordinator_bench_test.go

@@ -148,7 +148,11 @@ func BenchmarkCoordinatorHeartbeatResponse(b *testing.B) {
 			currentTables = append(currentTables, tableID)
 			captureID := fmt.Sprint(i % captureCount)
 			rep, err := replication.NewReplicationSet(
+<<<<<<< HEAD
 				tableID, 0, map[string]*tablepb.TableStatus{
+=======
+				span, tablepb.Checkpoint{}, map[string]*tablepb.TableStatus{
+>>>>>>> 3b8d55b1cd (scheduler(ticdc): fix invlaid checkpoint when redo enabled (#9851))
 					captureID: {
 						TableID: tableID,
 						State:   tablepb.TableStateReplicating,

cdc/scheduler/internal/v3/keyspan/main_test.go

@@ -0,0 +1,24 @@
+// Copyright 2022 PingCAP, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package keyspan
+
+import (
+	"testing"
+
+	"github.com/pingcap/tiflow/pkg/leakutil"
+)
+
+func TestMain(m *testing.M) {
+	leakutil.SetUpLeakTest(m)
+}

cdc/scheduler/internal/v3/member/main_test.go

@@ -0,0 +1,24 @@
+// Copyright 2022 PingCAP, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package member
+
+import (
+	"testing"
+
+	"github.com/pingcap/tiflow/pkg/leakutil"
+)
+
+func TestMain(m *testing.M) {
+	leakutil.SetUpLeakTest(m)
+}

cdc/scheduler/internal/v3/replication/main_test.go

@@ -0,0 +1,24 @@
+// Copyright 2022 PingCAP, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package replication
+
+import (
+	"testing"
+
+	"github.com/pingcap/tiflow/pkg/leakutil"
+)
+
+func TestMain(m *testing.M) {
+	leakutil.SetUpLeakTest(m)
+}

cdc/scheduler/internal/v3/replication/replication_manager.go

@@ -57,11 +57,24 @@ type MoveTable struct {
 
 // AddTable is a schedule task for adding a table.
 type AddTable struct {
+<<<<<<< HEAD
 	TableID      model.TableID
 	CaptureID    model.CaptureID
 	CheckpointTs model.Ts
 }
 
+=======
+	Span       tablepb.Span
+	CaptureID  model.CaptureID
+	Checkpoint tablepb.Checkpoint
+}
+
+func (t AddTable) String() string {
+	return fmt.Sprintf("AddTable, span: %s, capture: %s, checkpointTs: %d, resolvedTs: %d",
+		t.Span.String(), t.CaptureID, t.Checkpoint.CheckpointTs, t.Checkpoint.ResolvedTs)
+}
+
+>>>>>>> 3b8d55b1cd (scheduler(ticdc): fix invlaid checkpoint when redo enabled (#9851))
 // RemoveTable is a schedule task for removing a table.
 type RemoveTable struct {
 	TableID   model.TableID
@@ -151,11 +164,25 @@ func (r *Manager) HandleCaptureChanges(
 				tableStatus[table.TableID][captureID] = &table
 			}
 		}
+<<<<<<< HEAD
 		for tableID, status := range tableStatus {
 			table, err := NewReplicationSet(
 				tableID, checkpointTs, status, r.changefeedID)
 			if err != nil {
 				return nil, errors.Trace(err)
+=======
+		var err error
+		spanStatusMap.Ascend(func(span tablepb.Span, status map[string]*tablepb.TableStatus) bool {
+			checkpoint := tablepb.Checkpoint{
+				CheckpointTs: checkpointTs,
+				// Note that the real resolved ts is stored in the status.
+				ResolvedTs: checkpointTs,
+			}
+			table, err1 := NewReplicationSet(span, checkpoint, status, r.changefeedID)
+			if err1 != nil {
+				err = errors.Trace(err1)
+				return false
+>>>>>>> 3b8d55b1cd (scheduler(ticdc): fix invlaid checkpoint when redo enabled (#9851))
 			}
 			r.tables[tableID] = table
 		}
@@ -372,10 +399,16 @@ func (r *Manager) handleAddTableTask(
 ) ([]*schedulepb.Message, error) {
 	r.acceptAddTableTask++
 	var err error
+<<<<<<< HEAD
 	table := r.tables[task.TableID]
 	if table == nil {
 		table, err = NewReplicationSet(
 			task.TableID, task.CheckpointTs, nil, r.changefeedID)
+=======
+	table, ok := r.spans.Get(task.Span)
+	if !ok {
+		table, err = NewReplicationSet(task.Span, task.Checkpoint, nil, r.changefeedID)
+>>>>>>> 3b8d55b1cd (scheduler(ticdc): fix invlaid checkpoint when redo enabled (#9851))
 		if err != nil {
 			return nil, errors.Trace(err)
 		}