feat: support TaskTrackerBehavior(PADDLING)

powerjob-common/src/main/java/tech/powerjob/common/enums/TaskTrackerBehavior.java

@@ -0,0 +1,42 @@
+package tech.powerjob.common.enums;
+
+import lombok.AllArgsConstructor;
+import lombok.Getter;
+
+/**
+ * TaskTracker 行为枚举
+ *
+ * @author tjq
+ * @since 2024/2/24
+ */
+@Getter
+@AllArgsConstructor
+public enum TaskTrackerBehavior {
+
+    /**
+     * 普通：不特殊处理，参与集群计算，会导致 TaskTracker 负载比常规节点高。适用于节点数不那么多，任务不那么繁重的场景
+     */
+    NORMAL(1),
+    /**
+     * 划水：只负责管理节点，不参与计算，稳定性最优。适用于节点数量非常多的大规模计算场景，少一个计算节点来换取稳定性提升
+     */
+    PADDLING(11)
+    ;
+
+
+    private final Integer v;
+
+    public static TaskTrackerBehavior of(Integer type) {
+
+        if (type == null) {
+            return NORMAL;
+        }
+
+        for (TaskTrackerBehavior t : values()) {
+            if (t.v.equals(type)) {
+                return t;
+            }
+        }
+        return NORMAL;
+    }
+}

powerjob-common/src/main/java/tech/powerjob/common/model/JobAdvancedRuntimeConfig.java

@@ -0,0 +1,25 @@
+package tech.powerjob.common.model;
+
+import lombok.Getter;
+import lombok.Setter;
+import lombok.ToString;
+import lombok.experimental.Accessors;
+
+/**
+ * 任务运行时高级配置
+ *
+ * @author tjq
+ * @since 2024/2/24
+ */
+@Getter
+@Setter
+@ToString
+@Accessors(chain = true)
+public class JobAdvancedRuntimeConfig {
+
+    /**
+     * MR 任务专享参数，TaskTracker 行为 {@link tech.powerjob.common.enums.TaskTrackerBehavior}
+     */
+    private Integer taskTrackerBehavior;
+
+}

powerjob-common/src/main/java/tech/powerjob/common/request/ServerScheduleJobReq.java

@@ -97,4 +97,9 @@ public class ServerScheduleJobReq implements PowerSerializable {
      * 日志配置
      */
     private String logConfig;
+
+    /**
+     * 高级运行时配置
+     */
+    private String advancedRuntimeConfig;
 }

powerjob-common/src/main/java/tech/powerjob/common/request/http/SaveJobInfoRequest.java

@@ -5,6 +5,7 @@ import tech.powerjob.common.enums.ExecuteType;
 import tech.powerjob.common.enums.ProcessorType;
 import tech.powerjob.common.enums.TimeExpressionType;
 import tech.powerjob.common.model.AlarmConfig;
+import tech.powerjob.common.model.JobAdvancedRuntimeConfig;
 import tech.powerjob.common.model.LogConfig;
 import tech.powerjob.common.model.LifeCycle;
 import tech.powerjob.common.utils.CommonUtils;
@@ -155,6 +156,10 @@ public class SaveJobInfoRequest {
      */
     private LogConfig logConfig;
 
+    /**
+     * 高级运行时配置
+     */
+    private JobAdvancedRuntimeConfig advancedRuntimeConfig;
 
     /**
      * Check non-null properties.

powerjob-common/src/main/java/tech/powerjob/common/response/JobInfoDTO.java

@@ -2,6 +2,7 @@ package tech.powerjob.common.response;
 
 import lombok.Data;
 import tech.powerjob.common.model.AlarmConfig;
+import tech.powerjob.common.model.JobAdvancedRuntimeConfig;
 import tech.powerjob.common.model.LogConfig;
 
 import java.util.Date;
@@ -144,4 +145,5 @@ public class JobInfoDTO {
      */
     private LogConfig logConfig;
 
+    private JobAdvancedRuntimeConfig advancedRuntimeConfig;
 }

powerjob-server/powerjob-server-core/src/main/java/tech/powerjob/server/core/service/impl/job/JobConverter.java

@@ -9,6 +9,7 @@ import tech.powerjob.common.enums.ExecuteType;
 import tech.powerjob.common.enums.ProcessorType;
 import tech.powerjob.common.enums.TimeExpressionType;
 import tech.powerjob.common.model.AlarmConfig;
+import tech.powerjob.common.model.JobAdvancedRuntimeConfig;
 import tech.powerjob.common.model.LifeCycle;
 import tech.powerjob.common.model.LogConfig;
 import tech.powerjob.common.request.http.SaveJobInfoRequest;
@@ -40,6 +41,7 @@ public class JobConverter {
         saveJobInfoRequest.setLifeCycle(LifeCycle.parse(jobInfoDO.getLifecycle()));
         saveJobInfoRequest.setAlarmConfig(JsonUtils.parseObjectIgnoreException(jobInfoDO.getAlarmConfig(), AlarmConfig.class));
         saveJobInfoRequest.setLogConfig(JsonUtils.parseObjectIgnoreException(jobInfoDO.getLogConfig(), LogConfig.class));
+        saveJobInfoRequest.setAdvancedRuntimeConfig(JsonUtils.parseObjectIgnoreException(jobInfoDO.getAdvancedRuntimeConfig(), JobAdvancedRuntimeConfig.class));
         return saveJobInfoRequest;
     }
 
@@ -49,6 +51,14 @@ public class JobConverter {
         if (jobInfoDO.getAlarmConfig() != null) {
             jobInfoDTO.setAlarmConfig(JSON.parseObject(jobInfoDO.getAlarmConfig(), AlarmConfig.class));
         }
+
+        if (StringUtils.isNotEmpty(jobInfoDO.getLogConfig())) {
+            jobInfoDTO.setLogConfig(JSON.parseObject(jobInfoDO.getLogConfig(), LogConfig.class));
+        }
+
+        if (StringUtils.isNotEmpty(jobInfoDO.getAdvancedRuntimeConfig())) {
+            jobInfoDTO.setAdvancedRuntimeConfig(JSON.parseObject(jobInfoDO.getAdvancedRuntimeConfig(), JobAdvancedRuntimeConfig.class));
+        }
         return jobInfoDTO;
     }
 

powerjob-server/powerjob-server-core/src/main/java/tech/powerjob/server/core/service/impl/job/JobServiceImpl.java

@@ -116,6 +116,10 @@ public class JobServiceImpl implements JobService {
         if (request.getLogConfig() != null) {
             jobInfoDO.setLogConfig(JSONObject.toJSONString(request.getLogConfig()));
         }
+        // 日志配置
+        if (request.getAdvancedRuntimeConfig() != null) {
+            jobInfoDO.setAdvancedRuntimeConfig(JSONObject.toJSONString(request.getAdvancedRuntimeConfig()));
+        }
         JobInfoDO res = jobInfoRepository.saveAndFlush(jobInfoDO);
         return res.getId();
     }

powerjob-server/powerjob-server-persistence/src/main/java/tech/powerjob/server/persistence/remote/model/JobInfoDO.java

@@ -163,4 +163,10 @@ public class JobInfoDO {
      * 日志配置，包括日志级别、日志方式等配置信息
      */
     private String logConfig;
+
+    /**
+     * 高级运行时配置
+     * 不需要用于索引的高级运行参数，后续统一存储到这里，便于版本升级（尽可能保证数据库表结构稳定）
+     */
+    private String advancedRuntimeConfig;
 }

powerjob-server/powerjob-server-starter/src/main/java/tech/powerjob/server/web/response/JobInfoVO.java

@@ -7,6 +7,7 @@ import tech.powerjob.common.enums.ExecuteType;
 import tech.powerjob.common.enums.ProcessorType;
 import tech.powerjob.common.enums.TimeExpressionType;
 import tech.powerjob.common.model.AlarmConfig;
+import tech.powerjob.common.model.JobAdvancedRuntimeConfig;
 import tech.powerjob.common.model.LogConfig;
 import tech.powerjob.common.model.LifeCycle;
 import tech.powerjob.common.utils.CommonUtils;
@@ -161,6 +162,8 @@ public class JobInfoVO {
      */
     private LogConfig logConfig;
 
+    private JobAdvancedRuntimeConfig advancedRuntimeConfig;
+
     public static JobInfoVO from(JobInfoDO jobInfoDO) {
         JobInfoVO jobInfoVO = new JobInfoVO();
         BeanUtils.copyProperties(jobInfoDO, jobInfoVO);
@@ -199,6 +202,12 @@ public class JobInfoVO {
             jobInfoVO.setLogConfig(new LogConfig());
         }
 
+        if (StringUtils.isEmpty(jobInfoDO.getAdvancedRuntimeConfig())) {
+            jobInfoVO.setAdvancedRuntimeConfig(new JobAdvancedRuntimeConfig());
+        } else {
+            jobInfoVO.setAdvancedRuntimeConfig(JSONObject.parseObject(jobInfoDO.getAdvancedRuntimeConfig(), JobAdvancedRuntimeConfig.class));
+        }
+
         return jobInfoVO;
     }
 }

powerjob-worker/src/main/java/tech/powerjob/worker/core/tracker/task/TaskTracker.java

@@ -5,14 +5,17 @@ import lombok.extern.slf4j.Slf4j;
 import tech.powerjob.common.enums.ExecuteType;
 import tech.powerjob.common.enums.InstanceStatus;
 import tech.powerjob.common.model.InstanceDetail;
+import tech.powerjob.common.model.JobAdvancedRuntimeConfig;
 import tech.powerjob.common.request.ServerScheduleJobReq;
 import tech.powerjob.common.request.TaskTrackerReportInstanceStatusReq;
+import tech.powerjob.common.serialize.JsonUtils;
 import tech.powerjob.worker.common.WorkerRuntime;
 import tech.powerjob.worker.common.utils.TransportUtils;
 import tech.powerjob.worker.pojo.model.InstanceInfo;
 
 import java.util.Collections;
 import java.util.Map;
+import java.util.Optional;
 import java.util.concurrent.atomic.AtomicBoolean;
 
 /**
@@ -35,6 +38,8 @@ public abstract class TaskTracker {
      */
     protected final InstanceInfo instanceInfo;
     protected final ExecuteType executeType;
+
+    protected final JobAdvancedRuntimeConfig advancedRuntimeConfig;
     /**
      * 追加的工作流上下文数据
      *
@@ -76,9 +81,11 @@ public abstract class TaskTracker {
         instanceInfo.setTaskRetryNum(req.getTaskRetryNum());
         instanceInfo.setLogConfig(req.getLogConfig());
         instanceInfo.setInstanceTimeoutMS(req.getInstanceTimeoutMS());
+        instanceInfo.setAdvancedRuntimeConfig(req.getAdvancedRuntimeConfig());
 
         // 常用变量初始化
         executeType = ExecuteType.valueOf(req.getExecuteType());
+        advancedRuntimeConfig = Optional.ofNullable(req.getAdvancedRuntimeConfig()).map(x -> JsonUtils.parseObjectIgnoreException(x, JobAdvancedRuntimeConfig.class)).orElse(new JobAdvancedRuntimeConfig());
 
         // 特殊处理超时时间
         if (instanceInfo.getInstanceTimeoutMS() <= 0) {

powerjob-worker/src/main/java/tech/powerjob/worker/core/tracker/task/heavy/HeavyTaskTracker.java

@@ -11,6 +11,7 @@ import lombok.extern.slf4j.Slf4j;
 import org.apache.commons.lang3.StringUtils;
 import tech.powerjob.common.RemoteConstant;
 import tech.powerjob.common.enums.ExecuteType;
+import tech.powerjob.common.enums.TaskTrackerBehavior;
 import tech.powerjob.common.enums.TimeExpressionType;
 import tech.powerjob.common.request.ServerScheduleJobReq;
 import tech.powerjob.common.request.WorkerQueryExecutorClusterReq;
@@ -489,7 +490,13 @@ public abstract class HeavyTaskTracker extends TaskTracker {
                     // 获取 ProcessorTracker 地址，如果 Task 中自带了 Address，则使用该 Address
                     String ptAddress = task.getAddress();
                     if (StringUtils.isEmpty(ptAddress) || RemoteConstant.EMPTY_ADDRESS.equals(ptAddress)) {
-                        ptAddress = availablePtIps.get(index.getAndIncrement() % availablePtIps.size());
+                        if (taskNeedByPassTaskTracker()) {
+                            do {
+                                ptAddress = availablePtIps.get(index.getAndIncrement() % availablePtIps.size());
+                            } while (workerRuntime.getWorkerAddress().equals(ptAddress));
+                        } else {
+                            ptAddress = availablePtIps.get(index.getAndIncrement() % availablePtIps.size());
+                        }
                     }
                     dispatchTask(task, ptAddress);
                 });
@@ -502,6 +509,13 @@ public abstract class HeavyTaskTracker extends TaskTracker {
 
             log.debug("[TaskTracker-{}] dispatched {} tasks,using time {}.", instanceId, currentDispatchNum, stopwatch.stop());
         }
+
+        private boolean taskNeedByPassTaskTracker() {
+            if (ExecuteType.MAP.equals(executeType) || ExecuteType.MAP_REDUCE.equals(executeType)) {
+                return TaskTrackerBehavior.PADDLING.getV().equals(advancedRuntimeConfig.getTaskTrackerBehavior());
+            }
+            return false;
+        }
     }
 
     /**

powerjob-worker/src/main/java/tech/powerjob/worker/pojo/model/InstanceInfo.java

@@ -53,4 +53,6 @@ public class InstanceInfo implements Serializable {
     private int taskRetryNum;
 
     private String logConfig;
+
+    private String advancedRuntimeConfig;
 }

powerjob-worker/src/main/java/tech/powerjob/worker/pojo/request/TaskTrackerStartTaskReq.java

@@ -34,6 +34,7 @@ public class TaskTrackerStartTaskReq implements PowerSerializable {
 
     private String logConfig;
 
+    private String advancedRuntimeConfig;
 
     /**
      * 创建 TaskTrackerStartTaskReq，该构造方法必须在 TaskTracker 节点调用
@@ -51,5 +52,6 @@ public class TaskTrackerStartTaskReq implements PowerSerializable {
         this.subInstanceId = task.getSubInstanceId();
 
         this.logConfig = instanceInfo.getLogConfig();
+        this.advancedRuntimeConfig = instanceInfo.getAdvancedRuntimeConfig();
     }
 }