Skip to content

Commit adb3bec

Browse files
baisui1981baisui1981
committed
resolve dolphinescheduler trigger faild bug
1 parent df50ec4 commit adb3bec

9 files changed

Lines changed: 26 additions & 62 deletions

File tree

‎tis-datax/executor/dolphinscheduler-task-tis-datasync/scp-remote.sh‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,2 +1,2 @@
11
#rsync --include="dolphinscheduler-task-tis-datasync-*.jar" -vr ./target/* root@192.168.28.201:/opt/misc/apache-dolphinscheduler-3.2.2-bin/libs/
2-
scp ./target/dolphinscheduler-task-tis-datasync-4.0.1-shaded.jar root@192.168.28.201:/opt/misc/apache-dolphinscheduler-3.2.2-bin/libs/dolphinscheduler-task-datasync-3.2.2.jar
2+
scp ./target/dolphinscheduler-task-tis-datasync-4.3.0-SNAPSHOT-shaded.jar root@192.168.28.201:/opt/misc/apache-dolphinscheduler-3.2.2-bin/libs/dolphinscheduler-task-datasync-3.2.2.jar

‎tis-datax/executor/dolphinscheduler-task-tis-datasync/scp-tis-datax-executor.sh‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,2 @@
1+
#rsync --include="dolphinscheduler-task-tis-datasync-*.jar" -vr ./target/* root@192.168.28.201:/opt/misc/apache-dolphinscheduler-3.2.2-bin/libs/
2+
scp ../tis-datax-executor.tar.gz root@192.168.28.201:/opt/misc/apache-dolphinscheduler-3.2.2-bin/standalone-server/

‎tis-datax/executor/powerjob-worker-samples/src/main/java/com/qlangtech/tis/datax/powerjob/TISInitializeProcessor.java‎

Lines changed: 0 additions & 48 deletions
Original file line numberDiff line numberDiff line change
@@ -18,54 +18,6 @@ public ProcessResult process(TaskContext context) throws Exception {
1818

1919
this.initializeProcess(new PowerJobTaskContext(context));
2020
return new ProcessResult(true);
21-
// OmsLogger omsLogger = context.getOmsLogger();
22-
// Pair<Boolean, JSONObject> instanceParams = TISTableDumpProcessor.getInstanceParams(context);
23-
// RpcServiceReference rpcSvcRef = TISTableDumpProcessor.getRpcServiceReference();
24-
// StatusRpcClientFactory.AssembleSvcCompsite svc = rpcSvcRef.get();
25-
// if (!instanceParams.getLeft()) {
26-
// // 说明是定时任务触发
27-
// omsLogger.info("trigger by crontab,now shall create taskId");
28-
// InitializeNodeCfg initNodeCfg = InitializeNodeCfg.parse(context);
29-
//
30-
// IExecChainContext.TriggerNewTaskParam triggerParams =
31-
// new IExecChainContext.TriggerNewTaskParam(context.getWorkflowContext().getWfInstanceId(),
32-
// initNodeCfg.getDataXName(), initNodeCfg.isTisDataflowType()) {
33-
// @Override
34-
// public List<HttpUtils.PostParam> params() {
35-
// List<HttpUtils.PostParam> params = super.params();
36-
// if (initNodeCfg.isTisDataflowType()) {
37-
// // 为了满足 OfflineDatasourceAction.doExecuteWorkflow 执行
38-
// params.add(new HttpUtils.PostParam("id", initNodeCfg.getWorkflowId()));
39-
// params.add(new HttpUtils.PostParam(IFullBuildContext.DRY_RUN, false));
40-
// }
41-
// return params;
42-
// }
43-
// };
44-
// /**=======================================================================
45-
// *TriggerNewTask
46-
// =======================================================================*/
47-
// PowerjobTriggerBuildResult triggerResult = IExecChainContext.triggerNewTask(triggerParams);
48-
// svc.appendLog(Level.INFO, triggerResult.getTaskid(), Optional.empty(), "start to execute data synchronize pipeline:" + String.valueOf(initNodeCfg));
49-
//
50-
// WorkflowContext wfContext = context.getWorkflowContext();
51-
// omsLogger.info("create task context,taskId:{},name:{}", triggerResult.getTaskid(), initNodeCfg.getDataXName());
52-
// JSONObject iparams = IExecChainContext.createInstanceParams(triggerResult.getTaskid(), () -> initNodeCfg.getDataXName(),
53-
// false, Optional.of(triggerResult.getPluginCfgsMetas()));
54-
// for (Map.Entry<String, Object> e : iparams.entrySet()) {
55-
// wfContext.appendData2WfContext(e.getKey(), e.getValue());
56-
// }
57-
// wfContext.appendData2WfContext(JobParams.KEY_JAVA_MEMORY_SPEC, triggerResult.getJavaMemorySpec());
58-
// if (triggerResult.getPreviousTaskId() != null) {
59-
// wfContext.appendData2WfContext(JobParams.KEY_PREVIOUS_TASK_ID, triggerResult.getPreviousTaskId());
60-
// }
61-
// } else {
62-
// // trigger by mannual 手动触发
63-
// Integer taskId = TISTableDumpProcessor.parseTaskId(instanceParams.getRight());
64-
// svc.appendLog(Level.INFO, taskId, Optional.empty()
65-
// , "start to execute data synchronize pipeline:" + JsonUtil.toString(instanceParams, false));
66-
// }
67-
68-
6921
}
7022

7123

‎tis-datax/tis-datax-common-plugin/src/main/java/com/qlangtech/tis/plugin/datax/BasicWorkflowPayload.java‎

Lines changed: 5 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -503,10 +503,11 @@ private void setInitStatus(AbstractChildProcessStatus status) {
503503
}
504504

505505
protected JSONObject createInitNodeJson() {
506-
JSONObject initNode = new JSONObject();
507-
initNode.put(StoreResourceType.DATAX_NAME, dataxProcessor.identityValue());
508-
// 是否是dataflow的处理类型
509-
initNode.put(StoreResourceType.KEY_STORE_RESOURCE_TYPE, dataxProcessor.getResType().getType());
506+
JSONObject initNode = dataxProcessor.createNode();
507+
// new JSONObject();
508+
// initNode.put(StoreResourceType.DATAX_NAME, dataxProcessor.identityValue());
509+
// // 是否是dataflow的处理类型
510+
// initNode.put(StoreResourceType.KEY_STORE_RESOURCE_TYPE, dataxProcessor.getResType().getType());
510511
return initNode;
511512
}
512513

‎tis-datax/tis-datax-common-plugin/src/main/java/com/qlangtech/tis/plugin/datax/TriggrWorkflowJobs.java‎

Lines changed: 4 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -145,13 +145,14 @@ public void offer(WorkFlowBuildHistoryPayload workFlowBuildHistoryPayload) {
145145

146146

147147
private void appendLog(WorkFlowBuildHistoryPayload workFlowBuildHistoryPayload, Optional<ExecResult> execResult) {
148-
JSONObject wfHistory = new JSONObject();
148+
// JSONObject wfHistory = new JSONObject();
149149
IDataxProcessor dataxProcessor = Objects.requireNonNull(workFlowBuildHistoryPayload.dataxProcessor, "dataxProcessor can not be null");
150+
JSONObject wfHistory = dataxProcessor.createNode();
150151
wfHistory.put(KEY_WORKFLOW_INSTANCE_ID, workFlowBuildHistoryPayload.getSPIWorkflowInstanceId());
151152
wfHistory.put(KEY_TIS_TASK_ID, workFlowBuildHistoryPayload.getTisTaskId());
152153
wfHistory.put(KEY_WORKFLOW_BUILD_HISTORY_FACTORY, workFlowBuildHistoryPayload.getFactory().getName());
153-
wfHistory.put(StoreResourceType.KEY_STORE_RESOURCE_TYPE, dataxProcessor.getResType().getType());
154-
wfHistory.put(IFullBuildContext.KEY_APP_NAME, dataxProcessor.identityValue());
154+
// wfHistory.put(StoreResourceType.KEY_STORE_RESOURCE_TYPE, dataxProcessor.getResType().getType());
155+
// wfHistory.put(IFullBuildContext.KEY_APP_NAME, dataxProcessor.identityValue());
155156

156157
// DataxProcessor.load()
157158
execResult.ifPresent((er) -> {

‎tis-datax/tis-datax-dolphinscheduler-plugin/src/main/java/com/qlangtech/tis/plugin/datax/doplinscheduler/DSWorkflowPayload.java‎

Lines changed: 9 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -27,6 +27,7 @@
2727
import com.qlangtech.tis.coredefine.module.action.PowerjobTriggerBuildResult;
2828
import com.qlangtech.tis.dao.ICommonDAOContext;
2929
import com.qlangtech.tis.datax.IDataxProcessor;
30+
import com.qlangtech.tis.datax.StoreResourceType;
3031
import com.qlangtech.tis.datax.job.DataXJobWorker;
3132
import com.qlangtech.tis.exec.ExecutePhaseRange;
3233
import com.qlangtech.tis.fullbuild.IFullBuildContext;
@@ -123,7 +124,7 @@ protected DSWorkflowInstance createWorkflowInstance(Long spiWorkflowId, ExecuteP
123124

124125
@Override
125126
protected ReplicasSpec getResourceSeplicasSpec() {
126-
// return ReplicasSpec.createDftPowerjobServerReplicasSpec();
127+
// return ReplicasSpec.createDftPowerjobServerReplicasSpec();
127128
ExportTISPipelineToDolphinscheduler dsExport = Objects.requireNonNull(this.exportCfg, "exportCfg can not be null");
128129
return Objects.requireNonNull(dsExport.memorySpec, "memorySpec can not be null").getMemorySpec();
129130
}
@@ -135,7 +136,7 @@ protected DataXJobWorker getSPIJobWorker() {
135136

136137
/**
137138
* 为了避免
138-
*
139+
* <p>
139140
* // @see com.qlangtech.tis.datax.powerjob.CfgsSnapshotConsumer 中执行 synchronizTpisAndConfs方法processTaskIds由于
140141
* taskid不变导致停止本地文件更新本地配置，所以需要让 baseTisTaskIdWithLocalExecutor 参数每次来都往上递增
141142
*/
@@ -283,7 +284,7 @@ private void innerSaveJob(
283284
// https://github.com/apache/dolphinscheduler/blob/57c80f2af5d1f29417fdbea09b27221f54987655/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/TaskDefinitionServiceImpl.java#L855
284285
SelectedTabTriggers tabTriggers = null;
285286
ISelectedTab selectedTab = null;
286-
JSONObject mrParams;
287+
287288
boolean containPostTrigger = false;
288289
Optional<IWorkflowNode> changedWfNode = null;
289290

@@ -310,7 +311,10 @@ private void innerSaveJob(
310311
, createParam(IFullBuildContext.DRY_RUN, ParamType.BOOLEAN)
311312
, createParam(PluginAndCfgsSnapshotUtils.KEY_PLUGIN_CFGS_METAS, ParamType.VARCHAR)
312313
, createParam(JobParams.KEY_JAVA_MEMORY_SPEC, ParamType.VARCHAR)
313-
, createParam(JobParams.KEY_PREVIOUS_TASK_ID, ParamType.INTEGER)));
314+
, createParam(JobParams.KEY_PREVIOUS_TASK_ID, ParamType.INTEGER)
315+
// , createParam(StoreResourceType.KEY_STORE_RESOURCE_TYPE, ParamType.VARCHAR)
316+
));
317+
314318
JSONObject initNode = createInitNodeJson();
315319
this.setDisableGrpcRemoteServerConnect(initNode);
316320
//
@@ -331,7 +335,7 @@ private void innerSaveJob(
331335
//======================================
332336
tabTriggers = entry.getValue();
333337
selectedTab = entry.getKey();
334-
mrParams = new JSONObject(tabTriggers.createMRParams());
338+
JSONObject mrParams = new JSONObject(tabTriggers.createMRParams());
335339
this.setDisableGrpcRemoteServerConnect(mrParams);
336340
if (tabTriggers.getPostTrigger() != null) {
337341
containPostTrigger = true;

‎tis-datax/tis-datax-dolphinscheduler-plugin/src/main/java/com/qlangtech/tis/plugin/datax/doplinscheduler/export/DolphinSchedulerEndpoint.java‎

Lines changed: 0 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -21,7 +21,6 @@
2121
import com.alibaba.citrus.turbine.Context;
2222
import com.google.common.collect.Lists;
2323
import com.qlangtech.tis.config.ParamsConfig;
24-
import com.qlangtech.tis.extension.Descriptor;
2524
import com.qlangtech.tis.extension.TISExtension;
2625
import com.qlangtech.tis.lang.TisException;
2726
import com.qlangtech.tis.manage.common.ConfigFileContext.Header;

‎tis-datax/tis-datax-dolphinscheduler-plugin/src/main/resources/com/qlangtech/tis/plugin/datax/doplinscheduler/export/DolphinSchedulerEndpoint.md‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,4 @@
1+
## serverPath
2+
本地安装的Apache Dolphinscheduler服务端地址，例如：http://192.168.28.201:12345/dolphinscheduler
3+
## serverToken
4+
在 Apache Dolphinscheduler服务端->安全中心->令牌管理（ http://192.168.28.201:12345/dolphinscheduler/ui/security/token-manage ），中配置的用户访问令牌。

‎tis-datax/tis-hive-flat-table-builder-plugin/src/main/java/com/qlangtech/tis/hive/DefaultHiveConnGetter.java‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -220,6 +220,7 @@ public IHiveMetaStore visit(IKerberosUserToken token) {
220220

221221
private IHiveMetaStore createHiveMetaStore() {
222222
try {
223+
// final IMetaStoreClient storeClient = Hive.get(hiveCfg, true).getMSC();
223224
final IMetaStoreClient storeClient = Hive.getWithFastCheck(hiveCfg, false).getMSC();
224225
return new DefaultHiveMetaStore(hiveCfg, storeClient, metaStoreUrls);
225226
} catch (Exception e) {

0 commit comments

Comments
 (0)