Skip to content

Commit cf1d21d

Browse files
committed
[Fix-16617][Seatunnel] Use master option for engine deploy mode
1 parent c9e373e commit cf1d21d

2 files changed

Lines changed: 117 additions & 1 deletion

File tree

dolphinscheduler-task-plugin/dolphinscheduler-task-seatunnel/src/main/java/org/apache/dolphinscheduler/plugin/task/seatunnel/self/SeatunnelEngineTask.java

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -46,7 +46,7 @@ public void init() {
4646
public List<String> buildOptions() throws Exception {
4747
List<String> args = super.buildOptions();
4848
if (!Objects.isNull(seatunnelParameters.getDeployMode())) {
49-
args.add(Constants.DEPLOY_MODE_OPTIONS);
49+
args.add(Constants.MASTER_OPTIONS);
5050
args.add(seatunnelParameters.getDeployMode().getCommand());
5151
}
5252
if (StringUtils.isNotBlank(seatunnelParameters.getOthers())) {
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,116 @@
1+
/*
2+
* Licensed to the Apache Software Foundation (ASF) under one or more
3+
* contributor license agreements. See the NOTICE file distributed with
4+
* this work for additional information regarding copyright ownership.
5+
* The ASF licenses this file to You under the Apache License, Version 2.0
6+
* (the "License"); you may not use this file except in compliance with
7+
* the License. You may obtain a copy of the License at
8+
*
9+
* http://www.apache.org/licenses/LICENSE-2.0
10+
*
11+
* Unless required by applicable law or agreed to in writing, software
12+
* distributed under the License is distributed on an "AS IS" BASIS,
13+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14+
* See the License for the specific language governing permissions and
15+
* limitations under the License.
16+
*/
17+
18+
package org.apache.dolphinscheduler.plugin.task.seatunnel.self;
19+
20+
import org.apache.dolphinscheduler.common.utils.JSONUtils;
21+
import org.apache.dolphinscheduler.plugin.task.api.TaskExecutionContext;
22+
import org.apache.dolphinscheduler.plugin.task.seatunnel.DeployModeEnum;
23+
24+
import org.apache.commons.io.FileUtils;
25+
26+
import java.lang.reflect.Field;
27+
28+
import org.junit.jupiter.api.AfterEach;
29+
import org.junit.jupiter.api.Assertions;
30+
import org.junit.jupiter.api.BeforeEach;
31+
import org.junit.jupiter.api.Test;
32+
import org.mockito.MockedStatic;
33+
import org.mockito.Mockito;
34+
35+
public class SeatunnelEngineTaskTest {
36+
37+
private static final String EXECUTE_PATH = "/tmp";
38+
private static final String TASK_ID = "1234";
39+
private static final String RAW_SCRIPT =
40+
"env {\n job.mode = \"BATCH\"\n}\nsource {\n FakeSource {}\n}\nsink {\n Console {}\n}";
41+
42+
private MockedStatic<FileUtils> mockedStaticFileUtils;
43+
44+
@BeforeEach
45+
public void setUp() {
46+
mockedStaticFileUtils = Mockito.mockStatic(FileUtils.class);
47+
}
48+
49+
@AfterEach
50+
public void after() {
51+
mockedStaticFileUtils.close();
52+
}
53+
54+
@Test
55+
public void buildOptionsUsesMasterForDeployMode() throws Exception {
56+
SeatunnelEngineParameters seatunnelParameters = newSeatunnelEngineParameters();
57+
seatunnelParameters.setDeployMode(DeployModeEnum.cluster);
58+
59+
SeatunnelEngineTask task = newSeatunnelEngineTask(seatunnelParameters);
60+
61+
String command = String.join(" ", task.buildOptions());
62+
String expectedCommand = String.format("--config %s/seatunnel_%s.conf --master cluster", EXECUTE_PATH, TASK_ID);
63+
Assertions.assertEquals(expectedCommand, command);
64+
}
65+
66+
@Test
67+
public void buildOptionsOmitsMasterWhenDeployModeMissing() throws Exception {
68+
SeatunnelEngineParameters seatunnelParameters = newSeatunnelEngineParameters();
69+
70+
SeatunnelEngineTask task = newSeatunnelEngineTask(seatunnelParameters);
71+
72+
String command = String.join(" ", task.buildOptions());
73+
String expectedCommand = String.format("--config %s/seatunnel_%s.conf", EXECUTE_PATH, TASK_ID);
74+
Assertions.assertEquals(expectedCommand, command);
75+
}
76+
77+
@Test
78+
public void buildOptionsAppendsOthersWhenPresent() throws Exception {
79+
SeatunnelEngineParameters seatunnelParameters = newSeatunnelEngineParameters();
80+
seatunnelParameters.setDeployMode(DeployModeEnum.local);
81+
seatunnelParameters.setOthers("--name demo-job");
82+
83+
SeatunnelEngineTask task = newSeatunnelEngineTask(seatunnelParameters);
84+
85+
String command = String.join(" ", task.buildOptions());
86+
String expectedCommand =
87+
String.format("--config %s/seatunnel_%s.conf --master local --name demo-job", EXECUTE_PATH, TASK_ID);
88+
Assertions.assertEquals(expectedCommand, command);
89+
}
90+
91+
private SeatunnelEngineParameters newSeatunnelEngineParameters() {
92+
SeatunnelEngineParameters seatunnelParameters = new SeatunnelEngineParameters();
93+
seatunnelParameters.setUseCustom(true);
94+
seatunnelParameters.setRawScript(RAW_SCRIPT);
95+
return seatunnelParameters;
96+
}
97+
98+
private SeatunnelEngineTask newSeatunnelEngineTask(SeatunnelEngineParameters seatunnelParameters) throws Exception {
99+
TaskExecutionContext taskExecutionContext = new TaskExecutionContext();
100+
taskExecutionContext.setExecutePath(EXECUTE_PATH);
101+
taskExecutionContext.setTaskAppId(TASK_ID);
102+
taskExecutionContext.setTaskParams(JSONUtils.toJsonString(seatunnelParameters));
103+
104+
SeatunnelEngineTask task = new SeatunnelEngineTask(taskExecutionContext);
105+
task.setSeatunnelParameters(seatunnelParameters);
106+
setPrivateSeatunnelParameters(task, seatunnelParameters);
107+
return task;
108+
}
109+
110+
private void setPrivateSeatunnelParameters(SeatunnelEngineTask task,
111+
SeatunnelEngineParameters seatunnelParameters) throws Exception {
112+
Field field = SeatunnelEngineTask.class.getDeclaredField("seatunnelParameters");
113+
field.setAccessible(true);
114+
field.set(task, seatunnelParameters);
115+
}
116+
}

0 commit comments

Comments
 (0)