server: honor scheduler concurrency for coordinator

coordinator/controller.go

@@ -853,6 +853,10 @@ func (c *Controller) ListChangefeeds(_ context.Context, keyspace string) ([]*con
 	return infos, statuses, nil
 }
 
+// GetChangefeed returns a copy of the changefeed info and the current status.
+// API callers mutate the returned info when validating update requests, so the
+// copy prevents those writes from racing with coordinator goroutines that read
+// the in-memory changefeed state.
 func (c *Controller) GetChangefeed(
 	_ context.Context,
 	changefeedDisplayName common.ChangeFeedDisplayName,
@@ -869,6 +873,11 @@ func (c *Controller) GetChangefeed(
 		return nil, nil, errors.ErrChangeFeedNotExists.GenWithStackByArgs(changefeedDisplayName.Name)
 	}
 
+	info, err := cf.GetInfo().Clone()
+	if err != nil {
+		return nil, nil, errors.Trace(err)
+	}
+
 	maintainerID := cf.GetNodeID()
 	nodeInfo := c.nodeManager.GetNodeInfo(maintainerID)
 	maintainerAddr := ""
@@ -877,7 +886,7 @@ func (c *Controller) GetChangefeed(
 	}
 	status := &config.ChangeFeedStatus{CheckpointTs: cf.GetStatus().CheckpointTs, LastSyncedTs: cf.GetStatus().LastSyncedTs, LogCoordinatorResolvedTs: cf.GetLogCoordinatorResolvedTs()}
 	status.SetMaintainerAddr(maintainerAddr)
-	return cf.GetInfo(), status, nil
+	return info, status, nil
 }
 
 // getChangefeed returns the changefeed by id, return nil if not found

coordinator/controller_test.go

@@ -15,6 +15,7 @@ package coordinator
 
 import (
 	"context"
+	"sync"
 	"testing"
 	"time"
 
@@ -329,6 +330,9 @@ func TestUpdateChangefeed(t *testing.T) {
 }
 
 func TestGetChangefeed(t *testing.T) {
+	// Scenario: API-facing GetChangefeed must not expose coordinator-owned mutable info.
+	// Steps: fetch a changefeed, mutate the returned copy, and verify the in-memory
+	// changefeed state is unchanged while status fields are still reported.
 	ctrl := gomock.NewController(t)
 	backend := mock_changefeed.NewMockBackend(ctrl)
 	changefeedDB := changefeed.NewChangefeedDB(1216)
@@ -353,10 +357,67 @@ func TestGetChangefeed(t *testing.T) {
 	require.Equal(t, ret.State, config.StateStopped)
 	require.Equal(t, uint64(1), status.CheckpointTs)
 
+	ret.SinkURI = "kafka://127.0.0.1:9092"
+	ret.Config = nil
+	storedInfo := changefeedDB.GetByID(cfID).GetInfo()
+	require.Equal(t, "mysql://127.0.0.1:3306", storedInfo.SinkURI)
+	require.NotNil(t, storedInfo.Config)
+
 	_, _, err = controller.GetChangefeed(context.Background(), common.NewChangeFeedDisplayName("test1", "default"))
 	require.True(t, errors.ErrChangeFeedNotExists.Equal(err))
 }
 
+func TestGetChangefeedReturnedInfoMutationDoesNotRaceWithStoredInfo(t *testing.T) {
+	// Scenario: API handlers may mutate the info returned from GetChangefeed while coordinator
+	// goroutines read the stored info. Steps: mutate the returned copy and read the stored
+	// info concurrently, then verify the stored fields remain unchanged.
+	ctrl := gomock.NewController(t)
+	backend := mock_changefeed.NewMockBackend(ctrl)
+	changefeedDB := changefeed.NewChangefeedDB(1216)
+	nodeManager := watcher.NewNodeManager(nil, nil)
+	controller := &Controller{
+		backend:      backend,
+		changefeedDB: changefeedDB,
+		nodeManager:  nodeManager,
+	}
+	cfID := common.NewChangeFeedIDWithName("test", common.DefaultKeyspaceName)
+	cf := changefeed.NewChangefeed(cfID, &config.ChangeFeedInfo{
+		ChangefeedID: cfID,
+		Config:       config.GetDefaultReplicaConfig(),
+		State:        config.StateStopped,
+		SinkURI:      "mysql://127.0.0.1:3306",
+	}, 1, true)
+	changefeedDB.AddStoppedChangefeed(cf)
+
+	ret, _, err := controller.GetChangefeed(context.Background(), cfID.DisplayName)
+	require.NoError(t, err)
+
+	var (
+		wg            sync.WaitGroup
+		storedChanged bool
+	)
+	wg.Add(2)
+	go func() {
+		defer wg.Done()
+		for i := 0; i < 1000; i++ {
+			ret.SinkURI = "kafka://127.0.0.1:9092"
+			ret.TargetTs = uint64(i + 1)
+		}
+	}()
+	go func() {
+		defer wg.Done()
+		storedInfo := changefeedDB.GetByID(cfID).GetInfo()
+		for i := 0; i < 1000; i++ {
+			if storedInfo.SinkURI != "mysql://127.0.0.1:3306" || storedInfo.TargetTs != 0 {
+				storedChanged = true
+				return
+			}
+		}
+	}()
+	wg.Wait()
+	require.False(t, storedChanged)
+}
+
 func TestRemoveChangefeed(t *testing.T) {
 	ctrl := gomock.NewController(t)
 	backend := mock_changefeed.NewMockBackend(ctrl)

server/module_election.go

@@ -38,19 +38,25 @@ type elector struct {
 	// election used for coordinator
 	election *concurrency.Election
 	// election used for log coordinator
-	logElection *concurrency.Election
-	svr         *server
+	logElection                     *concurrency.Election
+	coordinatorMaxTaskConcurrency   int
+	coordinatorCheckBalanceInterval time.Duration
+	svr                             *server
 }
 
-func NewElector(server *server) common.SubModule {
+// NewElector creates the coordinator elector with scheduler settings captured from server startup config.
+func NewElector(server *server, schedulerCfg *config.SchedulerConfig) common.SubModule {
 	election := concurrency.NewElection(server.session,
 		etcd.CaptureOwnerKey(server.EtcdClient.GetClusterID()))
 	logElection := concurrency.NewElection(server.session,
 		etcd.LogCoordinatorKey(server.EtcdClient.GetClusterID()))
+	maxTaskConcurrency, checkBalanceInterval := coordinatorSchedulerSettings(schedulerCfg)
 	return &elector{
-		election:    election,
-		logElection: logElection,
-		svr:         server,
+		election:                        election,
+		logElection:                     logElection,
+		coordinatorMaxTaskConcurrency:   maxTaskConcurrency,
+		coordinatorCheckBalanceInterval: checkBalanceInterval,
+		svr:                             server,
 	}
 }
 
@@ -140,8 +146,8 @@ func (e *elector) campaignCoordinator(ctx context.Context) error {
 			changefeed.NewEtcdBackend(e.svr.EtcdClient),
 			e.svr.EtcdClient.GetGCServiceID(),
 			coordinatorVersion,
-			10000,
-			time.Minute,
+			e.coordinatorMaxTaskConcurrency,
+			e.coordinatorCheckBalanceInterval,
 		)
 		e.svr.setCoordinator(co)
 		err = co.Run(ctx)
@@ -191,6 +197,10 @@ func (e *elector) campaignCoordinator(ctx context.Context) error {
 	}
 }
 
+func coordinatorSchedulerSettings(schedulerCfg *config.SchedulerConfig) (int, time.Duration) {
+	return schedulerCfg.MaxTaskConcurrency, time.Duration(schedulerCfg.CheckBalanceInterval)
+}
+
 func (e *elector) campaignLogCoordinator(ctx context.Context) error {
 	// Limit the frequency of elections to avoid putting too much pressure on the etcd server
 	rl := rate.NewLimiter(rate.Every(time.Second), 1 /* burst */)

server/module_election_test.go

@@ -0,0 +1,45 @@
+// Copyright 2026 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 server
+
+import (
+	"testing"
+	"time"
+
+	"github.com/pingcap/ticdc/pkg/config"
+	"github.com/stretchr/testify/require"
+)
+
+func TestCoordinatorSchedulerSettingsUsesCapturedConfig(t *testing.T) {
+	// Scenario: the coordinator should honor the scheduler config captured during server startup.
+	// Steps: install a different global config, read settings from an explicit scheduler config,
+	// and verify both the concurrency limit and balance interval come from the captured config.
+	original := config.GetGlobalServerConfig()
+	t.Cleanup(func() {
+		config.StoreGlobalServerConfig(original)
+	})
+
+	globalCfg := config.GetDefaultServerConfig()
+	globalCfg.Debug.Scheduler.MaxTaskConcurrency = 9
+	globalCfg.Debug.Scheduler.CheckBalanceInterval = config.TomlDuration(44 * time.Second)
+	config.StoreGlobalServerConfig(globalCfg)
+
+	cfg := config.GetDefaultServerConfig()
+	cfg.Debug.Scheduler.MaxTaskConcurrency = 3
+	cfg.Debug.Scheduler.CheckBalanceInterval = config.TomlDuration(22 * time.Second)
+
+	maxTaskConcurrency, checkBalanceInterval := coordinatorSchedulerSettings(cfg.Debug.Scheduler)
+	require.Equal(t, 3, maxTaskConcurrency)
+	require.Equal(t, 22*time.Second, checkBalanceInterval)
+}

server/server.go

@@ -214,7 +214,7 @@ func (c *server) initialize(ctx context.Context) error {
 
 	c.nodeModules = []common.SubModule{
 		nodeManager,
-		NewElector(c),
+		NewElector(c, conf.Debug.Scheduler),
 	}
 
 	c.subModules = []common.SubModule{