Add a timeout config for bulkinsert request (#25789)

Signed-off-by: jaime <yun.zhang@zilliz.com>

internal/datanode/services.go

@@ -24,6 +24,7 @@ import (
 	"fmt"
 	"path"
 	"strconv"
+	"time"
 
 	"github.com/cockroachdb/errors"
 	"github.com/milvus-io/milvus/pkg/util/tsoutil"
@@ -436,7 +437,7 @@ func (node *DataNode) Import(ctx context.Context, req *datapb.ImportTaskRequest)
 	importResult.Infos = append(importResult.Infos, &commonpb.KeyValuePair{Key: importutil.ProgressPercent, Value: "0"})
 
 	// Spawn a new context to ignore cancellation from parental context.
-	newCtx, cancel := context.WithTimeout(context.TODO(), ImportCallTimeout)
+	newCtx, cancel := context.WithTimeout(context.TODO(), paramtable.Get().DataNodeCfg.BulkInsertTimeoutSeconds.GetAsDuration(time.Second))
 	defer cancel()
 
 	// function to report import state to RootCoord.

pkg/util/paramtable/component_param.go

@@ -2339,6 +2339,9 @@ type dataNodeConfig struct {
 	// DataNode send timetick interval per collection
 	DataNodeTimeTickInterval ParamItem `refreshable:"false"`
 
+	// timeout for bulkinsert
+	BulkInsertTimeoutSeconds ParamItem `refreshable:"true"`
+
 	// Skip BF
 	SkipBFStatsLoad ParamItem `refreshable:"true"`
 }
@@ -2498,6 +2501,14 @@ func (p *dataNodeConfig) init(base *BaseTable) {
 		DefaultValue: "false",
 	}
 	p.SkipBFStatsLoad.Init(base.mgr)
+
+	p.BulkInsertTimeoutSeconds = ParamItem{
+		Key:          "datanode.bulkinsert.timeout.seconds",
+		Version:      "2.3.0",
+		PanicIfEmpty: false,
+		DefaultValue: "18000",
+	}
+	p.BulkInsertTimeoutSeconds.Init(base.mgr)
 }
 
 // /////////////////////////////////////////////////////////////////////////////

pkg/util/paramtable/component_param_test.go

@@ -389,6 +389,10 @@ func TestComponentParam(t *testing.T) {
 		period := Params.SyncPeriod
 		t.Logf("SyncPeriod: %v", period)
 		assert.Equal(t, 10*time.Minute, Params.SyncPeriod.GetAsDuration(time.Second))
+
+		bulkinsertTimeout := Params.BulkInsertTimeoutSeconds
+		t.Logf("BulkInsertTimeoutSeconds: %v", bulkinsertTimeout)
+		assert.Equal(t, "18000", Params.BulkInsertTimeoutSeconds.GetValue())
 	})
 
 	t.Run("test indexNodeConfig", func(t *testing.T) {