From f536338bb6ab385d6bde43684c39fb71a6e72cde Mon Sep 17 00:00:00 2001 From: qupeng Date: Wed, 29 Nov 2023 21:14:48 +0800 Subject: [PATCH] cdc: fixes minor bugs #10168 and #10169 (#10170) close pingcap/tiflow#10168 Signed-off-by: qupeng --- cdc/kv/client.go | 4 ++++ cdc/processor/sinkmanager/tasks.go | 4 ++-- cdc/processor/sinkmanager/tasks_test.go | 2 +- 3 files changed, 7 insertions(+), 3 deletions(-) diff --git a/cdc/kv/client.go b/cdc/kv/client.go index a419c787ecd..b1055693fd5 100644 --- a/cdc/kv/client.go +++ b/cdc/kv/client.go @@ -73,6 +73,8 @@ const ( // failed region will be reloaded via `BatchLoadRegionsWithKeyRange` API. So we // don't need to force reload region anymore. regionScheduleReload = false + + scanRegionsConcurrency = 1024 ) // time interval to force kv client to terminate gRPC stream and reconnect @@ -432,6 +434,8 @@ func (s *eventFeedSession) eventFeed(ctx context.Context) error { g.Go(func() error { return s.logSlowRegions(ctx) }) g.Go(func() error { + g, ctx := errgroup.WithContext(ctx) + g.SetLimit(scanRegionsConcurrency) for { select { case <-ctx.Done(): diff --git a/cdc/processor/sinkmanager/tasks.go b/cdc/processor/sinkmanager/tasks.go index e4b94551191..dd415428c83 100644 --- a/cdc/processor/sinkmanager/tasks.go +++ b/cdc/processor/sinkmanager/tasks.go @@ -39,8 +39,8 @@ var ( maxUpdateIntervalSize = defaultMaxUpdateIntervalSize // Sink manager schedules table tasks based on lag. Limit the max task range - // can be helpful to reduce changefeed latency. - maxTaskTimeRange = 5 * time.Second + // can be helpful to reduce changefeed latency for large initial data. + maxTaskTimeRange = 30 * time.Minute ) // Used to record the progress of the table. diff --git a/cdc/processor/sinkmanager/tasks_test.go b/cdc/processor/sinkmanager/tasks_test.go index fac3788a07f..15a1f559259 100644 --- a/cdc/processor/sinkmanager/tasks_test.go +++ b/cdc/processor/sinkmanager/tasks_test.go @@ -37,7 +37,7 @@ func TestValidateAndAdjustBound(t *testing.T) { StartTs: 439333515018895365, CommitTs: 439333515018895366, }, - taskTimeRange: 10 * time.Second, + taskTimeRange: 60 * time.Minute, expectAdjust: true, }, {