fix: err cannot be captured in defer in data_sync_service (milvus-io#…

…35448) milvus-io#35447 Signed-off-by: fengjun2016 <[email protected]>
tedxu · Aug 14, 2024 · 36d5986 · 36d5986
1 parent c42976e
commit 36d5986
Showing 1 changed file with 7 additions and 3 deletions.
diff --git a/internal/flushcommon/pipeline/data_sync_service.go b/internal/flushcommon/pipeline/data_sync_service.go
@@ -247,12 +247,15 @@ func getServiceWithChannel(initCtx context.Context, params *util.PipelineParams,
 
 	// init flowgraph
 	fg := flowgraph.NewTimeTickedFlowGraph(params.Ctx)
-	dmStreamNode, err := newDmInputNode(initCtx, params.DispClient, info.GetVchan().GetSeekPosition(), config, input)
+
+	var dmStreamNode *flowgraph.InputNode
+	dmStreamNode, err = newDmInputNode(initCtx, params.DispClient, info.GetVchan().GetSeekPosition(), config, input)
 	if err != nil {
 		return nil, err
 	}
 
-	ddNode, err := newDDNode(
+	var ddNode *ddNode
+	ddNode, err = newDDNode(
 		params.Ctx,
 		collectionID,
 		channelName,
@@ -267,7 +270,8 @@ func getServiceWithChannel(initCtx context.Context, params *util.PipelineParams,
 	}
 
 	writeNode := newWriteNode(params.Ctx, params.WriteBufferManager, ds.timetickSender, config)
-	ttNode, err := newTTNode(config, params.WriteBufferManager, params.CheckpointUpdater)
+	var ttNode *ttNode
+	ttNode, err = newTTNode(config, params.WriteBufferManager, params.CheckpointUpdater)
 	if err != nil {
 		return nil, err
 	}