ttl: ttl use timer framework to trigger jobs (#45469)

close #45468

domain/domain.go

@@ -3093,7 +3093,7 @@ func (do *Domain) StartTTLJobManager() {
 			logutil.BgLogger().Info("ttlJobManager exited.")
 		}()
 
-		ttlJobManager := ttlworker.NewJobManager(do.ddl.GetID(), do.sysSessionPool, do.store, do.etcdClient)
+		ttlJobManager := ttlworker.NewJobManager(do.ddl.GetID(), do.sysSessionPool, do.store, do.etcdClient, do.ddl.OwnerManager().IsOwner)
 		do.ttlJobManager.Store(ttlJobManager)
 		ttlJobManager.Start()
 

timer/runtime/runtime.go

@@ -115,6 +115,13 @@ func (rt *TimerGroupRuntime) Start() {
 	go rt.loop()
 }
 
+// Running returns whether the runtime is running
+func (rt *TimerGroupRuntime) Running() bool {
+	rt.mu.Lock()
+	defer rt.mu.Unlock()
+	return rt.ctx != nil && rt.cancel != nil
+}
+
 func (rt *TimerGroupRuntime) initCtx() {
 	rt.ctx, rt.cancel = context.WithCancel(context.Background())
 }

timer/runtime/runtime_test.go

@@ -67,9 +67,11 @@ func TestRuntimeStartStop(t *testing.T) {
 		Build()
 
 	runtime.Start()
+	require.True(t, runtime.Running())
 	waitDone(timerProcessed, time.Minute)
 	go func() {
 		runtime.Stop()
+		require.False(t, runtime.Running())
 		cancel()
 	}()
 	waitDone(ctx.Done(), time.Minute)

timer/runtime/worker.go

@@ -26,7 +26,7 @@ import (
 )
 
 const (
-	workerRecvChanCap               = 8
+	workerRecvChanCap               = 128
 	workerRespChanCap               = 128
 	workerEventDefaultRetryInterval = 10 * time.Second
 	chanBlockInterval               = time.Second

ttl/client/command.go

@@ -74,8 +74,8 @@ type TriggerNewTTLJobTableResult struct {
 	DBName        string `json:"db_name"`
 	TableName     string `json:"table_name"`
 	PartitionName string `json:"partition_name,omitempty"`
-	JobID         string `json:"job_id"`
-	ErrorMessage  string `json:"error_message"`
+	JobID         string `json:"job_id,omitempty"`
+	ErrorMessage  string `json:"error_message,omitempty"`
 }
 
 // TriggerNewTTLJobResponse is the response detail for trigger_ttl_job command

ttl/client/notification.go

@@ -58,13 +58,41 @@ func NewMockNotificationClient() NotificationClient {
 }
 
 // Notify implements the NotificationClient
-func (c *mockClient) Notify(_ context.Context, typ string, data string) error {
+func (c *mockClient) Notify(ctx context.Context, typ string, data string) error {
 	c.Lock()
 	defer c.Unlock()
 
-	for _, ch := range c.notificationWatchers[typ] {
-		ch <- clientv3.WatchResponse{}
+	watchers, ok := c.notificationWatchers[typ]
+	if !ok {
+		return nil
 	}
+
+	var unsent []chan clientv3.WatchResponse
+loop:
+	for i, ch := range watchers {
+		select {
+		case <-ctx.Done():
+			return ctx.Err()
+		case ch <- clientv3.WatchResponse{}:
+		default:
+			unsent = make([]chan clientv3.WatchResponse, len(watchers), 0)
+			copy(unsent, watchers[i:])
+			break loop
+		}
+	}
+
+	if len(unsent) > 0 {
+		go func() {
+			for _, ch := range unsent {
+				select {
+				case <-ctx.Done():
+					return
+				case ch <- clientv3.WatchResponse{}:
+				}
+			}
+		}()
+	}
+
 	return nil
 }
 

ttl/ttlworker/BUILD.bazel

@@ -26,6 +26,7 @@ go_library(
         "//store/driver/error",
         "//timer/api",
         "//timer/runtime",
+        "//timer/tablestore",
         "//ttl/cache",
         "//ttl/client",
         "//ttl/metrics",
@@ -37,12 +38,13 @@ go_library(
         "//util/logutil",
         "//util/sqlexec",
         "//util/timeutil",
-        "@com_github_google_uuid//:uuid",
         "@com_github_ngaut_pools//:pools",
         "@com_github_pingcap_errors//:errors",
         "@com_github_pingcap_failpoint//:failpoint",
         "@com_github_tikv_client_go_v2//tikv",
+        "@com_github_tikv_client_go_v2//tikvrpc",
         "@io_etcd_go_etcd_client_v3//:client",
+        "@org_golang_x_exp//maps",
         "@org_golang_x_exp//slices",
         "@org_golang_x_time//rate",
         "@org_uber_go_multierr//:multierr",
@@ -67,7 +69,7 @@ go_test(
     embed = [":ttlworker"],
     flaky = True,
     race = "on",
-    shard_count = 41,
+    shard_count = 46,
     deps = [
         "//domain",
         "//infoschema",
@@ -99,6 +101,8 @@ go_test(
         "@com_github_stretchr_testify//mock",
         "@com_github_stretchr_testify//require",
         "@com_github_tikv_client_go_v2//testutils",
+        "@com_github_tikv_client_go_v2//tikv",
+        "@com_github_tikv_client_go_v2//tikvrpc",
         "@org_golang_x_time//rate",
         "@org_uber_go_atomic//:atomic",
         "@org_uber_go_zap//:zap",