processor(ticdc): Extract all closures to make code maintainable (#8337)

ref #8132
pingcap · Feb 24, 2023 · c9f207d · c9f207d
1 parent 25110ef
commit c9f207d
Show file tree

Hide file tree

Showing 6 changed files with 414 additions and 163 deletions.
diff --git a/cdc/processor/sinkmanager/manager.go b/cdc/processor/sinkmanager/manager.go
@@ -341,8 +341,8 @@ func (m *SinkManager) generateSinkTasks() error {
 	// Task upperbound is limited by barrierTs and schemaResolvedTs.
 	// But receivedSorterResolvedTs can be less than barrierTs, in which case
 	// the table is just scheduled to this node.
-	getUpperBound := func(tableSink *tableSinkWrapper) engine.Position {
-		upperBoundTs := tableSink.getReceivedSorterResolvedTs()
+	getUpperBound := func(tableSinkReceivedSorterResolvedTs model.Ts) engine.Position {
+		upperBoundTs := tableSinkReceivedSorterResolvedTs
 
 		barrierTs := m.lastBarrierTs.Load()
 		if upperBoundTs > barrierTs {
@@ -406,7 +406,7 @@ func (m *SinkManager) generateSinkTasks() error {
 			tableSink := tables[i]
 			slowestTableProgress := progs[i]
 			lowerBound := slowestTableProgress.nextLowerBoundPos
-			upperBound := getUpperBound(tableSink)
+			upperBound := getUpperBound(tableSink.getReceivedSorterResolvedTs())
 
 			// The table has no available progress.
 			if lowerBound.Compare(upperBound) >= 0 {
@@ -493,8 +493,8 @@ func (m *SinkManager) generateSinkTasks() error {
 
 func (m *SinkManager) generateRedoTasks() error {
 	// We use the table's resolved ts as the upper bound to fetch events.
-	getUpperBound := func(tableSink *tableSinkWrapper) engine.Position {
-		upperBoundTs := tableSink.getReceivedSorterResolvedTs()
+	getUpperBound := func(tableSinkReceivedSorterResolvedTs model.Ts) engine.Position {
+		upperBoundTs := tableSinkReceivedSorterResolvedTs
 
 		// If a task carries events after schemaResolvedTs, mounter group threads
 		// can be blocked on waiting schemaResolvedTs get advanced.
@@ -552,7 +552,7 @@ func (m *SinkManager) generateRedoTasks() error {
 			tableSink := tables[i]
 			slowestTableProgress := progs[i]
 			lowerBound := slowestTableProgress.nextLowerBoundPos
-			upperBound := getUpperBound(tableSink)
+			upperBound := getUpperBound(tableSink.getReceivedSorterResolvedTs())
 
 			// The table has no available progress.
 			if lowerBound.Compare(upperBound) >= 0 {

diff --git a/cdc/processor/sinkmanager/redo_log_worker.go b/cdc/processor/sinkmanager/redo_log_worker.go
@@ -73,11 +73,11 @@ func (w *redoWorker) handleTasks(ctx context.Context, taskChan <-chan *redoTask)
 
 func (w *redoWorker) handleTask(ctx context.Context, task *redoTask) (finalErr error) {
 	lowerBound := task.lowerBound
-	upperBound := task.getUpperBound(task.tableSink)
+	upperBound := task.getUpperBound(task.tableSink.getReceivedSorterResolvedTs())
 	lowerPhs := oracle.GetTimeFromTS(lowerBound.CommitTs)
 	upperPhs := oracle.GetTimeFromTS(upperBound.CommitTs)
-	if upperPhs.Sub(lowerPhs) > maxTaskRange {
-		upperCommitTs := oracle.GoTimeToTS(lowerPhs.Add(maxTaskRange))
+	if upperPhs.Sub(lowerPhs) > maxTaskTimeRange {
+		upperCommitTs := oracle.GoTimeToTS(lowerPhs.Add(maxTaskTimeRange))
 		upperBound = engine.Position{
 			StartTs:  upperCommitTs - 1,
 			CommitTs: upperCommitTs,