From 28390dfe09208000ec221bd4c81990f3f2ba8841 Mon Sep 17 00:00:00 2001 From: wayblink Date: Fri, 10 Mar 2023 11:03:52 +0800 Subject: [PATCH] Add bulkinsert integration test (#22621) Signed-off-by: wayblink --- internal/datacoord/server.go | 1 + internal/querynode/impl.go | 26 +- internal/rootcoord/root_coord.go | 3 + tests/integration/bulkinsert_test.go | 337 ++++++++++++++++++++++++++ tests/integration/meta_watcher.go | 2 +- tests/integration/minicluster.go | 31 ++- tests/integration/minicluster_test.go | 10 + 7 files changed, 390 insertions(+), 20 deletions(-) create mode 100644 tests/integration/bulkinsert_test.go diff --git a/internal/datacoord/server.go b/internal/datacoord/server.go index 78b59ba562..ee519ede46 100644 --- a/internal/datacoord/server.go +++ b/internal/datacoord/server.go @@ -937,6 +937,7 @@ func (s *Server) Stop() error { s.stopCompactionTrigger() s.stopCompactionHandler() } + s.indexBuilder.Stop() return nil } diff --git a/internal/querynode/impl.go b/internal/querynode/impl.go index 69fe83575f..a472eaa586 100644 --- a/internal/querynode/impl.go +++ b/internal/querynode/impl.go @@ -43,7 +43,6 @@ import ( "github.com/milvus-io/milvus/internal/proto/querypb" "github.com/milvus-io/milvus/internal/util/commonpbutil" "github.com/milvus-io/milvus/internal/util/metricsinfo" - "github.com/milvus-io/milvus/internal/util/paramtable" "github.com/milvus-io/milvus/internal/util/sessionutil" "github.com/milvus-io/milvus/internal/util/timerecord" "github.com/milvus-io/milvus/internal/util/typeutil" @@ -279,7 +278,7 @@ func (node *QueryNode) getStatisticsWithDmlChannel(ctx context.Context, req *que // WatchDmChannels create consumers on dmChannels to receive Incremental data，which is the important part of real-time query func (node *QueryNode) WatchDmChannels(ctx context.Context, in *querypb.WatchDmChannelsRequest) (*commonpb.Status, error) { - nodeID := paramtable.GetNodeID() + nodeID := node.session.ServerID // check node healthy if !node.lifetime.Add(commonpbutil.IsHealthy) { err := fmt.Errorf("query node %d is not ready", nodeID) @@ -374,7 +373,7 @@ func (node *QueryNode) UnsubDmChannel(ctx context.Context, req *querypb.UnsubDmC zap.String("channel", req.GetChannelName()), ) // check node healthy - nodeID := paramtable.GetNodeID() + nodeID := node.session.ServerID if !node.lifetime.Add(commonpbutil.IsHealthyOrStopping) { err := fmt.Errorf("query node %d is not ready", nodeID) status := &commonpb.Status{ @@ -434,7 +433,8 @@ func (node *QueryNode) UnsubDmChannel(ctx context.Context, req *querypb.UnsubDmC // LoadSegments load historical data into query node, historical data can be vector data or index func (node *QueryNode) LoadSegments(ctx context.Context, in *querypb.LoadSegmentsRequest) (*commonpb.Status, error) { - nodeID := paramtable.GetNodeID() + nodeID := node.session.ServerID + log.Info("wayblink", zap.Int64("nodeID", nodeID)) // check node healthy if !node.lifetime.Add(commonpbutil.IsHealthy) { err := fmt.Errorf("query node %d is not ready", nodeID) @@ -615,7 +615,7 @@ func (node *QueryNode) ReleasePartitions(ctx context.Context, in *querypb.Releas // ReleaseSegments remove the specified segments from query node according segmentIDs, partitionIDs, and collectionID func (node *QueryNode) ReleaseSegments(ctx context.Context, in *querypb.ReleaseSegmentsRequest) (*commonpb.Status, error) { - nodeID := paramtable.GetNodeID() + nodeID := node.session.ServerID if !node.lifetime.Add(commonpbutil.IsHealthyOrStopping) { err := fmt.Errorf("query node %d is not ready", nodeID) status := &commonpb.Status{ @@ -712,7 +712,7 @@ func filterSegmentInfo(segmentInfos []*querypb.SegmentInfo, segmentIDs map[int64 // Search performs replica search tasks. func (node *QueryNode) Search(ctx context.Context, req *querypb.SearchRequest) (*internalpb.SearchResults, error) { - nodeID := paramtable.GetNodeID() + nodeID := node.session.ServerID if !node.IsStandAlone && req.GetReq().GetBase().GetTargetID() != nodeID { return &internalpb.SearchResults{ Status: &commonpb.Status{ @@ -794,7 +794,7 @@ func (node *QueryNode) Search(ctx context.Context, req *querypb.SearchRequest) ( } func (node *QueryNode) searchWithDmlChannel(ctx context.Context, req *querypb.SearchRequest, dmlChannel string) (*internalpb.SearchResults, error) { - nodeID := paramtable.GetNodeID() + nodeID := node.session.ServerID metrics.QueryNodeSQCount.WithLabelValues(fmt.Sprint(nodeID), metrics.SearchLabel, metrics.TotalLabel).Inc() failRet := &internalpb.SearchResults{ Status: &commonpb.Status{ @@ -919,7 +919,7 @@ func (node *QueryNode) searchWithDmlChannel(ctx context.Context, req *querypb.Se } func (node *QueryNode) queryWithDmlChannel(ctx context.Context, req *querypb.QueryRequest, dmlChannel string) (*internalpb.RetrieveResults, error) { - nodeID := paramtable.GetNodeID() + nodeID := node.session.ServerID metrics.QueryNodeSQCount.WithLabelValues(fmt.Sprint(nodeID), metrics.QueryLabel, metrics.TotalLabel).Inc() failRet := &internalpb.RetrieveResults{ Status: &commonpb.Status{ @@ -1055,7 +1055,7 @@ func (node *QueryNode) Query(ctx context.Context, req *querypb.QueryRequest) (*i zap.Uint64("guaranteeTimestamp", req.Req.GetGuaranteeTimestamp()), zap.Uint64("timeTravel", req.GetReq().GetTravelTimestamp())) - nodeID := paramtable.GetNodeID() + nodeID := node.session.ServerID if req.GetReq().GetBase().GetTargetID() != nodeID { return &internalpb.RetrieveResults{ Status: &commonpb.Status{ @@ -1158,7 +1158,7 @@ func (node *QueryNode) SyncReplicaSegments(ctx context.Context, req *querypb.Syn // ShowConfigurations returns the configurations of queryNode matching req.Pattern func (node *QueryNode) ShowConfigurations(ctx context.Context, req *internalpb.ShowConfigurationsRequest) (*internalpb.ShowConfigurationsResponse, error) { - nodeID := paramtable.GetNodeID() + nodeID := node.session.ServerID if !node.lifetime.Add(commonpbutil.IsHealthyOrStopping) { log.Warn("QueryNode.ShowConfigurations failed", zap.Int64("nodeId", nodeID), @@ -1195,7 +1195,7 @@ func (node *QueryNode) ShowConfigurations(ctx context.Context, req *internalpb.S // GetMetrics return system infos of the query node, such as total memory, memory usage, cpu usage ... func (node *QueryNode) GetMetrics(ctx context.Context, req *milvuspb.GetMetricsRequest) (*milvuspb.GetMetricsResponse, error) { - nodeID := paramtable.GetNodeID() + nodeID := node.session.ServerID if !node.lifetime.Add(commonpbutil.IsHealthyOrStopping) { log.Ctx(ctx).Warn("QueryNode.GetMetrics failed", zap.Int64("nodeId", nodeID), @@ -1260,7 +1260,7 @@ func (node *QueryNode) GetMetrics(ctx context.Context, req *milvuspb.GetMetricsR } func (node *QueryNode) GetDataDistribution(ctx context.Context, req *querypb.GetDataDistributionRequest) (*querypb.GetDataDistributionResponse, error) { - nodeID := paramtable.GetNodeID() + nodeID := node.session.ServerID log := log.With( zap.Int64("msgID", req.GetBase().GetMsgID()), zap.Int64("nodeID", nodeID), @@ -1355,7 +1355,7 @@ func (node *QueryNode) GetDataDistribution(ctx context.Context, req *querypb.Get func (node *QueryNode) SyncDistribution(ctx context.Context, req *querypb.SyncDistributionRequest) (*commonpb.Status, error) { log := log.Ctx(ctx).With(zap.Int64("collectionID", req.GetCollectionID()), zap.String("channel", req.GetChannel())) - nodeID := paramtable.GetNodeID() + nodeID := node.session.ServerID // check node healthy if !node.lifetime.Add(commonpbutil.IsHealthyOrStopping) { err := fmt.Errorf("query node %d is not ready", nodeID) diff --git a/internal/rootcoord/root_coord.go b/internal/rootcoord/root_coord.go index a509c1fed0..888d504660 100644 --- a/internal/rootcoord/root_coord.go +++ b/internal/rootcoord/root_coord.go @@ -696,6 +696,9 @@ func (c *Core) Stop() error { c.stopExecutor() c.stopScheduler() c.cancelIfNotNil() + if c.quotaCenter != nil { + c.quotaCenter.stop() + } c.wg.Wait() c.revokeSession() return nil diff --git a/tests/integration/bulkinsert_test.go b/tests/integration/bulkinsert_test.go new file mode 100644 index 0000000000..7772f8395d --- /dev/null +++ b/tests/integration/bulkinsert_test.go @@ -0,0 +1,337 @@ +// Licensed to the LF AI & Data foundation under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you 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, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package integration + +import ( + "context" + "fmt" + "strconv" + "testing" + "time" + + "github.com/cockroachdb/errors" + "github.com/milvus-io/milvus-proto/go-api/commonpb" + "github.com/milvus-io/milvus-proto/go-api/milvuspb" + "github.com/milvus-io/milvus-proto/go-api/schemapb" + "github.com/milvus-io/milvus/internal/common" + "github.com/milvus-io/milvus/internal/log" + "github.com/milvus-io/milvus/internal/util/distance" + "github.com/milvus-io/milvus/internal/util/funcutil" + "github.com/milvus-io/milvus/internal/util/importutil" + + "github.com/golang/protobuf/proto" + "github.com/stretchr/testify/assert" + "go.uber.org/zap" +) + +const ( + TempFilesPath = "/tmp/integration_test/import/" + Dim = 128 +) + +// test bulk insert E2E +// 1, create collection with a vector column and a varchar column +// 2, generate numpy files +// 3, import +// 4, create index +// 5, load +// 6, search +func TestBulkInsert(t *testing.T) { + ctx := context.Background() + c, err := StartMiniCluster(ctx) + assert.NoError(t, err) + err = c.Start() + assert.NoError(t, err) + defer c.Stop() + assert.NoError(t, err) + + prefix := "TestBulkInsert" + dbName := "" + collectionName := prefix + funcutil.GenRandomStr() + int64Field := "int64" + floatVecField := "embeddings" + scalarField := "image_path" + dim := 128 + + constructCollectionSchema := func() *schemapb.CollectionSchema { + pk := &schemapb.FieldSchema{ + Name: int64Field, + IsPrimaryKey: true, + Description: "pk", + DataType: schemapb.DataType_Int64, + TypeParams: nil, + IndexParams: nil, + AutoID: true, + } + fVec := &schemapb.FieldSchema{ + Name: floatVecField, + IsPrimaryKey: false, + Description: "", + DataType: schemapb.DataType_FloatVector, + TypeParams: []*commonpb.KeyValuePair{ + { + Key: "dim", + Value: strconv.Itoa(dim), + }, + }, + IndexParams: nil, + AutoID: false, + } + scalar := &schemapb.FieldSchema{ + Name: scalarField, + IsPrimaryKey: false, + Description: "", + DataType: schemapb.DataType_VarChar, + TypeParams: []*commonpb.KeyValuePair{ + { + Key: "max_length", + Value: "65535", + }, + }, + IndexParams: nil, + AutoID: false, + } + return &schemapb.CollectionSchema{ + Name: collectionName, + Description: "", + AutoID: false, + Fields: []*schemapb.FieldSchema{ + pk, + fVec, + scalar, + }, + } + } + schema := constructCollectionSchema() + marshaledSchema, err := proto.Marshal(schema) + assert.NoError(t, err) + + createCollectionStatus, err := c.proxy.CreateCollection(ctx, &milvuspb.CreateCollectionRequest{ + DbName: dbName, + CollectionName: collectionName, + Schema: marshaledSchema, + ShardsNum: 2, + }) + assert.NoError(t, err) + if createCollectionStatus.GetErrorCode() != commonpb.ErrorCode_Success { + log.Warn("createCollectionStatus fail reason", zap.String("reason", createCollectionStatus.GetReason())) + } + assert.Equal(t, createCollectionStatus.GetErrorCode(), commonpb.ErrorCode_Success) + + log.Info("CreateCollection result", zap.Any("createCollectionStatus", createCollectionStatus)) + showCollectionsResp, err := c.proxy.ShowCollections(ctx, &milvuspb.ShowCollectionsRequest{}) + assert.NoError(t, err) + assert.Equal(t, showCollectionsResp.GetStatus().GetErrorCode(), commonpb.ErrorCode_Success) + log.Info("ShowCollections result", zap.Any("showCollectionsResp", showCollectionsResp)) + + err = GenerateNumpyFile(c.chunkManager.RootPath()+"/"+"embeddings.npy", 100, schemapb.DataType_FloatVector, []*commonpb.KeyValuePair{ + { + Key: "dim", + Value: strconv.Itoa(Dim), + }, + }) + assert.NoError(t, err) + err = GenerateNumpyFile(c.chunkManager.RootPath()+"/"+"image_path.npy", 100, schemapb.DataType_VarChar, []*commonpb.KeyValuePair{ + { + Key: "max_length", + Value: strconv.Itoa(65535), + }, + }) + assert.NoError(t, err) + + bulkInsertFiles := []string{ + c.chunkManager.RootPath() + "/" + "embeddings.npy", + c.chunkManager.RootPath() + "/" + "image_path.npy", + } + + health1, err := c.dataCoord.CheckHealth(ctx, &milvuspb.CheckHealthRequest{}) + assert.NoError(t, err) + log.Info("dataCoord health", zap.Any("health1", health1)) + importResp, err := c.proxy.Import(ctx, &milvuspb.ImportRequest{ + CollectionName: collectionName, + Files: bulkInsertFiles, + }) + assert.NoError(t, err) + log.Info("Import result", zap.Any("importResp", importResp), zap.Int64s("tasks", importResp.GetTasks())) + + tasks := importResp.GetTasks() + for _, task := range tasks { + loop: + for { + importTaskState, err := c.proxy.GetImportState(ctx, &milvuspb.GetImportStateRequest{ + Task: task, + }) + assert.NoError(t, err) + switch importTaskState.GetState() { + case commonpb.ImportState_ImportCompleted: + break loop + case commonpb.ImportState_ImportFailed: + break loop + case commonpb.ImportState_ImportFailedAndCleaned: + break loop + default: + log.Info("import task state", zap.Int64("id", task), zap.String("state", importTaskState.GetState().String())) + time.Sleep(time.Second * time.Duration(3)) + continue + } + } + } + + health2, err := c.dataCoord.CheckHealth(ctx, &milvuspb.CheckHealthRequest{}) + assert.NoError(t, err) + log.Info("dataCoord health", zap.Any("health2", health2)) + + segments, err := c.metaWatcher.ShowSegments() + assert.NoError(t, err) + assert.NotEmpty(t, segments) + for _, segment := range segments { + log.Info("ShowSegments result", zap.String("segment", segment.String())) + } + + // create index + createIndexStatus, err := c.proxy.CreateIndex(ctx, &milvuspb.CreateIndexRequest{ + CollectionName: collectionName, + FieldName: floatVecField, + IndexName: "_default", + ExtraParams: []*commonpb.KeyValuePair{ + { + Key: "dim", + Value: strconv.Itoa(dim), + }, + { + Key: common.MetricTypeKey, + Value: distance.L2, + }, + { + Key: "index_type", + Value: "HNSW", + }, + { + Key: "M", + Value: "64", + }, + { + Key: "efConstruction", + Value: "512", + }, + }, + }) + if createIndexStatus.GetErrorCode() != commonpb.ErrorCode_Success { + log.Warn("createIndexStatus fail reason", zap.String("reason", createIndexStatus.GetReason())) + } + assert.NoError(t, err) + assert.Equal(t, commonpb.ErrorCode_Success, createIndexStatus.GetErrorCode()) + + // load + loadStatus, err := c.proxy.LoadCollection(ctx, &milvuspb.LoadCollectionRequest{ + DbName: dbName, + CollectionName: collectionName, + }) + assert.NoError(t, err) + if loadStatus.GetErrorCode() != commonpb.ErrorCode_Success { + log.Warn("loadStatus fail reason", zap.String("reason", loadStatus.GetReason())) + } + assert.Equal(t, commonpb.ErrorCode_Success, loadStatus.GetErrorCode()) + for { + loadProgress, err := c.proxy.GetLoadingProgress(ctx, &milvuspb.GetLoadingProgressRequest{ + CollectionName: collectionName, + }) + if err != nil { + panic("GetLoadingProgress fail") + } + if loadProgress.GetProgress() == 100 { + break + } + time.Sleep(500 * time.Millisecond) + } + + // search + expr := fmt.Sprintf("%s > 0", "int64") + nq := 10 + topk := 10 + roundDecimal := -1 + nprobe := 10 + + searchReq := constructSearchRequest("", collectionName, expr, + floatVecField, nq, dim, nprobe, topk, roundDecimal) + + searchResult, err := c.proxy.Search(ctx, searchReq) + + if searchResult.GetStatus().GetErrorCode() != commonpb.ErrorCode_Success { + log.Warn("searchResult fail reason", zap.String("reason", searchResult.GetStatus().GetReason())) + } + assert.NoError(t, err) + assert.Equal(t, commonpb.ErrorCode_Success, searchResult.GetStatus().GetErrorCode()) + + log.Info("======================") + log.Info("======================") + log.Info("TestBulkInsert succeed") + log.Info("======================") + log.Info("======================") +} + +func GenerateNumpyFile(filePath string, rowCount int, dType schemapb.DataType, typeParams []*commonpb.KeyValuePair) error { + if dType == schemapb.DataType_VarChar { + var data []string + for i := 0; i < rowCount; i++ { + data = append(data, "str") + } + err := importutil.CreateNumpyFile(filePath, data) + if err != nil { + return err + } + } + if dType == schemapb.DataType_FloatVector { + dimStr, ok := funcutil.KeyValuePair2Map(typeParams)["dim"] + if !ok { + return errors.New("FloatVector field needs dim parameter") + } + dim, err := strconv.Atoi(dimStr) + if err != nil { + return err + } + //data := make([][]float32, rowCount) + var data [][Dim]float32 + for i := 0; i < rowCount; i++ { + vec := [Dim]float32{} + for j := 0; j < dim; j++ { + vec[j] = 1.1 + } + //v := reflect.Indirect(reflect.ValueOf(vec)) + //log.Info("type", zap.Any("type", v.Kind())) + data = append(data, vec) + //v2 := reflect.Indirect(reflect.ValueOf(data)) + //log.Info("type", zap.Any("type", v2.Kind())) + } + err = importutil.CreateNumpyFile(filePath, data) + if err != nil { + return err + } + } + return nil +} + +func TestGenerateNumpyFile(t *testing.T) { + err := GenerateNumpyFile(TempFilesPath+"embeddings.npy", 100, schemapb.DataType_FloatVector, []*commonpb.KeyValuePair{ + { + Key: "dim", + Value: strconv.Itoa(Dim), + }, + }) + assert.NoError(t, err) + log.Error("err", zap.Error(err)) +} diff --git a/tests/integration/meta_watcher.go b/tests/integration/meta_watcher.go index fd09cb3281..f187bb95dd 100644 --- a/tests/integration/meta_watcher.go +++ b/tests/integration/meta_watcher.go @@ -50,7 +50,7 @@ func (watcher *EtcdMetaWatcher) ShowSessions() ([]*sessionutil.Session, error) { } func (watcher *EtcdMetaWatcher) ShowSegments() ([]*datapb.SegmentInfo, error) { - metaBasePath := path.Join(watcher.rootPath, "/meta/datacoord-meta/s/") + metaBasePath := path.Join(watcher.rootPath, "/meta/datacoord-meta/s/") + "/" return listSegments(watcher.etcdCli, metaBasePath, func(s *datapb.SegmentInfo) bool { return true }) diff --git a/tests/integration/minicluster.go b/tests/integration/minicluster.go index d0bb8c7873..be97f57a08 100644 --- a/tests/integration/minicluster.go +++ b/tests/integration/minicluster.go @@ -33,6 +33,7 @@ import ( querycoord "github.com/milvus-io/milvus/internal/querycoordv2" "github.com/milvus-io/milvus/internal/querynode" "github.com/milvus-io/milvus/internal/rootcoord" + "github.com/milvus-io/milvus/internal/storage" "github.com/milvus-io/milvus/internal/types" "github.com/milvus-io/milvus/internal/util/dependency" "github.com/milvus-io/milvus/internal/util/etcd" @@ -97,7 +98,8 @@ type MiniCluster struct { params map[string]string clusterConfig ClusterConfig - factory dependency.Factory + factory dependency.Factory + chunkManager storage.ChunkManager etcdCli *clientv3.Client @@ -135,6 +137,11 @@ func StartMiniCluster(ctx context.Context, opts ...Option) (cluster *MiniCluster if cluster.factory == nil { cluster.factory = dependency.NewDefaultFactory(true) + chunkManager, err := cluster.factory.NewPersistentStorageChunkManager(cluster.ctx) + if err != nil { + return nil, err + } + cluster.chunkManager = chunkManager } if cluster.etcdCli == nil { @@ -417,29 +424,40 @@ func (cluster *MiniCluster) Start() error { func (cluster *MiniCluster) Stop() error { log.Info("mini cluster stop") cluster.proxy.Stop() + log.Info("mini cluster proxy stopped") cluster.rootCoord.Stop() + log.Info("mini cluster rootCoord stopped") cluster.dataCoord.Stop() + log.Info("mini cluster dataCoord stopped") //cluster.indexCoord.Stop() cluster.queryCoord.Stop() + log.Info("mini cluster queryCoord stopped") for _, dataNode := range cluster.dataNodes { dataNode.Stop() } + log.Info("mini cluster datanodes stopped") for _, queryNode := range cluster.queryNodes { queryNode.Stop() } + log.Info("mini cluster indexnodes stopped") for _, indexNode := range cluster.indexNodes { indexNode.Stop() } + log.Info("mini cluster querynodes stopped") cluster.etcdCli.KV.Delete(cluster.ctx, Params.EtcdCfg.RootPath.GetValue(), clientv3.WithPrefix()) defer cluster.etcdCli.Close() - chunkManager, err := cluster.factory.NewPersistentStorageChunkManager(cluster.ctx) - if err != nil { - log.Warn("fail to create chunk manager to clean test data", zap.Error(err)) - } else { - chunkManager.RemoveWithPrefix(cluster.ctx, chunkManager.RootPath()) + + if cluster.chunkManager == nil { + chunkManager, err := cluster.factory.NewPersistentStorageChunkManager(cluster.ctx) + if err != nil { + log.Warn("fail to create chunk manager to clean test data", zap.Error(err)) + } else { + cluster.chunkManager = chunkManager + } } + cluster.chunkManager.RemoveWithPrefix(cluster.ctx, cluster.chunkManager.RootPath()) return nil } @@ -451,6 +469,7 @@ func DefaultParams() map[string]string { //"runtime.role": typeutil.StandaloneRole, Params.IntegrationTestCfg.IntegrationMode.Key: "true", Params.CommonCfg.StorageType.Key: "local", + Params.DataNodeCfg.MemoryForceSyncEnable.Key: "false", // local execution will print too many logs } } diff --git a/tests/integration/minicluster_test.go b/tests/integration/minicluster_test.go index a66f2085c1..9f05db47d9 100644 --- a/tests/integration/minicluster_test.go +++ b/tests/integration/minicluster_test.go @@ -26,6 +26,16 @@ import ( "github.com/stretchr/testify/assert" ) +func TestMiniClusterStartAndStop(t *testing.T) { + ctx := context.Background() + c, err := StartMiniCluster(ctx) + assert.NoError(t, err) + err = c.Start() + assert.NoError(t, err) + err = c.Stop() + assert.NoError(t, err) +} + func TestAddRemoveDataNode(t *testing.T) { ctx := context.Background() c, err := StartMiniCluster(ctx)