You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@dolphinscheduler.apache.org by ki...@apache.org on 2021/02/18 13:43:01 UTC
[incubator-dolphinscheduler] branch 1.3.6-prepare updated:
[1.3.6-prepare][cherry-pick]support distributed tracing #4270 (#4771)
This is an automated email from the ASF dual-hosted git repository.
kirs pushed a commit to branch 1.3.6-prepare
in repository https://gitbox.apache.org/repos/asf/incubator-dolphinscheduler.git
The following commit(s) were added to refs/heads/1.3.6-prepare by this push:
new c2d1145 [1.3.6-prepare][cherry-pick]support distributed tracing #4270 (#4771)
c2d1145 is described below
commit c2d1145366790b700a1da65bffaeb481d63561f2
Author: Kirs <ac...@163.com>
AuthorDate: Thu Feb 18 21:41:10 2021 +0800
[1.3.6-prepare][cherry-pick]support distributed tracing #4270 (#4771)
---
.../dolphinscheduler/common/utils/JSONUtils.java | 37 +++++++++
.../remote/codec/NettyDecoder.java | 33 ++++++++
.../remote/codec/NettyEncoder.java | 7 ++
.../dolphinscheduler/remote/command/Command.java | 14 ++++
.../{CommandHeader.java => CommandContext.java} | 50 +++++------
.../remote/command/CommandHeader.java | 26 ++++++
.../master/consumer/TaskPriorityQueueConsumer.java | 19 +++--
.../master/runner/MasterBaseTaskExecThread.java | 26 +++---
.../master/runner/MasterSchedulerService.java | 22 +++--
.../consumer/TaskPriorityQueueConsumerTest.java | 27 +++---
.../exceptions/TaskPriorityQueueException.java | 50 ++++-------
.../service/queue}/TaskPriority.java | 97 ++++++++++++++--------
.../service/queue/TaskPriorityQueueImpl.java | 53 ++----------
.../service/queue/TaskPriorityTest.java | 81 ++++++++++++++++++
.../service}/queue/TaskUpdateQueueTest.java | 23 ++---
pom.xml | 1 +
16 files changed, 366 insertions(+), 200 deletions(-)
diff --git a/dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/utils/JSONUtils.java b/dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/utils/JSONUtils.java
index f0aed91..4974cea 100644
--- a/dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/utils/JSONUtils.java
+++ b/dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/utils/JSONUtils.java
@@ -16,6 +16,8 @@
*/
package org.apache.dolphinscheduler.common.utils;
+import static java.nio.charset.StandardCharsets.UTF_8;
+
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
@@ -91,6 +93,21 @@ public class JSONUtils {
return null;
}
+ /**
+ * deserialize
+ *
+ * @param src byte array
+ * @param clazz class
+ * @param <T> deserialize type
+ * @return deserialize type
+ */
+ public static <T> T parseObject(byte[] src, Class<T> clazz) {
+ if (src == null) {
+ return null;
+ }
+ String json = new String(src, UTF_8);
+ return parseObject(json, clazz);
+ }
/**
* json to list
@@ -114,6 +131,26 @@ public class JSONUtils {
}
+ /**
+ * serialize to json byte
+ *
+ * @param obj object
+ * @param <T> object type
+ * @return byte array
+ */
+ public static <T> byte[] toJsonByteArray(T obj) {
+ if (obj == null) {
+ return null;
+ }
+ String json = "";
+ try {
+ json = toJsonString(obj);
+ } catch (Exception e) {
+ logger.error("json serialize exception.", e);
+ }
+
+ return json.getBytes(UTF_8);
+ }
/**
* check json object valid
diff --git a/dolphinscheduler-remote/src/main/java/org/apache/dolphinscheduler/remote/codec/NettyDecoder.java b/dolphinscheduler-remote/src/main/java/org/apache/dolphinscheduler/remote/codec/NettyDecoder.java
index caa4fbd..55911f0 100644
--- a/dolphinscheduler-remote/src/main/java/org/apache/dolphinscheduler/remote/codec/NettyDecoder.java
+++ b/dolphinscheduler-remote/src/main/java/org/apache/dolphinscheduler/remote/codec/NettyDecoder.java
@@ -22,6 +22,7 @@ import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.ReplayingDecoder;
import org.apache.dolphinscheduler.remote.command.Command;
+import org.apache.dolphinscheduler.remote.command.CommandContext;
import org.apache.dolphinscheduler.remote.command.CommandHeader;
import org.apache.dolphinscheduler.remote.command.CommandType;
@@ -51,16 +52,34 @@ public class NettyDecoder extends ReplayingDecoder<NettyDecoder.State> {
switch (state()){
case MAGIC:
checkMagic(in.readByte());
+ checkpoint(State.VERSION);
+ // fallthru
+ case VERSION:
+ checkVersion(in.readByte());
checkpoint(State.COMMAND);
+ // fallthru
case COMMAND:
commandHeader.setType(in.readByte());
checkpoint(State.OPAQUE);
+ // fallthru
case OPAQUE:
commandHeader.setOpaque(in.readLong());
+ checkpoint(State.CONTEXT_LENGTH);
+ // fallthru
+ case CONTEXT_LENGTH:
+ commandHeader.setContextLength(in.readInt());
+ checkpoint(State.CONTEXT);
+ // fallthru
+ case CONTEXT:
+ byte[] context = new byte[commandHeader.getContextLength()];
+ in.readBytes(context);
+ commandHeader.setContext(context);
checkpoint(State.BODY_LENGTH);
+ // fallthru
case BODY_LENGTH:
commandHeader.setBodyLength(in.readInt());
checkpoint(State.BODY);
+ // fallthru
case BODY:
byte[] body = new byte[commandHeader.getBodyLength()];
in.readBytes(body);
@@ -68,6 +87,7 @@ public class NettyDecoder extends ReplayingDecoder<NettyDecoder.State> {
Command packet = new Command();
packet.setType(commandType(commandHeader.getType()));
packet.setOpaque(commandHeader.getOpaque());
+ packet.setContext(CommandContext.valueOf(commandHeader.getContext()));
packet.setBody(body);
out.add(packet);
//
@@ -99,10 +119,23 @@ public class NettyDecoder extends ReplayingDecoder<NettyDecoder.State> {
}
}
+ /**
+ * check version
+ * @param version
+ */
+ private void checkVersion(byte version) {
+ if (version != Command.VERSION) {
+ throw new IllegalArgumentException("illegal protocol [version]" + version);
+ }
+ }
+
enum State{
MAGIC,
+ VERSION,
COMMAND,
OPAQUE,
+ CONTEXT_LENGTH,
+ CONTEXT,
BODY_LENGTH,
BODY;
}
diff --git a/dolphinscheduler-remote/src/main/java/org/apache/dolphinscheduler/remote/codec/NettyEncoder.java b/dolphinscheduler-remote/src/main/java/org/apache/dolphinscheduler/remote/codec/NettyEncoder.java
index 4e9836a..785ee5a 100644
--- a/dolphinscheduler-remote/src/main/java/org/apache/dolphinscheduler/remote/codec/NettyEncoder.java
+++ b/dolphinscheduler-remote/src/main/java/org/apache/dolphinscheduler/remote/codec/NettyEncoder.java
@@ -42,11 +42,18 @@ public class NettyEncoder extends MessageToByteEncoder<Command> {
throw new Exception("encode msg is null");
}
out.writeByte(Command.MAGIC);
+ out.writeByte(Command.VERSION);
out.writeByte(msg.getType().ordinal());
out.writeLong(msg.getOpaque());
+ writeContext(msg, out);
out.writeInt(msg.getBody().length);
out.writeBytes(msg.getBody());
}
+ private void writeContext(Command msg, ByteBuf out) {
+ byte[] headerBytes = msg.getContext().toBytes();
+ out.writeInt(headerBytes.length);
+ out.writeBytes(headerBytes);
+ }
}
diff --git a/dolphinscheduler-remote/src/main/java/org/apache/dolphinscheduler/remote/command/Command.java b/dolphinscheduler-remote/src/main/java/org/apache/dolphinscheduler/remote/command/Command.java
index ed46e1f..9baa321 100644
--- a/dolphinscheduler-remote/src/main/java/org/apache/dolphinscheduler/remote/command/Command.java
+++ b/dolphinscheduler-remote/src/main/java/org/apache/dolphinscheduler/remote/command/Command.java
@@ -28,6 +28,7 @@ public class Command implements Serializable {
private static final AtomicLong REQUEST_ID = new AtomicLong(1);
public static final byte MAGIC = (byte) 0xbabe;
+ public static final byte VERSION = 0;
public Command(){
this.opaque = REQUEST_ID.getAndIncrement();
@@ -48,6 +49,11 @@ public class Command implements Serializable {
private long opaque;
/**
+ * request context
+ */
+ private CommandContext context = new CommandContext();
+
+ /**
* data body
*/
private byte[] body;
@@ -76,6 +82,14 @@ public class Command implements Serializable {
this.body = body;
}
+ public CommandContext getContext() {
+ return context;
+ }
+
+ public void setContext(CommandContext context) {
+ this.context = context;
+ }
+
@Override
public int hashCode() {
final int prime = 31;
diff --git a/dolphinscheduler-remote/src/main/java/org/apache/dolphinscheduler/remote/command/CommandHeader.java b/dolphinscheduler-remote/src/main/java/org/apache/dolphinscheduler/remote/command/CommandContext.java
similarity index 55%
copy from dolphinscheduler-remote/src/main/java/org/apache/dolphinscheduler/remote/command/CommandHeader.java
copy to dolphinscheduler-remote/src/main/java/org/apache/dolphinscheduler/remote/command/CommandContext.java
index 78948a5..8f0b790 100644
--- a/dolphinscheduler-remote/src/main/java/org/apache/dolphinscheduler/remote/command/CommandHeader.java
+++ b/dolphinscheduler-remote/src/main/java/org/apache/dolphinscheduler/remote/command/CommandContext.java
@@ -14,51 +14,43 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
+
package org.apache.dolphinscheduler.remote.command;
+import org.apache.dolphinscheduler.common.utils.JSONUtils;
+
import java.io.Serializable;
+import java.util.LinkedHashMap;
+import java.util.Map;
/**
- * command header
+ * command context
*/
-public class CommandHeader implements Serializable {
-
- /**
- * type
- */
- private byte type;
-
- /**
- * request unique identification
- */
- private long opaque;
+public class CommandContext implements Serializable {
- /**
- * body length
- */
- private int bodyLength;
+ private Map<String, String> items = new LinkedHashMap<>();
- public int getBodyLength() {
- return bodyLength;
+ public Map<String, String> getItems() {
+ return items;
}
- public void setBodyLength(int bodyLength) {
- this.bodyLength = bodyLength;
+ public void setItems(Map<String, String> items) {
+ this.items = items;
}
- public byte getType() {
- return type;
+ public void put(String key, String value) {
+ items.put(key, value);
}
- public void setType(byte type) {
- this.type = type;
+ public String get(String key) {
+ return items.get(key);
}
- public long getOpaque() {
- return opaque;
+ public byte[] toBytes() {
+ return JSONUtils.toJsonByteArray(this);
}
- public void setOpaque(long opaque) {
- this.opaque = opaque;
+ public static CommandContext valueOf(byte[] src) {
+ return JSONUtils.parseObject(src, CommandContext.class);
}
-}
+}
\ No newline at end of file
diff --git a/dolphinscheduler-remote/src/main/java/org/apache/dolphinscheduler/remote/command/CommandHeader.java b/dolphinscheduler-remote/src/main/java/org/apache/dolphinscheduler/remote/command/CommandHeader.java
index 78948a5..9e83a42 100644
--- a/dolphinscheduler-remote/src/main/java/org/apache/dolphinscheduler/remote/command/CommandHeader.java
+++ b/dolphinscheduler-remote/src/main/java/org/apache/dolphinscheduler/remote/command/CommandHeader.java
@@ -34,6 +34,16 @@ public class CommandHeader implements Serializable {
private long opaque;
/**
+ * context length
+ */
+ private int contextLength;
+
+ /**
+ * context
+ */
+ private byte[] context;
+
+ /**
* body length
*/
private int bodyLength;
@@ -61,4 +71,20 @@ public class CommandHeader implements Serializable {
public void setOpaque(long opaque) {
this.opaque = opaque;
}
+
+ public int getContextLength() {
+ return contextLength;
+ }
+
+ public void setContextLength(int contextLength) {
+ this.contextLength = contextLength;
+ }
+
+ public byte[] getContext() {
+ return context;
+ }
+
+ public void setContext(byte[] context) {
+ this.context = context;
+ }
}
diff --git a/dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/master/consumer/TaskPriorityQueueConsumer.java b/dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/master/consumer/TaskPriorityQueueConsumer.java
index 20d3ec1..10f9710 100644
--- a/dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/master/consumer/TaskPriorityQueueConsumer.java
+++ b/dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/master/consumer/TaskPriorityQueueConsumer.java
@@ -40,6 +40,7 @@ import org.apache.dolphinscheduler.server.master.dispatch.context.ExecutionConte
import org.apache.dolphinscheduler.server.master.dispatch.enums.ExecutorType;
import org.apache.dolphinscheduler.server.master.dispatch.exceptions.ExecuteException;
import org.apache.dolphinscheduler.service.process.ProcessService;
+import org.apache.dolphinscheduler.service.queue.TaskPriority;
import org.apache.dolphinscheduler.service.queue.TaskPriorityQueue;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -67,7 +68,7 @@ public class TaskPriorityQueueConsumer extends Thread{
* taskUpdateQueue
*/
@Autowired
- private TaskPriorityQueue<String> taskPriorityQueue;
+ private TaskPriorityQueue<TaskPriority> taskPriorityQueue;
/**
* processService
@@ -96,7 +97,7 @@ public class TaskPriorityQueueConsumer extends Thread{
@Override
public void run() {
- List<String> failedDispatchTasks = new ArrayList<>();
+ List<TaskPriority> failedDispatchTasks = new ArrayList<>();
while (Stopper.isRunning()){
try {
int fetchTaskNum = masterConfig.getMasterDispatchTaskNumber();
@@ -107,15 +108,14 @@ public class TaskPriorityQueueConsumer extends Thread{
continue;
}
// if not task , blocking here
- String taskPriorityInfo = taskPriorityQueue.take();
- TaskPriority taskPriority = TaskPriority.of(taskPriorityInfo);
- boolean dispatchResult = dispatch(taskPriority.getTaskId());
+ TaskPriority taskPriority = taskPriorityQueue.take();
+ boolean dispatchResult = dispatch(taskPriority);
if(!dispatchResult){
- failedDispatchTasks.add(taskPriorityInfo);
+ failedDispatchTasks.add(taskPriority);
}
}
if (!failedDispatchTasks.isEmpty()) {
- for (String dispatchFailedTask : failedDispatchTasks) {
+ for (TaskPriority dispatchFailedTask : failedDispatchTasks) {
taskPriorityQueue.put(dispatchFailedTask);
}
// If there are tasks in a cycle that cannot find the worker group,
@@ -134,12 +134,13 @@ public class TaskPriorityQueueConsumer extends Thread{
/**
* dispatch task
*
- * @param taskInstanceId taskInstanceId
+ * @param taskPriority taskPriority
* @return result
*/
- protected boolean dispatch(int taskInstanceId) {
+ protected boolean dispatch(TaskPriority taskPriority) {
boolean result = false;
try {
+ int taskInstanceId = taskPriority.getTaskId();
TaskExecutionContext context = getTaskExecutionContext(taskInstanceId);
ExecutionContext executionContext = new ExecutionContext(context.toCommand(), ExecutorType.WORKER, context.getWorkerGroup());
diff --git a/dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/master/runner/MasterBaseTaskExecThread.java b/dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/master/runner/MasterBaseTaskExecThread.java
index 8297213..10de282 100644
--- a/dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/master/runner/MasterBaseTaskExecThread.java
+++ b/dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/master/runner/MasterBaseTaskExecThread.java
@@ -32,11 +32,11 @@ import org.apache.dolphinscheduler.server.log.TaskLogDiscriminator;
import org.apache.dolphinscheduler.server.master.config.MasterConfig;
import org.apache.dolphinscheduler.service.bean.SpringApplicationContext;
import org.apache.dolphinscheduler.service.process.ProcessService;
+import org.apache.dolphinscheduler.service.queue.TaskPriority;
import org.apache.dolphinscheduler.service.queue.TaskPriorityQueue;
import org.apache.dolphinscheduler.service.queue.TaskPriorityQueueImpl;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import static org.apache.dolphinscheduler.common.Constants.*;
import java.util.Date;
import java.util.concurrent.Callable;
@@ -216,14 +216,14 @@ public class MasterBaseTaskExecThread implements Callable<Boolean> {
logger.info("task ready to submit: {}", taskInstance);
/**
- * taskPriorityInfo
+ * taskPriority
*/
- String taskPriorityInfo = buildTaskPriorityInfo(processInstance.getProcessInstancePriority().getCode(),
+ TaskPriority taskPriority = buildTaskPriority(processInstance.getProcessInstancePriority().getCode(),
processInstance.getId(),
taskInstance.getProcessInstancePriority().getCode(),
taskInstance.getId(),
org.apache.dolphinscheduler.common.Constants.DEFAULT_WORKER_GROUP);
- taskUpdateQueue.put(taskPriorityInfo);
+ taskUpdateQueue.put(taskPriority);
logger.info(String.format("master submit success, task : %s", taskInstance.getName()) );
return true;
}catch (Exception e){
@@ -234,33 +234,27 @@ public class MasterBaseTaskExecThread implements Callable<Boolean> {
}
/**
- * buildTaskPriorityInfo
+ * buildTaskPriority
*
* @param processInstancePriority processInstancePriority
* @param processInstanceId processInstanceId
* @param taskInstancePriority taskInstancePriority
* @param taskInstanceId taskInstanceId
* @param workerGroup workerGroup
- * @return TaskPriorityInfo
+ * @return TaskPriority
*/
- private String buildTaskPriorityInfo(int processInstancePriority,
+ private TaskPriority buildTaskPriority(int processInstancePriority,
int processInstanceId,
int taskInstancePriority,
int taskInstanceId,
String workerGroup){
- return processInstancePriority +
- UNDERLINE +
- processInstanceId +
- UNDERLINE +
- taskInstancePriority +
- UNDERLINE +
- taskInstanceId +
- UNDERLINE +
- workerGroup;
+ return new TaskPriority(processInstancePriority, processInstanceId,
+ taskInstancePriority, taskInstanceId, workerGroup);
}
/**
* submit wait complete
+ *
* @return true
*/
protected Boolean submitWaitComplete(){
diff --git a/dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/master/runner/MasterSchedulerService.java b/dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/master/runner/MasterSchedulerService.java
index 33a7f4f..0985c1f 100644
--- a/dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/master/runner/MasterSchedulerService.java
+++ b/dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/master/runner/MasterSchedulerService.java
@@ -114,7 +114,6 @@ public class MasterSchedulerService extends Thread {
public void run() {
logger.info("master scheduler started");
while (Stopper.isRunning()){
- InterProcessMutex mutex = null;
try {
boolean runCheckFlag = OSUtils.checkResource(masterConfig.getMasterMaxCpuloadAvg(), masterConfig.getMasterReservedMemory());
if(!runCheckFlag) {
@@ -122,7 +121,17 @@ public class MasterSchedulerService extends Thread {
continue;
}
if (zkMasterClient.getZkClient().getState() == CuratorFrameworkState.STARTED) {
+ scheduleProcess();
+ }
+ } catch (Exception e) {
+ logger.error("master scheduler thread error", e);
+ }
+ }
+ }
+ private void scheduleProcess() throws Exception {
+ InterProcessMutex mutex = null;
+ try {
mutex = zkMasterClient.blockAcquireMutex();
int activeCount = masterExecService.getActiveCount();
@@ -138,7 +147,12 @@ public class MasterSchedulerService extends Thread {
this.masterConfig.getMasterExecThreads() - activeCount, command);
if (processInstance != null) {
logger.info("start master exec thread , split DAG ...");
- masterExecService.execute(new MasterExecThread(processInstance, processService, nettyRemotingClient));
+ masterExecService.execute(
+ new MasterExecThread(
+ processInstance
+ , processService
+ , nettyRemotingClient
+ ));
}
}catch (Exception e){
logger.error("scan command error ", e);
@@ -148,14 +162,10 @@ public class MasterSchedulerService extends Thread {
//indicate that no command ,sleep for 1s
Thread.sleep(Constants.SLEEP_TIME_MILLIS);
}
- }
- } catch (Exception e){
- logger.error("master scheduler thread error",e);
} finally{
zkMasterClient.releaseMutex(mutex);
}
}
- }
private String getLocalAddress(){
return OSUtils.getAddr(masterConfig.getListenPort());
diff --git a/dolphinscheduler-server/src/test/java/org/apache/dolphinscheduler/server/master/consumer/TaskPriorityQueueConsumerTest.java b/dolphinscheduler-server/src/test/java/org/apache/dolphinscheduler/server/master/consumer/TaskPriorityQueueConsumerTest.java
index 1ba823c..eb914f4 100644
--- a/dolphinscheduler-server/src/test/java/org/apache/dolphinscheduler/server/master/consumer/TaskPriorityQueueConsumerTest.java
+++ b/dolphinscheduler-server/src/test/java/org/apache/dolphinscheduler/server/master/consumer/TaskPriorityQueueConsumerTest.java
@@ -33,6 +33,7 @@ import org.apache.dolphinscheduler.server.registry.ZookeeperRegistryCenter;
import org.apache.dolphinscheduler.server.zk.SpringZKServer;
import org.apache.dolphinscheduler.service.bean.SpringApplicationContext;
import org.apache.dolphinscheduler.service.process.ProcessService;
+import org.apache.dolphinscheduler.service.queue.TaskPriority;
import org.apache.dolphinscheduler.service.queue.TaskPriorityQueue;
import org.apache.dolphinscheduler.service.zk.ZookeeperCachedOperator;
import org.apache.dolphinscheduler.service.zk.ZookeeperConfig;
@@ -61,7 +62,7 @@ public class TaskPriorityQueueConsumerTest {
@Autowired
- private TaskPriorityQueue taskPriorityQueue;
+ private TaskPriorityQueue<TaskPriority> taskPriorityQueue;
@Autowired
private TaskPriorityQueueConsumer taskPriorityQueueConsumer;
@@ -131,9 +132,8 @@ public class TaskPriorityQueueConsumerTest {
taskInstance.setProcessDefine(processDefinition);
Mockito.doReturn(taskInstance).when(processService).getTaskInstanceDetailByTaskId(1);
- Mockito.doReturn(taskInstance).when(processService).findTaskInstanceById(1);
-
- taskPriorityQueue.put("2_1_2_1_default");
+ TaskPriority taskPriority = new TaskPriority(2, 1, 2, 1, "default");
+ taskPriorityQueue.put(taskPriority);
TimeUnit.SECONDS.sleep(10);
@@ -169,7 +169,8 @@ public class TaskPriorityQueueConsumerTest {
processDefinition.setProjectId(1);
taskInstance.setProcessDefine(processDefinition);
Mockito.doReturn(taskInstance).when(processService).getTaskInstanceDetailByTaskId(1);
- taskPriorityQueue.put("2_1_2_1_default");
+ TaskPriority taskPriority = new TaskPriority(2, 1, 2, 1, "default");
+ taskPriorityQueue.put(taskPriority);
DataSource dataSource = new DataSource();
dataSource.setId(1);
@@ -232,7 +233,8 @@ public class TaskPriorityQueueConsumerTest {
processDefinition.setProjectId(1);
taskInstance.setProcessDefine(processDefinition);
Mockito.doReturn(taskInstance).when(processService).getTaskInstanceDetailByTaskId(1);
- taskPriorityQueue.put("2_1_2_1_default");
+ TaskPriority taskPriority = new TaskPriority(2, 1, 2, 1, "default");
+ taskPriorityQueue.put(taskPriority);
DataSource dataSource = new DataSource();
dataSource.setId(80);
@@ -299,7 +301,8 @@ public class TaskPriorityQueueConsumerTest {
processDefinition.setProjectId(1);
taskInstance.setProcessDefine(processDefinition);
Mockito.doReturn(taskInstance).when(processService).getTaskInstanceDetailByTaskId(1);
- taskPriorityQueue.put("2_1_2_1_default");
+ TaskPriority taskPriority = new TaskPriority(2, 1, 2, 1, "default");
+ taskPriorityQueue.put(taskPriority);
DataSource dataSource = new DataSource();
dataSource.setId(1);
@@ -391,7 +394,8 @@ public class TaskPriorityQueueConsumerTest {
Mockito.doReturn(taskInstance).when(processService).getTaskInstanceDetailByTaskId(1);
Mockito.doReturn(taskInstance).when(processService).findTaskInstanceById(1);
- taskPriorityQueue.put("2_1_2_1_NoWorkGroup");
+ TaskPriority taskPriority = new TaskPriority(2, 1, 2, 1, "NoWorkGroup");
+ taskPriorityQueue.put(taskPriority);
TimeUnit.SECONDS.sleep(10);
@@ -444,7 +448,9 @@ public class TaskPriorityQueueConsumerTest {
Mockito.doReturn(taskInstance).when(processService).getTaskInstanceDetailByTaskId(1);
Mockito.doReturn(taskInstance).when(processService).findTaskInstanceById(1);
- boolean res = taskPriorityQueueConsumer.dispatch(1);
+ TaskPriority taskPriority = new TaskPriority();
+ taskPriority.setTaskId(1);
+ boolean res = taskPriorityQueueConsumer.dispatch(taskPriority);
Assert.assertFalse(res);
}
@@ -611,7 +617,8 @@ public class TaskPriorityQueueConsumerTest {
Mockito.doReturn(taskInstance).when(processService).getTaskInstanceDetailByTaskId(1);
Mockito.doReturn(taskInstance).when(processService).findTaskInstanceById(1);
- taskPriorityQueue.put("2_1_2_1_NoWorkGroup");
+ TaskPriority taskPriority = new TaskPriority(2, 1, 2, 1, "NoWorkGroup");
+ taskPriorityQueue.put(taskPriority);
taskPriorityQueueConsumer.run();
diff --git a/dolphinscheduler-remote/src/main/java/org/apache/dolphinscheduler/remote/command/CommandHeader.java b/dolphinscheduler-service/src/main/java/org/apache/dolphinscheduler/service/exceptions/TaskPriorityQueueException.java
similarity index 54%
copy from dolphinscheduler-remote/src/main/java/org/apache/dolphinscheduler/remote/command/CommandHeader.java
copy to dolphinscheduler-service/src/main/java/org/apache/dolphinscheduler/service/exceptions/TaskPriorityQueueException.java
index 78948a5..30a7214 100644
--- a/dolphinscheduler-remote/src/main/java/org/apache/dolphinscheduler/remote/command/CommandHeader.java
+++ b/dolphinscheduler-service/src/main/java/org/apache/dolphinscheduler/service/exceptions/TaskPriorityQueueException.java
@@ -14,51 +14,31 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package org.apache.dolphinscheduler.remote.command;
-import java.io.Serializable;
+package org.apache.dolphinscheduler.service.exceptions;
/**
- * command header
+ * task priority queue exception
*/
-public class CommandHeader implements Serializable {
+public class TaskPriorityQueueException extends Exception {
/**
- * type
+ * Construct a new runtime exception with the detail message
+ *
+ * @param message message
*/
- private byte type;
-
- /**
- * request unique identification
- */
- private long opaque;
+ public TaskPriorityQueueException(String message) {
+ super(message);
+ }
/**
- * body length
+ * Construct a new runtime exception with the detail message and cause
+ *
+ * @param message message
+ * @param cause cause
*/
- private int bodyLength;
-
- public int getBodyLength() {
- return bodyLength;
- }
-
- public void setBodyLength(int bodyLength) {
- this.bodyLength = bodyLength;
+ public TaskPriorityQueueException(String message, Throwable cause) {
+ super(message, cause);
}
- public byte getType() {
- return type;
- }
-
- public void setType(byte type) {
- this.type = type;
- }
-
- public long getOpaque() {
- return opaque;
- }
-
- public void setOpaque(long opaque) {
- this.opaque = opaque;
- }
}
diff --git a/dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/entity/TaskPriority.java b/dolphinscheduler-service/src/main/java/org/apache/dolphinscheduler/service/queue/TaskPriority.java
similarity index 56%
rename from dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/entity/TaskPriority.java
rename to dolphinscheduler-service/src/main/java/org/apache/dolphinscheduler/service/queue/TaskPriority.java
index 991eeed..a872f6d 100644
--- a/dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/entity/TaskPriority.java
+++ b/dolphinscheduler-service/src/main/java/org/apache/dolphinscheduler/service/queue/TaskPriority.java
@@ -15,14 +15,15 @@
* limitations under the License.
*/
-package org.apache.dolphinscheduler.server.entity;
+package org.apache.dolphinscheduler.service.queue;
-import static org.apache.dolphinscheduler.common.Constants.*;
+import java.util.Map;
+import java.util.Objects;
/**
* task priority info
*/
-public class TaskPriority {
+public class TaskPriority implements Comparable<TaskPriority> {
/**
* processInstancePriority
@@ -50,9 +51,9 @@ public class TaskPriority {
private String groupName;
/**
- * ${processInstancePriority}_${processInstanceId}_${taskInstancePriority}_${taskId}_${groupName}
+ * context
*/
- private String taskPriorityInfo;
+ private Map<String, String> context;
public TaskPriority(){}
@@ -65,15 +66,6 @@ public class TaskPriority {
this.taskInstancePriority = taskInstancePriority;
this.taskId = taskId;
this.groupName = groupName;
- this.taskPriorityInfo = this.processInstancePriority +
- UNDERLINE +
- this.processInstanceId +
- UNDERLINE +
- this.taskInstancePriority +
- UNDERLINE +
- this.taskId +
- UNDERLINE +
- this.groupName;
}
public int getProcessInstancePriority() {
@@ -104,6 +96,10 @@ public class TaskPriority {
return taskId;
}
+ public Map<String, String> getContext() {
+ return context;
+ }
+
public void setTaskId(int taskId) {
this.taskId = taskId;
}
@@ -116,32 +112,61 @@ public class TaskPriority {
this.groupName = groupName;
}
- public String getTaskPriorityInfo() {
- return taskPriorityInfo;
+ public void setContext(Map<String, String> context) {
+ this.context = context;
}
- public void setTaskPriorityInfo(String taskPriorityInfo) {
- this.taskPriorityInfo = taskPriorityInfo;
- }
+ @Override
+ public int compareTo(TaskPriority other) {
+ if (this.getProcessInstancePriority() > other.getProcessInstancePriority()) {
+ return 1;
+ }
+ if (this.getProcessInstancePriority() < other.getProcessInstancePriority()) {
+ return -1;
+ }
- /**
- * taskPriorityInfo convert taskPriority
- *
- * @param taskPriorityInfo taskPriorityInfo
- * @return TaskPriority
- */
- public static TaskPriority of(String taskPriorityInfo){
- String[] parts = taskPriorityInfo.split(UNDERLINE);
+ if (this.getProcessInstanceId() > other.getProcessInstanceId()) {
+ return 1;
+ }
+ if (this.getProcessInstanceId() < other.getProcessInstanceId()) {
+ return -1;
+ }
+
+ if (this.getTaskInstancePriority() > other.getTaskInstancePriority()) {
+ return 1;
+ }
+ if (this.getTaskInstancePriority() < other.getTaskInstancePriority()) {
+ return -1;
+ }
+
+ if (this.getTaskId() > other.getTaskId()) {
+ return 1;
+ }
+ if (this.getTaskId() < other.getTaskId()) {
+ return -1;
+ }
- if (parts.length != 5) {
- throw new IllegalArgumentException(String.format("TaskPriority : %s illegal.", taskPriorityInfo));
+ return this.getGroupName().compareTo(other.getGroupName());
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
}
- TaskPriority taskPriority = new TaskPriority(
- Integer.parseInt(parts[0]),
- Integer.parseInt(parts[1]),
- Integer.parseInt(parts[2]),
- Integer.parseInt(parts[3]),
- parts[4]);
- return taskPriority;
+ TaskPriority that = (TaskPriority) o;
+ return processInstancePriority == that.processInstancePriority
+ && processInstanceId == that.processInstanceId
+ && taskInstancePriority == that.taskInstancePriority
+ && taskId == that.taskId
+ && Objects.equals(groupName, that.groupName);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(processInstancePriority, processInstanceId, taskInstancePriority, taskId, groupName);
}
}
diff --git a/dolphinscheduler-service/src/main/java/org/apache/dolphinscheduler/service/queue/TaskPriorityQueueImpl.java b/dolphinscheduler-service/src/main/java/org/apache/dolphinscheduler/service/queue/TaskPriorityQueueImpl.java
index 2677ddb..bf31faf 100644
--- a/dolphinscheduler-service/src/main/java/org/apache/dolphinscheduler/service/queue/TaskPriorityQueueImpl.java
+++ b/dolphinscheduler-service/src/main/java/org/apache/dolphinscheduler/service/queue/TaskPriorityQueueImpl.java
@@ -16,21 +16,16 @@
*/
package org.apache.dolphinscheduler.service.queue;
-
-import org.springframework.stereotype.Service;
-
-import java.util.Comparator;
import java.util.concurrent.PriorityBlockingQueue;
-import static org.apache.dolphinscheduler.common.Constants.TASK_INFO_LENGTH;
-import static org.apache.dolphinscheduler.common.Constants.UNDERLINE;
+import org.springframework.stereotype.Service;
/**
* A singleton of a task queue implemented with zookeeper
* tasks queue implementation
*/
@Service
-public class TaskPriorityQueueImpl implements TaskPriorityQueue<String> {
+public class TaskPriorityQueueImpl implements TaskPriorityQueue<TaskPriority> {
/**
* queue size
*/
@@ -39,67 +34,35 @@ public class TaskPriorityQueueImpl implements TaskPriorityQueue<String> {
/**
* queue
*/
- private PriorityBlockingQueue<String> queue = new PriorityBlockingQueue<>(QUEUE_MAX_SIZE, new TaskInfoComparator());
+ private PriorityBlockingQueue<TaskPriority> queue = new PriorityBlockingQueue<>(QUEUE_MAX_SIZE);
/**
* put task takePriorityInfo
*
* @param taskPriorityInfo takePriorityInfo
- * @throws Exception
*/
@Override
- public void put(String taskPriorityInfo) throws Exception {
+ public void put(TaskPriority taskPriorityInfo) {
queue.put(taskPriorityInfo);
}
/**
* take taskInfo
+ *
* @return taskInfo
- * @throws Exception
*/
@Override
- public String take() throws Exception {
+ public TaskPriority take() throws InterruptedException {
return queue.take();
}
/**
* queue size
+ *
* @return size
- * @throws Exception
*/
@Override
- public int size() throws Exception {
+ public int size() {
return queue.size();
}
-
- /**
- * TaskInfoComparator
- */
- private class TaskInfoComparator implements Comparator<String>{
-
- /**
- * compare o1 o2
- * @param o1 o1
- * @param o2 o2
- * @return compare result
- */
- @Override
- public int compare(String o1, String o2) {
- String s1 = o1;
- String s2 = o2;
- String[] s1Array = s1.split(UNDERLINE);
- if(s1Array.length > TASK_INFO_LENGTH){
- // warning: if this length > 5, need to be changed
- s1 = s1.substring(0, s1.lastIndexOf(UNDERLINE) );
- }
-
- String[] s2Array = s2.split(UNDERLINE);
- if(s2Array.length > TASK_INFO_LENGTH){
- // warning: if this length > 5, need to be changed
- s2 = s2.substring(0, s2.lastIndexOf(UNDERLINE) );
- }
-
- return s1.compareTo(s2);
- }
- }
}
diff --git a/dolphinscheduler-service/src/test/java/org/apache/dolphinscheduler/service/queue/TaskPriorityTest.java b/dolphinscheduler-service/src/test/java/org/apache/dolphinscheduler/service/queue/TaskPriorityTest.java
new file mode 100644
index 0000000..3fad4c6
--- /dev/null
+++ b/dolphinscheduler-service/src/test/java/org/apache/dolphinscheduler/service/queue/TaskPriorityTest.java
@@ -0,0 +1,81 @@
+/*
+ * 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.service.queue;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+
+import org.junit.Assert;
+import org.junit.Test;
+
+public class TaskPriorityTest {
+
+ @Test
+ public void testSort() {
+ TaskPriority priorityOne = new TaskPriority(1, 0, 0, 0, "default");
+ TaskPriority priorityTwo = new TaskPriority(2, 0, 0, 0, "default");
+ TaskPriority priorityThree = new TaskPriority(3, 0, 0, 0, "default");
+ List<TaskPriority> taskPrioritys = Arrays.asList(priorityOne, priorityThree, priorityTwo);
+ Collections.sort(taskPrioritys);
+ Assert.assertEquals(
+ Arrays.asList(priorityOne, priorityTwo, priorityThree),
+ taskPrioritys
+ );
+
+ priorityOne = new TaskPriority(0, 1, 0, 0, "default");
+ priorityTwo = new TaskPriority(0, 2, 0, 0, "default");
+ priorityThree = new TaskPriority(0, 3, 0, 0, "default");
+ taskPrioritys = Arrays.asList(priorityOne, priorityThree, priorityTwo);
+ Collections.sort(taskPrioritys);
+ Assert.assertEquals(
+ Arrays.asList(priorityOne, priorityTwo, priorityThree),
+ taskPrioritys
+ );
+
+ priorityOne = new TaskPriority(0, 0, 1, 0, "default");
+ priorityTwo = new TaskPriority(0, 0, 2, 0, "default");
+ priorityThree = new TaskPriority(0, 0, 3, 0, "default");
+ taskPrioritys = Arrays.asList(priorityOne, priorityThree, priorityTwo);
+ Collections.sort(taskPrioritys);
+ Assert.assertEquals(
+ Arrays.asList(priorityOne, priorityTwo, priorityThree),
+ taskPrioritys
+ );
+
+ priorityOne = new TaskPriority(0, 0, 0, 1, "default");
+ priorityTwo = new TaskPriority(0, 0, 0, 2, "default");
+ priorityThree = new TaskPriority(0, 0, 0, 3, "default");
+ taskPrioritys = Arrays.asList(priorityOne, priorityThree, priorityTwo);
+ Collections.sort(taskPrioritys);
+ Assert.assertEquals(
+ Arrays.asList(priorityOne, priorityTwo, priorityThree),
+ taskPrioritys
+ );
+
+ priorityOne = new TaskPriority(0, 0, 0, 0, "default_1");
+ priorityTwo = new TaskPriority(0, 0, 0, 0, "default_2");
+ priorityThree = new TaskPriority(0, 0, 0, 0, "default_3");
+ taskPrioritys = Arrays.asList(priorityOne, priorityThree, priorityTwo);
+ Collections.sort(taskPrioritys);
+ Assert.assertEquals(
+ Arrays.asList(priorityOne, priorityTwo, priorityThree),
+ taskPrioritys
+ );
+ }
+}
\ No newline at end of file
diff --git a/dolphinscheduler-service/src/test/java/queue/TaskUpdateQueueTest.java b/dolphinscheduler-service/src/test/java/org/apache/dolphinscheduler/service/queue/TaskUpdateQueueTest.java
similarity index 66%
rename from dolphinscheduler-service/src/test/java/queue/TaskUpdateQueueTest.java
rename to dolphinscheduler-service/src/test/java/org/apache/dolphinscheduler/service/queue/TaskUpdateQueueTest.java
index ca6c083..542952a 100644
--- a/dolphinscheduler-service/src/test/java/queue/TaskUpdateQueueTest.java
+++ b/dolphinscheduler-service/src/test/java/org/apache/dolphinscheduler/service/queue/TaskUpdateQueueTest.java
@@ -15,10 +15,8 @@
* limitations under the License.
*/
-package queue;
+package org.apache.dolphinscheduler.service.queue;
-import org.apache.dolphinscheduler.service.queue.TaskPriorityQueue;
-import org.apache.dolphinscheduler.service.queue.TaskPriorityQueueImpl;
import org.junit.Test;
import static org.junit.Assert.*;
@@ -31,19 +29,16 @@ public class TaskUpdateQueueTest {
@Test
public void testQueue() throws Exception{
- // ${processInstancePriority}_${processInstanceId}_${taskInstancePriority}_${taskId}_${groupName}
-
/**
* 1_1_2_1_default
* 1_1_2_2_default
* 1_1_0_3_default
* 1_1_0_4_default
*/
-
- String taskInfo1 = "1_1_2_1_default";
- String taskInfo2 = "1_1_2_2_default";
- String taskInfo3 = "1_1_0_3_default";
- String taskInfo4 = "1_1_0_4_default";
+ TaskPriority taskInfo1 = new TaskPriority(1, 1, 2, 1, "default");
+ TaskPriority taskInfo2 = new TaskPriority(1, 1, 2, 2, "default");
+ TaskPriority taskInfo3 = new TaskPriority(1, 1, 0, 3, "default");
+ TaskPriority taskInfo4 = new TaskPriority(1, 1, 0, 4, "default");
TaskPriorityQueue queue = new TaskPriorityQueueImpl();
queue.put(taskInfo1);
@@ -51,9 +46,9 @@ public class TaskUpdateQueueTest {
queue.put(taskInfo3);
queue.put(taskInfo4);
- assertEquals("1_1_0_3_default", queue.take());
- assertEquals("1_1_0_4_default", queue.take());
- assertEquals("1_1_2_1_default",queue.take());
- assertEquals("1_1_2_2_default",queue.take());
+ assertEquals(taskInfo3, queue.take());
+ assertEquals(taskInfo4, queue.take());
+ assertEquals(taskInfo1, queue.take());
+ assertEquals(taskInfo2, queue.take());
}
}
diff --git a/pom.xml b/pom.xml
index 80def76..443ae36 100644
--- a/pom.xml
+++ b/pom.xml
@@ -830,6 +830,7 @@
<include>**/service/zk/DefaultEnsembleProviderTest.java</include>
<include>**/service/zk/ZKServerTest.java</include>
<include>**/service/queue/TaskUpdateQueueTest.java</include>
+ <include>**/service/queue/TaskPriorityTest.java</include>
<include>**/dao/mapper/DataSourceUserMapperTest.java</include>
<include>**/dao/mapper/ErrorCommandMapperTest.java</include>
<include>**/dao/mapper/ProcessDefinitionMapperTest.java</include>