disttask: use the correct session pool for dist tasks (#44981)

close pingcap/tidb#44983

disttask/framework/storage/task_table.go

@@ -47,7 +47,12 @@ type SessionExecutor interface {
 // TaskManager is the manager of global/sub task.
 type TaskManager struct {
 	ctx    context.Context
-	sePool *pools.ResourcePool
+	sePool sessionPool
+}
+
+type sessionPool interface {
+	Get() (pools.Resource, error)
+	Put(resource pools.Resource)
 }
 
 var _ SessionExecutor = &TaskManager{}
@@ -60,7 +65,7 @@ var (
 )
 
 // NewTaskManager creates a new task manager.
-func NewTaskManager(ctx context.Context, sePool *pools.ResourcePool) *TaskManager {
+func NewTaskManager(ctx context.Context, sePool sessionPool) *TaskManager {
 	ctx = util.WithInternalSourceType(ctx, kv.InternalDistTask)
 	return &TaskManager{
 		ctx:    ctx,

domain/domain.go

@@ -186,7 +186,6 @@ type Domain struct {
 
 	mdlCheckCh      chan struct{}
 	stopAutoAnalyze atomicutil.Bool
-	resourcePool    *pools.ResourcePool
 }
 
 type mdlCheckTableInfo struct {
@@ -1129,7 +1128,6 @@ func (do *Domain) Init(
 		return sysExecutorFactory(do)
 	}
 	sysCtxPool := pools.NewResourcePool(sysFac, 128, 128, resourceIdleTimeout)
-	do.resourcePool = sysCtxPool
 	ctx, cancelFunc := context.WithCancel(context.Background())
 	do.cancel = cancelFunc
 	var callback ddl.Callback
@@ -1655,7 +1653,7 @@ func (do *Domain) InitDistTaskLoop(ctx context.Context) error {
 		}
 	})
 
-	taskManager := storage.NewTaskManager(ctx, do.resourcePool)
+	taskManager := storage.NewTaskManager(ctx, do.sysSessionPool)
 	var serverID string
 	if intest.InTest {
 		do.InitInfo4Test()