diff --git a/dm/dm/config/task.go b/dm/dm/config/task.go
index 091f93722c3..9922245c2b1 100644
--- a/dm/dm/config/task.go
+++ b/dm/dm/config/task.go
@@ -558,8 +558,10 @@ func (c *TaskConfig) adjust() error {
 				return terror.ErrConfigMydumperCfgNotFound.Generate(i, inst.MydumperConfigName)
 			}
 			globalConfigReferCount[configRefPrefixes[mydumperIdx]+inst.MydumperConfigName]++
-			inst.Mydumper = new(MydumperConfig)
-			*inst.Mydumper = *rule // ref mydumper config
+			if rule != nil {
+				inst.Mydumper = new(MydumperConfig)
+				*inst.Mydumper = *rule // ref mydumper config
+			}
 		}
 		if inst.Mydumper == nil {
 			if len(c.Mydumpers) != 0 {
@@ -586,8 +588,10 @@ func (c *TaskConfig) adjust() error {
 				return terror.ErrConfigLoaderCfgNotFound.Generate(i, inst.LoaderConfigName)
 			}
 			globalConfigReferCount[configRefPrefixes[loaderIdx]+inst.LoaderConfigName]++
-			inst.Loader = new(LoaderConfig)
-			*inst.Loader = *rule // ref loader config
+			if rule != nil {
+				inst.Loader = new(LoaderConfig)
+				*inst.Loader = *rule // ref loader config
+			}
 		}
 		if inst.Loader == nil {
 			if len(c.Loaders) != 0 {
@@ -606,8 +610,10 @@ func (c *TaskConfig) adjust() error {
 				return terror.ErrConfigSyncerCfgNotFound.Generate(i, inst.SyncerConfigName)
 			}
 			globalConfigReferCount[configRefPrefixes[syncerIdx]+inst.SyncerConfigName]++
-			inst.Syncer = new(SyncerConfig)
-			*inst.Syncer = *rule // ref syncer config
+			if rule != nil {
+				inst.Syncer = new(SyncerConfig)
+				*inst.Syncer = *rule // ref syncer config
+			}
 		}
 		if inst.Syncer == nil {
 			if len(c.Syncers) != 0 {
@@ -620,6 +626,21 @@ func (c *TaskConfig) adjust() error {
 			inst.Syncer.WorkerCount = inst.SyncerThread
 		}
 
+<<<<<<< HEAD
+=======
+		inst.ContinuousValidator = defaultValidatorConfig()
+		if inst.ContinuousValidatorConfigName != "" {
+			rule, ok := c.Validators[inst.ContinuousValidatorConfigName]
+			if !ok {
+				return terror.ErrContinuousValidatorCfgNotFound.Generate(i, inst.ContinuousValidatorConfigName)
+			}
+			globalConfigReferCount[configRefPrefixes[validatorIdx]+inst.ContinuousValidatorConfigName]++
+			if rule != nil {
+				inst.ContinuousValidator = *rule
+			}
+		}
+
+>>>>>>> 21608dce2 (dm: fix empty config cause dm-master panic (#5298))
 		// for backward compatible, set global config `ansi-quotes: true` if any syncer is true
 		if inst.Syncer.EnableANSIQuotes {
 			log.L().Warn("DM could discover proper ANSI_QUOTES, `enable-ansi-quotes` is no longer take effect")
@@ -673,7 +694,17 @@ func (c *TaskConfig) adjust() error {
 			unusedConfigs = append(unusedConfigs, mydumper)
 		}
 	}
+<<<<<<< HEAD
 	for loader := range c.Loaders {
+=======
+
+	for loader, cfg := range c.Loaders {
+		if cfg != nil {
+			if err1 := cfg.adjust(); err1 != nil {
+				return err1
+			}
+		}
+>>>>>>> 21608dce2 (dm: fix empty config cause dm-master panic (#5298))
 		if globalConfigReferCount[configRefPrefixes[loaderIdx]+loader] == 0 {
 			unusedConfigs = append(unusedConfigs, loader)
 		}
@@ -693,6 +724,7 @@ func (c *TaskConfig) adjust() error {
 		sort.Strings(unusedConfigs)
 		return terror.ErrConfigGlobalConfigsUnused.Generate(unusedConfigs)
 	}
+
 	// we postpone default time_zone init in each unit so we won't change the config value in task/sub_task config
 	if c.Timezone != "" {
 		if _, err := utils.ParseTimeZone(c.Timezone); err != nil {
diff --git a/dm/tests/dmctl_basic/check_list/check_task.sh b/dm/tests/dmctl_basic/check_list/check_task.sh
index 6f2200459b2..4845b5a9b13 100644
--- a/dm/tests/dmctl_basic/check_list/check_task.sh
+++ b/dm/tests/dmctl_basic/check_list/check_task.sh
@@ -64,3 +64,41 @@ function check_task_error_count() {
 		"\"failed\": 2" 1 \
 		"\"state\": \"fail\"" 0
 }
+<<<<<<< HEAD
+=======
+
+function check_task_only_warning() {
+	task_conf=$1
+	run_dm_ctl $WORK_DIR "127.0.0.1:$MASTER_PORT" \
+		"check-task $task_conf" \
+		"\"state\": \"warn\"" 1
+}
+
+function check_task_empty_dump_config() {
+	sed "/threads/d" $1 >/tmp/empty-dump.yaml
+	run_dm_ctl $WORK_DIR "127.0.0.1:$MASTER_PORT" \
+		"check-task /tmp/empty-dump.yaml" \
+		"pre-check is passed" 1
+}
+
+function check_task_empty_load_config() {
+	sed "/pool-size/d" $1 >/tmp/empty-load.yaml
+	cat /tmp/empty-load.yaml
+	run_dm_ctl $WORK_DIR "127.0.0.1:$MASTER_PORT" \
+		"check-task /tmp/empty-load.yaml" \
+		"pre-check is passed" 1
+}
+
+function check_task_empty_sync_config() {
+	sed "/worker-count/d" $1 >/tmp/empty-sync.yaml
+	run_dm_ctl $WORK_DIR "127.0.0.1:$MASTER_PORT" \
+		"check-task /tmp/empty-sync.yaml" \
+		"pre-check is passed" 1
+}
+
+function check_task_empty_config() {
+	check_task_empty_dump_config $1
+	check_task_empty_load_config $1
+	check_task_empty_sync_config $1
+}
+>>>>>>> 21608dce2 (dm: fix empty config cause dm-master panic (#5298))
diff --git a/dm/tests/dmctl_basic/check_list/start_task.sh b/dm/tests/dmctl_basic/check_list/start_task.sh
index acbb79cdf3c..7a51dd87702 100644
--- a/dm/tests/dmctl_basic/check_list/start_task.sh
+++ b/dm/tests/dmctl_basic/check_list/start_task.sh
@@ -11,3 +11,39 @@ function start_task_wrong_config_file() {
 		"start-task not_exists_config_file" \
 		"error in get file content" 1
 }
+<<<<<<< HEAD
+=======
+
+function start_task_wrong_start_time_format() {
+	task_conf=$1
+	run_dm_ctl $WORK_DIR "127.0.0.1:$MASTER_PORT" \
+		"start-task $task_conf --start-time '20060102 150405'" \
+		"error while parse start-time" 1
+}
+
+function start_task_not_pass_with_message() {
+	task_conf=$1
+	error_message=$2
+	run_dm_ctl $WORK_DIR "127.0.0.1:$MASTER_PORT" \
+		"start-task $task_conf" \
+		"$2" 1
+}
+
+function start_task_empty_config() {
+	cp $1 /tmp/empty-cfg.yaml
+	sed -i "/threads/d" /tmp/empty-cfg.yaml
+	sed -i "/pool-size/d" /tmp/empty-cfg.yaml
+	sed -i "/worker-count/d" /tmp/empty-cfg.yaml
+	run_dm_ctl $WORK_DIR "127.0.0.1:$MASTER_PORT" \
+		"start-task /tmp/empty-cfg.yaml" \
+		"\"result\": true" 2
+	run_dm_ctl $WORK_DIR "127.0.0.1:$MASTER_PORT" \
+		"config task empty-unit-task" \
+		"threads: 4" 1 \
+		"pool-size: 16" 1 \
+		"worker-count: 16" 1
+	run_dm_ctl $WORK_DIR "127.0.0.1:$MASTER_PORT" \
+		"stop-task $1" \
+		"\"result\": true" 2
+}
+>>>>>>> 21608dce2 (dm: fix empty config cause dm-master panic (#5298))
diff --git a/dm/tests/dmctl_basic/conf/empty-unit-task.yaml b/dm/tests/dmctl_basic/conf/empty-unit-task.yaml
new file mode 100644
index 00000000000..2eb41893d3d
--- /dev/null
+++ b/dm/tests/dmctl_basic/conf/empty-unit-task.yaml
@@ -0,0 +1,35 @@
+---
+name: empty-unit-task
+task-mode: all
+
+target-database:
+  host: "127.0.0.1"
+  port: 4000
+  user: "root"
+  password: ""
+
+mysql-instances:
+  - source-id: "mysql-replica-01"
+    mydumper-config-name: "global"
+    loader-config-name: "global"
+    syncer-config-name: "global"
+    block-allow-list:  "instance"
+
+mydumpers:
+  global:
+    threads: 4
+
+loaders:
+  global:
+    pool-size: 16
+
+syncers:
+  global:
+    worker-count: 32
+
+block-allow-list:
+  instance:
+    do-dbs: ["dmctl"]
+    do-tables:
+    -  db-name: "dmctl"
+       tbl-name: "~^t_[\\d]+"
\ No newline at end of file
diff --git a/dm/tests/dmctl_basic/run.sh b/dm/tests/dmctl_basic/run.sh
index 22c361f3074..55fd5a68564 100755
--- a/dm/tests/dmctl_basic/run.sh
+++ b/dm/tests/dmctl_basic/run.sh
@@ -270,6 +270,11 @@ function run() {
 	check_task_not_pass $cur/conf/dm-task2.yaml
 	check_task_error_count $cur/conf/dm-task3.yaml
 
+	echo "check task with empty unit config"
+	check_task_empty_config $cur/conf/empty-unit-task.yaml
+	echo "start task with empty unit config"
+	start_task_empty_config $cur/conf/empty-unit-task.yaml
+
 	cp $cur/conf/dm-task.yaml $WORK_DIR/dm-task-error-database-config.yaml
 	sed -i "s/password: \"\"/password: \"wrond password\"/g" $WORK_DIR/dm-task-error-database-config.yaml
 	check_task_error_database_config $WORK_DIR/dm-task-error-database-config.yaml