importinto: fix data race in test (#47965)

close #47776
pingcap · Oct 26, 2023 · a7366f9 · a7366f9
1 parent 3a3a7d1
commit a7366f9
Show file tree

Hide file tree

Showing 3 changed files with 11 additions and 1 deletion.
diff --git a/br/pkg/lightning/mydump/loader.go b/br/pkg/lightning/mydump/loader.go
@@ -705,7 +705,8 @@ func calculateFileBytes(ctx context.Context,
 	}
 	defer reader.Close()
 
-	compressReader, err := storage.NewLimitedInterceptReader(reader, compressType, storage.DecompressConfig{}, offset)
+	decompressConfig := storage.DecompressConfig{ZStdDecodeConcurrency: 1}
+	compressReader, err := storage.NewLimitedInterceptReader(reader, compressType, decompressConfig, offset)
 	if err != nil {
 		return 0, 0, errors.Trace(err)
 	}

diff --git a/pkg/disttask/framework/dispatcher/dispatcher_manager.go b/pkg/disttask/framework/dispatcher/dispatcher_manager.go
@@ -127,6 +127,9 @@ func NewManager(ctx context.Context, taskMgr TaskManager, serverID string) (*Man
 
 // Start the dispatcherManager, start the dispatchTaskLoop to start multiple dispatchers.
 func (dm *Manager) Start() {
+	failpoint.Inject("disableDispatcherManager", func() {
+		failpoint.Return()
+	})
 	dm.wg.Run(dm.dispatchTaskLoop)
 	dm.wg.Run(dm.gcSubtaskHistoryTableLoop)
 	dm.wg.Run(dm.cleanUpLoop)

diff --git a/pkg/disttask/importinto/dispatcher_testkit_test.go b/pkg/disttask/importinto/dispatcher_testkit_test.go
@@ -150,6 +150,12 @@ func TestDispatcherExtLocalSort(t *testing.T) {
 }
 
 func TestDispatcherExtGlobalSort(t *testing.T) {
+	// Domain start dispatcher manager automatically, we need to disable it as
+	// we test import task management in this case.
+	require.NoError(t, failpoint.Enable("github.com/pingcap/tidb/pkg/disttask/framework/dispatcher/disableDispatcherManager", "return(true)"))
+	t.Cleanup(func() {
+		require.NoError(t, failpoint.Disable("github.com/pingcap/tidb/pkg/disttask/framework/dispatcher/disableDispatcherManager"))
+	})
 	store := testkit.CreateMockStore(t)
 	tk := testkit.NewTestKit(t, store)
 	pool := pools.NewResourcePool(func() (pools.Resource, error) {