diff --git a/dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/enums/SlotCheckState.java b/dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/enums/SlotCheckState.java new file mode 100644 index 0000000000..731911bc3b --- /dev/null +++ b/dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/enums/SlotCheckState.java @@ -0,0 +1,23 @@ +/* + * 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 +} diff --git a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/registry/ServerNodeManager.java b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/registry/ServerNodeManager.java index 11f013ed31..7f11576d13 100644 --- a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/registry/ServerNodeManager.java +++ b/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 { diff --git a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/MasterSchedulerService.java b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/MasterSchedulerService.java index e0fa6b87d9..56b6aacaed 100644 --- a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/MasterSchedulerService.java +++ b/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 command2ProcessInstance(List commands) { - List 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 result = new ArrayList<>(); while (Stopper.isRunning()) { - if (ServerNodeManager.getMasterSize() == 0) { - return result; - } // todo: Can we use the slot to scan database? List 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() {