From 19b3a40a7b6e1cf65c17d53e4c2c94b9f4131ab2 Mon Sep 17 00:00:00 2001
From: lance6716 <lance6716@gmail.com>
Date: Wed, 16 Feb 2022 14:43:39 +0800
Subject: [PATCH] loader(dm): skip feature for nil etcd client (#4599)

ref pingcap/tiflow#4287
---
 dm/loader/util.go | 22 +++++++++++++++++++++-
 1 file changed, 21 insertions(+), 1 deletion(-)

diff --git a/dm/loader/util.go b/dm/loader/util.go
index 9abd144d0ed..c963b4fb58c 100644
--- a/dm/loader/util.go
+++ b/dm/loader/util.go
@@ -117,7 +117,7 @@ func getMydumpMetadata(cli *clientv3.Client, cfg *config.SubTaskConfig, workerNa
 	loc, _, err := dumpling.ParseMetaData(metafile, cfg.Flavor)
 	if err != nil {
 		if os.IsNotExist(err) {
-			worker, _, err2 := ha.GetLoadTask(cli, cfg.Name, cfg.SourceID)
+			worker, err2 := getLoadTask(cli, cfg.Name, cfg.SourceID)
 			if err2 != nil {
 				log.L().Warn("get load task", log.ShortError(err2))
 			}
@@ -171,7 +171,12 @@ func cleanDumpFiles(cfg *config.SubTaskConfig) {
 }
 
 // putLoadTask is called when start restoring data, to put load worker in etcd.
+// This is no-op when the `cli` argument is nil.
 func putLoadTask(cli *clientv3.Client, cfg *config.SubTaskConfig, workerName string) error {
+	// some usage like DM as a library, we don't support this feature
+	if cli == nil {
+		return nil
+	}
 	_, err := ha.PutLoadTask(cli, cfg.Name, cfg.SourceID, workerName)
 	if err != nil {
 		return err
@@ -181,7 +186,12 @@ func putLoadTask(cli *clientv3.Client, cfg *config.SubTaskConfig, workerName str
 }
 
 // delLoadTask is called when finish restoring data, to delete load worker in etcd.
+// This is no-op when the `cli` argument is nil.
 func delLoadTask(cli *clientv3.Client, cfg *config.SubTaskConfig, workerName string) error {
+	// some usage like DM as a library, we don't support this feature
+	if cli == nil {
+		return nil
+	}
 	_, _, err := ha.DelLoadTask(cli, cfg.Name, cfg.SourceID)
 	if err != nil {
 		return err
@@ -189,3 +199,13 @@ func delLoadTask(cli *clientv3.Client, cfg *config.SubTaskConfig, workerName str
 	log.L().Info("delete load worker in etcd for full mode", zap.String("task", cfg.Name), zap.String("source", cfg.SourceID), zap.String("worker", workerName))
 	return nil
 }
+
+// getLoadTask gets the worker which in load stage for the source of the subtask.
+// It will return "" and no error when the `cli` argument is nil.
+func getLoadTask(cli *clientv3.Client, task, sourceID string) (string, error) {
+	if cli == nil {
+		return "", nil
+	}
+	name, _, err := ha.GetLoadTask(cli, task, sourceID)
+	return name, err
+}