ttl: fix a bug that a TTL task will not be taken by other tidb after …

…failover sometimes (#45023) (#45039) close #45022
pingcap · Jun 29, 2023 · 44d13a3 · 44d13a3
1 parent 731bd28
commit 44d13a3
Show file tree

Hide file tree

Showing 3 changed files with 18 additions and 10 deletions.
diff --git a/ttl/ttlworker/task_manager.go b/ttl/ttlworker/task_manager.go
@@ -366,7 +366,7 @@ func (m *taskManager) lockScanTask(se session.Session, task *cache.TTLTask, now
 		if err != nil {
 			return errors.Wrapf(err, "execute sql: %s", countRunningTasks)
 		}
-		if !m.meetTTLRunningTask(int(rows[0].GetInt64(0))) {
+		if !m.meetTTLRunningTask(int(rows[0].GetInt64(0)), task.Status) {
 			return errors.WithStack(errTooManyRunningTasks)
 		}
 
@@ -544,7 +544,12 @@ func (m *taskManager) reportMetrics() {
 	metrics.DeletingTaskCnt.Set(float64(deletingTaskCnt))
 }
 
-func (m *taskManager) meetTTLRunningTask(count int) bool {
+func (m *taskManager) meetTTLRunningTask(count int, taskStatus cache.TaskStatus) bool {
+	if taskStatus == cache.TaskStatusRunning {
+		// always return true for already running task because it is already included in count
+		return true
+	}
+
 	ttlRunningTask := variable.TTLRunningTasks.Load()
 	// `-1` is the auto value, means we should calculate the limit according to the count of TiKV
 	if ttlRunningTask != -1 {

diff --git a/ttl/ttlworker/task_manager_integration_test.go b/ttl/ttlworker/task_manager_integration_test.go
@@ -325,16 +325,19 @@ func TestMeetTTLRunningTasks(t *testing.T) {
 	tk := testkit.NewTestKit(t, store)
 
 	// -1, the default value, means the count of TiKV
-	require.True(t, dom.TTLJobManager().TaskManager().MeetTTLRunningTasks(2))
-	require.False(t, dom.TTLJobManager().TaskManager().MeetTTLRunningTasks(3))
+	require.True(t, dom.TTLJobManager().TaskManager().MeetTTLRunningTasks(2, cache.TaskStatusWaiting))
+	require.False(t, dom.TTLJobManager().TaskManager().MeetTTLRunningTasks(3, cache.TaskStatusWaiting))
+	require.True(t, dom.TTLJobManager().TaskManager().MeetTTLRunningTasks(2, cache.TaskStatusRunning))
 
 	// positive number means the limitation
 	tk.MustExec("set global tidb_ttl_running_tasks = 32")
-	require.False(t, dom.TTLJobManager().TaskManager().MeetTTLRunningTasks(32))
-	require.True(t, dom.TTLJobManager().TaskManager().MeetTTLRunningTasks(31))
+	require.False(t, dom.TTLJobManager().TaskManager().MeetTTLRunningTasks(32, cache.TaskStatusWaiting))
+	require.True(t, dom.TTLJobManager().TaskManager().MeetTTLRunningTasks(31, cache.TaskStatusWaiting))
+	require.True(t, dom.TTLJobManager().TaskManager().MeetTTLRunningTasks(32, cache.TaskStatusRunning))
 
 	// set it back to auto value
 	tk.MustExec("set global tidb_ttl_running_tasks = -1")
-	require.True(t, dom.TTLJobManager().TaskManager().MeetTTLRunningTasks(2))
-	require.False(t, dom.TTLJobManager().TaskManager().MeetTTLRunningTasks(3))
+	require.True(t, dom.TTLJobManager().TaskManager().MeetTTLRunningTasks(2, cache.TaskStatusWaiting))
+	require.False(t, dom.TTLJobManager().TaskManager().MeetTTLRunningTasks(3, cache.TaskStatusWaiting))
+	require.True(t, dom.TTLJobManager().TaskManager().MeetTTLRunningTasks(3, cache.TaskStatusRunning))
 }
diff --git a/ttl/ttlworker/task_manager_test.go b/ttl/ttlworker/task_manager_test.go
@@ -65,8 +65,8 @@ func (m *taskManager) GetRunningTasks() []*runningScanTask {
 }
 
 // MeetTTLRunningTasks is an exported version of meetTTLRunningTask
-func (m *taskManager) MeetTTLRunningTasks(count int) bool {
-	return m.meetTTLRunningTask(count)
+func (m *taskManager) MeetTTLRunningTasks(count int, taskStatus cache.TaskStatus) bool {
+	return m.meetTTLRunningTask(count, taskStatus)
 }
 
 // ReportTaskFinished is an exported version of reportTaskFinished