diff --git a/bin/start_fe.sh b/bin/start_fe.sh
index 45b883ef1c5504..941e8c5a52daa4 100755
--- a/bin/start_fe.sh
+++ b/bin/start_fe.sh
@@ -36,6 +36,7 @@ OPTS="$(getopt \
     -l 'recovery_journal_id:' \
     -l 'console' \
     -l 'cluster_snapshot:' \
+    -l 'drop_backends' \
     -- "$@")"
 
 eval set -- "${OPTS}"
@@ -50,6 +51,7 @@ declare -a HELPER_ARGS=()
 declare -a METADATA_FAILURE_RECOVERY_ARGS=()
 declare -a RECOVERY_JOURNAL_ID_ARGS=()
 declare -a CLUSTER_SNAPSHOT_ARGS=()
+declare -a DROP_BACKENDS_ARGS=()
 while true; do
     case "$1" in
     --daemon)
@@ -85,6 +87,10 @@ while true; do
         CLUSTER_SNAPSHOT_ARGS=("--cluster_snapshot" "$2")
         shift 2
         ;;
+    --drop_backends)
+        DROP_BACKENDS_ARGS=("--drop_backends")
+        shift
+        ;;
     --)
         shift
         break
@@ -93,7 +99,7 @@ while true; do
         echo "Internal error"
         exit 1
         ;;
-    esac
+esac
 done
 
 DORIS_HOME="$(
@@ -433,12 +439,12 @@ if [[ "${IMAGE_TOOL}" -eq 1 ]]; then
         echo "Internal error, USE IMAGE_TOOL like: ./start_fe.sh --image image_path"
     fi
 elif [[ "${RUN_DAEMON}" -eq 1 ]]; then
-    nohup ${LIMIT:+${LIMIT}} "${JAVA}" ${final_java_opt:+${final_java_opt}} -XX:-OmitStackTraceInFastThrow -XX:OnOutOfMemoryError="kill -9 %p" ${coverage_opt:+${coverage_opt}} org.apache.doris.DorisFE "${HELPER_ARGS[@]}" "${METADATA_FAILURE_RECOVERY_ARGS[@]}" "${RECOVERY_JOURNAL_ID_ARGS[@]}" "${CLUSTER_SNAPSHOT_ARGS[@]}" "$@" >>"${STDOUT_LOGGER}" 2>&1 </dev/null &
+    nohup ${LIMIT:+${LIMIT}} "${JAVA}" ${final_java_opt:+${final_java_opt}} -XX:-OmitStackTraceInFastThrow -XX:OnOutOfMemoryError="kill -9 %p" ${coverage_opt:+${coverage_opt}} org.apache.doris.DorisFE "${HELPER_ARGS[@]}" "${METADATA_FAILURE_RECOVERY_ARGS[@]}" "${RECOVERY_JOURNAL_ID_ARGS[@]}" "${CLUSTER_SNAPSHOT_ARGS[@]}" "${DROP_BACKENDS_ARGS[@]}" "$@" >>"${STDOUT_LOGGER}" 2>&1 </dev/null &
 elif [[ "${RUN_CONSOLE}" -eq 1 ]]; then
     export DORIS_LOG_TO_STDERR=1
-    ${LIMIT:+${LIMIT}} "${JAVA}" ${final_java_opt:+${final_java_opt}} -XX:-OmitStackTraceInFastThrow -XX:OnOutOfMemoryError="kill -9 %p" ${coverage_opt:+${coverage_opt}} org.apache.doris.DorisFE "${HELPER_ARGS[@]}" ${OPT_VERSION:+${OPT_VERSION}} "${METADATA_FAILURE_RECOVERY_ARGS[@]}" "${RECOVERY_JOURNAL_ID_ARGS[@]}" "${CLUSTER_SNAPSHOT_ARGS[@]}" "$@" >>"${STDOUT_LOGGER}" </dev/null
+    ${LIMIT:+${LIMIT}} "${JAVA}" ${final_java_opt:+${final_java_opt}} -XX:-OmitStackTraceInFastThrow -XX:OnOutOfMemoryError="kill -9 %p" ${coverage_opt:+${coverage_opt}} org.apache.doris.DorisFE "${HELPER_ARGS[@]}" ${OPT_VERSION:+${OPT_VERSION}} "${METADATA_FAILURE_RECOVERY_ARGS[@]}" "${RECOVERY_JOURNAL_ID_ARGS[@]}" "${CLUSTER_SNAPSHOT_ARGS[@]}" "${DROP_BACKENDS_ARGS[@]}" "$@" >>"${STDOUT_LOGGER}" </dev/null
 else
-    ${LIMIT:+${LIMIT}} "${JAVA}" ${final_java_opt:+${final_java_opt}} -XX:-OmitStackTraceInFastThrow -XX:OnOutOfMemoryError="kill -9 %p" ${coverage_opt:+${coverage_opt}} org.apache.doris.DorisFE "${HELPER_ARGS[@]}" ${OPT_VERSION:+${OPT_VERSION}} "${METADATA_FAILURE_RECOVERY_ARGS[@]}" "${RECOVERY_JOURNAL_ID_ARGS[@]}" "${CLUSTER_SNAPSHOT_ARGS[@]}" "$@" >>"${STDOUT_LOGGER}" 2>&1 </dev/null
+    ${LIMIT:+${LIMIT}} "${JAVA}" ${final_java_opt:+${final_java_opt}} -XX:-OmitStackTraceInFastThrow -XX:OnOutOfMemoryError="kill -9 %p" ${coverage_opt:+${coverage_opt}} org.apache.doris.DorisFE "${HELPER_ARGS[@]}" ${OPT_VERSION:+${OPT_VERSION}} "${METADATA_FAILURE_RECOVERY_ARGS[@]}" "${RECOVERY_JOURNAL_ID_ARGS[@]}" "${CLUSTER_SNAPSHOT_ARGS[@]}" "${DROP_BACKENDS_ARGS[@]}" "$@" >>"${STDOUT_LOGGER}" 2>&1 </dev/null
 fi
 
 if [[ "${OPT_VERSION}" != "" ]]; then
diff --git a/docker/runtime/doris-compose/resource/init_fe.sh b/docker/runtime/doris-compose/resource/init_fe.sh
index c95c7e0416f001..4e3a3ffffe7b3c 100755
--- a/docker/runtime/doris-compose/resource/init_fe.sh
+++ b/docker/runtime/doris-compose/resource/init_fe.sh
@@ -188,7 +188,7 @@ start_cloud_fe() {
             exit $MV_RES
         fi
         health_log "Recovery script executed and renamed to ${RECOVERY_SCRIPT}.bak"
-        RECOVERY_ARGS="--metadata_failure_recovery --recovery_journal_id $JOURNAL_ID"
+        RECOVERY_ARGS="--metadata_failure_recovery --recovery_journal_id $JOURNAL_ID --drop_backends"
     fi
 
     if [ -f "$REGISTER_FILE" ] || [ -n "${CLUSTER_SNAPSHOT_FILE}" ] || [ -n "$RECOVERY_ARGS" ]; then
diff --git a/fe/fe-core/src/main/java/org/apache/doris/DorisFE.java b/fe/fe-core/src/main/java/org/apache/doris/DorisFE.java
index 8131bb3473271b..a3f5df40841032 100755
--- a/fe/fe-core/src/main/java/org/apache/doris/DorisFE.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/DorisFE.java
@@ -363,6 +363,8 @@ private static CommandLineOptions parseArgs(String[] args) {
                 .desc("Specify the recovery truncate journal id, and journals greater than this id will be removed")
                 .build());
         options.addOption("c", "cluster_snapshot", true, "Specify the cluster snapshot json file");
+        options.addOption(Option.builder().longOpt(FeConstants.DROP_BACKENDS_KEY)
+                .desc("When this FE becomes MASTER, drop all backends from cluster metadata (destructive)").build());
 
         CommandLine cmd = null;
         try {
@@ -407,6 +409,9 @@ private static CommandLineOptions parseArgs(String[] args) {
             }
             System.setProperty(FeConstants.RECOVERY_JOURNAL_ID_KEY, recoveryJournalId.trim());
         }
+        if (cmd.hasOption(FeConstants.DROP_BACKENDS_KEY)) {
+            System.setProperty(FeConstants.DROP_BACKENDS_KEY, "true");
+        }
         if (cmd.hasOption('b') || cmd.hasOption("bdb")) {
             if (cmd.hasOption('l') || cmd.hasOption("listdb")) {
                 // list bdb je databases
diff --git a/fe/fe-core/src/main/java/org/apache/doris/catalog/Env.java b/fe/fe-core/src/main/java/org/apache/doris/catalog/Env.java
index 44115f9b2fad65..eb66b5213addcf 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/catalog/Env.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/catalog/Env.java
@@ -55,6 +55,7 @@
 import org.apache.doris.clone.TabletChecker;
 import org.apache.doris.clone.TabletScheduler;
 import org.apache.doris.clone.TabletSchedulerStat;
+import org.apache.doris.cloud.system.CloudSystemInfoService;
 import org.apache.doris.common.AnalysisException;
 import org.apache.doris.common.Config;
 import org.apache.doris.common.ConfigBase;
@@ -272,6 +273,7 @@
 import org.apache.doris.statistics.StatisticsJobAppender;
 import org.apache.doris.statistics.StatisticsMetricCollector;
 import org.apache.doris.statistics.query.QueryStats;
+import org.apache.doris.system.Backend;
 import org.apache.doris.system.Frontend;
 import org.apache.doris.system.HeartbeatMgr;
 import org.apache.doris.system.SystemInfoService;
@@ -1826,6 +1828,25 @@ private void transferToMaster() {
             editLog.logMasterInfo(masterInfo);
             LOG.info("logMasterInfo:{}", masterInfo);
 
+            if (Boolean.getBoolean(FeConstants.DROP_BACKENDS_KEY)) {
+                LOG.info("drop_backends is set, dropping all backends...");
+                try {
+                    SystemInfoService systemInfoService = Env.getCurrentSystemInfo();
+                    List<Backend> bes = systemInfoService.getAllClusterBackendsNoException().values()
+                            .stream().collect(Collectors.toList());
+                    if (Config.isNotCloudMode()) {
+                        for (Backend be : bes) {
+                            systemInfoService.dropBackend(be.getHost(), be.getHeartbeatPort());
+                        }
+                    } else {
+                        ((CloudSystemInfoService) systemInfoService).updateCloudBackends(Collections.emptyList(), bes);
+                    }
+                } catch (Exception e) {
+                    LOG.warn("failed to drop backends", e);
+                }
+                LOG.info("finished dropping all backends");
+            }
+
             // for master, the 'isReady' is set behind.
             // but we are sure that all metadata is replayed if we get here.
             // so no need to check 'isReady' flag in this method
diff --git a/fe/fe-core/src/main/java/org/apache/doris/common/FeConstants.java b/fe/fe-core/src/main/java/org/apache/doris/common/FeConstants.java
index dc2e2aafbdc584..8d05fd2a698e3a 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/common/FeConstants.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/common/FeConstants.java
@@ -68,4 +68,5 @@ public class FeConstants {
 
     public static String METADATA_FAILURE_RECOVERY_KEY = "metadata_failure_recovery";
     public static String RECOVERY_JOURNAL_ID_KEY = "recovery_journal_id";
+    public static String DROP_BACKENDS_KEY = "drop_backends";
 }