[Fix-8980] [master] fixed repeat processing command (#9220)

This closes #8980
Co-authored-by: Nguoshupei <guoshupei@lixiang.com>

dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/enums/SlotCheckState.java

+/*
+ * Licensed to the Apache Software Foundation (ASF) 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 org.apache.dolphinscheduler.common.enums;
+
+public enum SlotCheckState {
+
+    PASS,INJECT,CHANGE
+}

dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/registry/ServerNodeManager.java

@@ -300,6 +300,7 @@ public class ServerNodeManager implements InitializingBean {
 
     private void updateMasterNodes() {
         MASTER_SLOT = 0;
+        MASTER_SIZE = 0;
         this.masterNodes.clear();
         String nodeLock = Constants.REGISTRY_DOLPHINSCHEDULER_LOCK_MASTERS;
         try {

dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/MasterSchedulerService.java

@@ -18,6 +18,7 @@
 package org.apache.dolphinscheduler.server.master.runner;
 
 import org.apache.dolphinscheduler.common.Constants;
+import org.apache.dolphinscheduler.common.enums.SlotCheckState;
 import org.apache.dolphinscheduler.common.thread.Stopper;
 import org.apache.dolphinscheduler.common.thread.ThreadUtils;
 import org.apache.dolphinscheduler.common.utils.NetUtils;
@@ -184,14 +185,15 @@ public class MasterSchedulerService extends Thread {
     }
 
     private List<ProcessInstance> command2ProcessInstance(List<Command> commands) {
-
         List<ProcessInstance> processInstances = Collections.synchronizedList(new ArrayList<>(commands.size()));
         CountDownLatch latch = new CountDownLatch(commands.size());
         for (final Command command : commands) {
             masterPrepareExecService.execute(() -> {
                 try {
                     // slot check again
-                    if (!slotCheck(command)) {
+                    SlotCheckState slotCheckState = slotCheck(command);
+                    if (slotCheckState.equals(SlotCheckState.CHANGE) || slotCheckState.equals(SlotCheckState.INJECT)) {
+                        logger.info("handle command {} skip, slot check state: {}", command.getId(), slotCheckState);
                         return;
                     }
                     ProcessInstance processInstance = processService.handleCommand(logger,
@@ -225,16 +227,18 @@ public class MasterSchedulerService extends Thread {
         int pageSize = masterConfig.getFetchCommandNum();
         List<Command> result = new ArrayList<>();
         while (Stopper.isRunning()) {
-            if (ServerNodeManager.getMasterSize() == 0) {
-                return result;
-            }
             // todo: Can we use the slot to scan database?
             List<Command> commandList = processService.findCommandPage(pageSize, pageNumber);
             if (commandList.size() == 0) {
                 return result;
             }
             for (Command command : commandList) {
-                if (slotCheck(command)) {
+                SlotCheckState slotCheckState = slotCheck(command);
+                if (slotCheckState.equals(SlotCheckState.CHANGE)) {
+                    // return and wait next scan, don't reset param, waste resources of cpu
+                    return new ArrayList<>();
+                }
+                if (slotCheckState.equals(SlotCheckState.PASS)) {
                     result.add(command);
                 }
             }
@@ -247,10 +251,18 @@ public class MasterSchedulerService extends Thread {
         return result;
     }
 
-    private boolean slotCheck(Command command) {
+    private SlotCheckState slotCheck(Command command) {
         int slot = ServerNodeManager.getSlot();
         int masterSize = ServerNodeManager.getMasterSize();
-        return masterSize != 0 && command.getId() % masterSize == slot;
+        SlotCheckState state;
+        if (masterSize <= 0) {
+            state = SlotCheckState.CHANGE;
+        } else if (command.getId() % masterSize == slot) {
+            state = SlotCheckState.PASS;
+        } else {
+            state = SlotCheckState.INJECT;
+        }
+        return state;
     }
 
     private String getLocalAddress() {