getResourceFilesList() {
+ return Collections.emptyList();
+ }
+
+ @Override
+ public String toString() {
+ return "EmrServerlessParameters{"
+ + "applicationId='" + applicationId + '\''
+ + ", executionRoleArn='" + executionRoleArn + '\''
+ + ", jobName='" + jobName + '\''
+ + ", startJobRunRequestJson='" + startJobRunRequestJson + '\''
+ + '}';
+ }
+}
diff --git a/dolphinscheduler-task-plugin/dolphinscheduler-task-emr-serverless/src/main/java/org/apache/dolphinscheduler/plugin/task/emrserverless/EmrServerlessTask.java b/dolphinscheduler-task-plugin/dolphinscheduler-task-emr-serverless/src/main/java/org/apache/dolphinscheduler/plugin/task/emrserverless/EmrServerlessTask.java
new file mode 100644
index 000000000000..e34c3c0287c7
--- /dev/null
+++ b/dolphinscheduler-task-plugin/dolphinscheduler-task-emr-serverless/src/main/java/org/apache/dolphinscheduler/plugin/task/emrserverless/EmrServerlessTask.java
@@ -0,0 +1,318 @@
+/*
+ * 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.plugin.task.emrserverless;
+
+import static com.fasterxml.jackson.databind.DeserializationFeature.ACCEPT_EMPTY_ARRAY_AS_NULL_OBJECT;
+import static com.fasterxml.jackson.databind.DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES;
+import static com.fasterxml.jackson.databind.DeserializationFeature.READ_UNKNOWN_ENUM_VALUES_AS_NULL;
+import static com.fasterxml.jackson.databind.MapperFeature.REQUIRE_SETTERS_FOR_GETTERS;
+
+import org.apache.dolphinscheduler.authentication.aws.AWSCredentialsProviderFactor;
+import org.apache.dolphinscheduler.authentication.aws.AwsConfigurationKeys;
+import org.apache.dolphinscheduler.common.constants.SystemConstants;
+import org.apache.dolphinscheduler.common.utils.JSONUtils;
+import org.apache.dolphinscheduler.common.utils.PropertyUtils;
+import org.apache.dolphinscheduler.plugin.task.api.AbstractRemoteTask;
+import org.apache.dolphinscheduler.plugin.task.api.TaskConstants;
+import org.apache.dolphinscheduler.plugin.task.api.TaskException;
+import org.apache.dolphinscheduler.plugin.task.api.TaskExecutionContext;
+import org.apache.dolphinscheduler.plugin.task.api.parameters.AbstractParameters;
+import org.apache.dolphinscheduler.plugin.task.api.utils.ParameterUtils;
+
+import org.apache.commons.lang3.StringUtils;
+
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+
+import lombok.extern.slf4j.Slf4j;
+
+import com.amazonaws.SdkBaseException;
+import com.amazonaws.auth.AWSCredentialsProvider;
+import com.amazonaws.auth.DefaultAWSCredentialsProviderChain;
+import com.amazonaws.services.emrserverless.AWSEMRServerless;
+import com.amazonaws.services.emrserverless.AWSEMRServerlessClientBuilder;
+import com.amazonaws.services.emrserverless.model.CancelJobRunRequest;
+import com.amazonaws.services.emrserverless.model.CancelJobRunResult;
+import com.amazonaws.services.emrserverless.model.GetJobRunRequest;
+import com.amazonaws.services.emrserverless.model.GetJobRunResult;
+import com.amazonaws.services.emrserverless.model.JobRun;
+import com.amazonaws.services.emrserverless.model.StartJobRunRequest;
+import com.amazonaws.services.emrserverless.model.StartJobRunResult;
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.PropertyNamingStrategy;
+import com.google.common.collect.Sets;
+
+/**
+ * Amazon EMR Serverless Task.
+ *
+ * Submits a job run to an EMR Serverless application and tracks it until completion.
+ * Supports Spark and Hive job types.
+ *
+ *
+ * @since dev-SNAPSHOT
+ */
+@Slf4j
+public class EmrServerlessTask extends AbstractRemoteTask {
+
+ /**
+ * EMR Serverless job run states that indicate the job is still in progress.
+ */
+ private static final HashSet WAITING_STATES = Sets.newHashSet(
+ "SUBMITTED", "PENDING", "SCHEDULED", "RUNNING");
+
+ /**
+ * ObjectMapper configured for AWS SDK request/response deserialization.
+ */
+ static final ObjectMapper objectMapper = new ObjectMapper()
+ .configure(FAIL_ON_UNKNOWN_PROPERTIES, false)
+ .configure(ACCEPT_EMPTY_ARRAY_AS_NULL_OBJECT, true)
+ .configure(READ_UNKNOWN_ENUM_VALUES_AS_NULL, true)
+ .configure(REQUIRE_SETTERS_FOR_GETTERS, true)
+ .setTimeZone(SystemConstants.DEFAULT_TIME_ZONE)
+ .setPropertyNamingStrategy(new PropertyNamingStrategy.UpperCamelCaseStrategy());
+
+ private final TaskExecutionContext taskExecutionContext;
+
+ private EmrServerlessParameters emrServerlessParameters;
+
+ private AWSEMRServerless emrServerlessClient;
+
+ /**
+ * applicationId from parameters
+ */
+ private String applicationId;
+
+ /**
+ * jobRunId returned by StartJobRun or recovered from appIds
+ */
+ private String jobRunId;
+
+ protected EmrServerlessTask(TaskExecutionContext taskExecutionContext) {
+ super(taskExecutionContext);
+ this.taskExecutionContext = taskExecutionContext;
+ }
+
+ @Override
+ public void init() {
+ final String taskParams = taskExecutionContext.getTaskParams();
+ emrServerlessParameters = JSONUtils.parseObject(taskParams, EmrServerlessParameters.class);
+ log.info("Initialize EMR Serverless task params: {}", JSONUtils.toPrettyJsonString(taskParams));
+
+ if (emrServerlessParameters == null || !emrServerlessParameters.checkParameters()) {
+ throw new EmrServerlessTaskException("EMR Serverless task params are not valid");
+ }
+
+ applicationId = emrServerlessParameters.getApplicationId();
+ emrServerlessClient = createEmrServerlessClient();
+ }
+
+ @Override
+ public void submitApplication() throws TaskException {
+ try {
+ StartJobRunRequest request = buildStartJobRunRequest();
+
+ log.info("Submitting EMR Serverless job run to application: {}", applicationId);
+ StartJobRunResult result = emrServerlessClient.startJobRun(request);
+
+ jobRunId = result.getJobRunId();
+ // Store applicationId:jobRunId for failover recovery
+ setAppIds(applicationId + ":" + jobRunId);
+ log.info("Successfully submitted EMR Serverless job run, jobRunId: {}", jobRunId);
+
+ } catch (EmrServerlessTaskException | SdkBaseException e) {
+ log.error("EMR Serverless task submit failed", e);
+ throw new TaskException("EMR Serverless task submit failed", e);
+ }
+ }
+
+ @Override
+ public void trackApplicationStatus() throws TaskException {
+ try {
+ // Recover applicationId and jobRunId from appIds if needed (failover case)
+ if (StringUtils.isEmpty(jobRunId) && StringUtils.isNotEmpty(getAppIds())) {
+ String[] parts = getAppIds().split(":");
+ if (parts.length == 2) {
+ applicationId = parts[0];
+ jobRunId = parts[1];
+ log.info("Recovered EMR Serverless job from appIds - applicationId: {}, jobRunId: {}",
+ applicationId, jobRunId);
+ }
+ }
+
+ if (StringUtils.isEmpty(jobRunId)) {
+ throw new EmrServerlessTaskException("jobRunId is empty, cannot track application status");
+ }
+
+ String currentState = getJobRunState();
+ while (WAITING_STATES.contains(currentState)) {
+ TimeUnit.SECONDS.sleep(10);
+ currentState = getJobRunState();
+ }
+
+ final int exitCode = mapStateToExitCode(currentState);
+ setExitStatusCode(exitCode);
+ log.info("EMR Serverless job run [{}] finished with state: {}, exitCode: {}",
+ jobRunId, currentState, exitCode);
+
+ } catch (EmrServerlessTaskException | SdkBaseException e) {
+ log.error("EMR Serverless task tracking failed", e);
+ setExitStatusCode(TaskConstants.EXIT_CODE_FAILURE);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new TaskException("EMR Serverless task tracking interrupted", e);
+ }
+ }
+
+ @Override
+ public void cancelApplication() throws TaskException {
+ if (StringUtils.isEmpty(jobRunId)) {
+ log.warn("jobRunId is empty, skip cancel");
+ return;
+ }
+ log.info("Cancelling EMR Serverless job run, applicationId: {}, jobRunId: {}", applicationId, jobRunId);
+ try {
+ CancelJobRunRequest request = new CancelJobRunRequest()
+ .withApplicationId(applicationId)
+ .withJobRunId(jobRunId);
+ CancelJobRunResult result = emrServerlessClient.cancelJobRun(request);
+ log.info("Cancel job run result: {}", result);
+ } catch (SdkBaseException e) {
+ throw new TaskException("Failed to cancel EMR Serverless job run", e);
+ }
+ }
+
+ @Override
+ public List getApplicationIds() throws TaskException {
+ return Collections.emptyList();
+ }
+
+ @Override
+ public AbstractParameters getParameters() {
+ return emrServerlessParameters;
+ }
+
+ /**
+ * Build StartJobRunRequest from parameters and user-provided JSON.
+ */
+ private StartJobRunRequest buildStartJobRunRequest() {
+ String startJobRunRequestJson;
+ try {
+ startJobRunRequestJson = ParameterUtils.convertParameterPlaceholders(
+ emrServerlessParameters.getStartJobRunRequestJson(),
+ ParameterUtils.convert(taskExecutionContext.getPrepareParamsMap()));
+ } catch (Exception e) {
+ throw new EmrServerlessTaskException("Failed to resolve parameter placeholders", e);
+ }
+
+ StartJobRunRequest request;
+ try {
+ request = objectMapper.readValue(startJobRunRequestJson, StartJobRunRequest.class);
+ } catch (JsonProcessingException e) {
+ throw new EmrServerlessTaskException(
+ "Cannot parse StartJobRunRequest from JSON: " + startJobRunRequestJson, e);
+ }
+
+ // Override applicationId and executionRoleArn from top-level parameters
+ request.setApplicationId(applicationId);
+ request.setExecutionRoleArn(emrServerlessParameters.getExecutionRoleArn());
+
+ // Set job name if provided
+ if (StringUtils.isNotEmpty(emrServerlessParameters.getJobName())) {
+ request.setName(emrServerlessParameters.getJobName());
+ } else {
+ request.setName(taskExecutionContext.getTaskName());
+ }
+
+ // Set client token for idempotency
+ request.setClientToken(taskExecutionContext.getTaskInstanceId() + "-" + System.currentTimeMillis());
+
+ return request;
+ }
+
+ /**
+ * Get the current state of the job run.
+ */
+ private String getJobRunState() {
+ GetJobRunRequest request = new GetJobRunRequest()
+ .withApplicationId(applicationId)
+ .withJobRunId(jobRunId);
+ GetJobRunResult result = emrServerlessClient.getJobRun(request);
+
+ if (result == null || result.getJobRun() == null) {
+ throw new EmrServerlessTaskException("Failed to get job run status");
+ }
+
+ JobRun jobRun = result.getJobRun();
+ String state = jobRun.getState();
+ log.info("EMR Serverless job run [applicationId:{}, jobRunId:{}] state: {}", applicationId, jobRunId, state);
+ return state;
+ }
+
+ /**
+ * Map EMR Serverless job run final state to DolphinScheduler exit code.
+ */
+ private int mapStateToExitCode(String state) {
+ if (state == null) {
+ return TaskConstants.EXIT_CODE_FAILURE;
+ }
+ switch (state) {
+ case "SUCCESS":
+ return TaskConstants.EXIT_CODE_SUCCESS;
+ case "CANCELLED":
+ return TaskConstants.EXIT_CODE_KILL;
+ case "FAILED":
+ default:
+ return TaskConstants.EXIT_CODE_FAILURE;
+ }
+ }
+
+ /**
+ * Create EMR Serverless client.
+ * Strategy: try aws.emr.* config first, fallback to DefaultAWSCredentialsProviderChain.
+ */
+ protected AWSEMRServerless createEmrServerlessClient() {
+ Map awsProperties = PropertyUtils.getByPrefix("aws.emr.", "");
+ AWSEMRServerlessClientBuilder builder = AWSEMRServerlessClientBuilder.standard();
+
+ AWSCredentialsProvider credentialsProvider;
+ try {
+ credentialsProvider = AWSCredentialsProviderFactor.credentialsProvider(awsProperties);
+ log.info("Using AWS credentials from aws.emr.* configuration");
+ } catch (Exception e) {
+ log.info("No valid aws.emr.* credentials config found, falling back to DefaultAWSCredentialsProviderChain");
+ credentialsProvider = DefaultAWSCredentialsProviderChain.getInstance();
+ }
+ builder.withCredentials(credentialsProvider);
+
+ String region = awsProperties.get(AwsConfigurationKeys.AWS_REGION);
+ String endpoint = awsProperties.get(AwsConfigurationKeys.AWS_ENDPOINT);
+
+ if (StringUtils.isNotEmpty(endpoint) && StringUtils.isNotEmpty(region)) {
+ builder.withEndpointConfiguration(
+ new com.amazonaws.client.builder.AwsClientBuilder.EndpointConfiguration(endpoint, region));
+ } else if (StringUtils.isNotEmpty(region)) {
+ builder.withRegion(region);
+ }
+
+ return builder.build();
+ }
+}
diff --git a/dolphinscheduler-task-plugin/dolphinscheduler-task-emr-serverless/src/main/java/org/apache/dolphinscheduler/plugin/task/emrserverless/EmrServerlessTaskChannel.java b/dolphinscheduler-task-plugin/dolphinscheduler-task-emr-serverless/src/main/java/org/apache/dolphinscheduler/plugin/task/emrserverless/EmrServerlessTaskChannel.java
new file mode 100644
index 000000000000..a63cea6bbb6a
--- /dev/null
+++ b/dolphinscheduler-task-plugin/dolphinscheduler-task-emr-serverless/src/main/java/org/apache/dolphinscheduler/plugin/task/emrserverless/EmrServerlessTaskChannel.java
@@ -0,0 +1,37 @@
+/*
+ * 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.plugin.task.emrserverless;
+
+import org.apache.dolphinscheduler.common.utils.JSONUtils;
+import org.apache.dolphinscheduler.plugin.task.api.AbstractTask;
+import org.apache.dolphinscheduler.plugin.task.api.TaskChannel;
+import org.apache.dolphinscheduler.plugin.task.api.TaskExecutionContext;
+import org.apache.dolphinscheduler.plugin.task.api.parameters.AbstractParameters;
+
+public class EmrServerlessTaskChannel implements TaskChannel {
+
+ @Override
+ public AbstractTask createTask(TaskExecutionContext taskRequest) {
+ return new EmrServerlessTask(taskRequest);
+ }
+
+ @Override
+ public AbstractParameters parseParameters(String taskParams) {
+ return JSONUtils.parseObject(taskParams, EmrServerlessParameters.class);
+ }
+}
diff --git a/dolphinscheduler-task-plugin/dolphinscheduler-task-emr-serverless/src/main/java/org/apache/dolphinscheduler/plugin/task/emrserverless/EmrServerlessTaskChannelFactory.java b/dolphinscheduler-task-plugin/dolphinscheduler-task-emr-serverless/src/main/java/org/apache/dolphinscheduler/plugin/task/emrserverless/EmrServerlessTaskChannelFactory.java
new file mode 100644
index 000000000000..74e4b0e7ccab
--- /dev/null
+++ b/dolphinscheduler-task-plugin/dolphinscheduler-task-emr-serverless/src/main/java/org/apache/dolphinscheduler/plugin/task/emrserverless/EmrServerlessTaskChannelFactory.java
@@ -0,0 +1,37 @@
+/*
+ * 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.plugin.task.emrserverless;
+
+import org.apache.dolphinscheduler.plugin.task.api.TaskChannel;
+import org.apache.dolphinscheduler.plugin.task.api.TaskChannelFactory;
+
+import com.google.auto.service.AutoService;
+
+@AutoService(TaskChannelFactory.class)
+public class EmrServerlessTaskChannelFactory implements TaskChannelFactory {
+
+ @Override
+ public String getName() {
+ return "EMR_SERVERLESS";
+ }
+
+ @Override
+ public TaskChannel create() {
+ return new EmrServerlessTaskChannel();
+ }
+}
diff --git a/dolphinscheduler-task-plugin/dolphinscheduler-task-emr-serverless/src/main/java/org/apache/dolphinscheduler/plugin/task/emrserverless/EmrServerlessTaskException.java b/dolphinscheduler-task-plugin/dolphinscheduler-task-emr-serverless/src/main/java/org/apache/dolphinscheduler/plugin/task/emrserverless/EmrServerlessTaskException.java
new file mode 100644
index 000000000000..ea46565b6981
--- /dev/null
+++ b/dolphinscheduler-task-plugin/dolphinscheduler-task-emr-serverless/src/main/java/org/apache/dolphinscheduler/plugin/task/emrserverless/EmrServerlessTaskException.java
@@ -0,0 +1,31 @@
+/*
+ * 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.plugin.task.emrserverless;
+
+import org.apache.dolphinscheduler.plugin.task.api.TaskException;
+
+public class EmrServerlessTaskException extends TaskException {
+
+ public EmrServerlessTaskException(String message) {
+ super(message);
+ }
+
+ public EmrServerlessTaskException(String message, Throwable cause) {
+ super(message, cause);
+ }
+}
diff --git a/dolphinscheduler-task-plugin/pom.xml b/dolphinscheduler-task-plugin/pom.xml
index 6f3d727d59b9..830650f50f55 100644
--- a/dolphinscheduler-task-plugin/pom.xml
+++ b/dolphinscheduler-task-plugin/pom.xml
@@ -62,6 +62,7 @@
dolphinscheduler-task-datafactory
dolphinscheduler-task-remoteshell
dolphinscheduler-task-aliyunserverlessspark
+ dolphinscheduler-task-emr-serverless
diff --git a/dolphinscheduler-ui/src/locales/en_US/project.ts b/dolphinscheduler-ui/src/locales/en_US/project.ts
index 59edc7dfeaed..495f3e8b10d1 100644
--- a/dolphinscheduler-ui/src/locales/en_US/project.ts
+++ b/dolphinscheduler-ui/src/locales/en_US/project.ts
@@ -741,6 +741,14 @@ export default {
emr_flow_define_json_tips: 'Please enter the definition of the job flow.',
emr_steps_define_json: 'stepsDefineJson',
emr_steps_define_json_tips: 'Please enter the definition of the emr step.',
+ emr_serverless_application_id: 'Application ID',
+ emr_serverless_application_id_tips: 'Please enter the EMR Serverless Application ID',
+ emr_serverless_execution_role_arn: 'Execution Role ARN',
+ emr_serverless_execution_role_arn_tips: 'Please enter the IAM execution role ARN',
+ emr_serverless_job_name: 'Job Name',
+ emr_serverless_job_name_tips: 'Optional. Defaults to the task name if empty',
+ emr_serverless_start_job_run_json: 'StartJobRun JSON',
+ emr_serverless_start_job_run_json_tips: 'Please enter the StartJobRun request JSON (jobDriver, configurationOverrides, etc.)',
zeppelin_note_id: 'zeppelinNoteId',
zeppelin_note_id_tips: 'Please enter the note id of your zeppelin note',
zeppelin_paragraph_id: 'zeppelinParagraphId',
diff --git a/dolphinscheduler-ui/src/locales/zh_CN/project.ts b/dolphinscheduler-ui/src/locales/zh_CN/project.ts
index e6b08093c568..0f3da8189eeb 100644
--- a/dolphinscheduler-ui/src/locales/zh_CN/project.ts
+++ b/dolphinscheduler-ui/src/locales/zh_CN/project.ts
@@ -723,6 +723,14 @@ export default {
emr_flow_define_json_tips: '请输入工作流定义',
emr_steps_define_json: 'stepsDefineJson',
emr_steps_define_json_tips: '请输入EMR步骤定义',
+ emr_serverless_application_id: 'Application ID',
+ emr_serverless_application_id_tips: '请输入 EMR Serverless Application ID',
+ emr_serverless_execution_role_arn: '执行角色 ARN',
+ emr_serverless_execution_role_arn_tips: '请输入 IAM 执行角色 ARN',
+ emr_serverless_job_name: '作业名称',
+ emr_serverless_job_name_tips: '可选，为空时默认使用任务名称',
+ emr_serverless_start_job_run_json: 'StartJobRun JSON',
+ emr_serverless_start_job_run_json_tips: '请输入 StartJobRun 请求 JSON（jobDriver、configurationOverrides 等）',
zeppelin_note_id: 'zeppelinNoteId',
zeppelin_note_id_tips: '请输入zeppelin note id',
zeppelin_paragraph_id: 'zeppelinParagraphId',
diff --git a/dolphinscheduler-ui/src/store/project/types.ts b/dolphinscheduler-ui/src/store/project/types.ts
index ba7773fa5b14..a58135b3e741 100644
--- a/dolphinscheduler-ui/src/store/project/types.ts
+++ b/dolphinscheduler-ui/src/store/project/types.ts
@@ -38,6 +38,7 @@ type TaskType =
| 'SWITCH'
| 'SEATUNNEL'
| 'EMR'
+ | 'EMR_SERVERLESS'
| 'ZEPPELIN'
| 'K8S'
| 'JUPYTER'
diff --git a/dolphinscheduler-ui/src/views/projects/task/components/node/fields/index.ts b/dolphinscheduler-ui/src/views/projects/task/components/node/fields/index.ts
index 43c1ed66d138..da4cce0763e5 100644
--- a/dolphinscheduler-ui/src/views/projects/task/components/node/fields/index.ts
+++ b/dolphinscheduler-ui/src/views/projects/task/components/node/fields/index.ts
@@ -66,6 +66,7 @@ export { useDataX } from './use-datax'
export { useConditions } from './use-conditions'
export { useDependent } from './use-dependent'
export { useEmr } from './use-emr'
+export { useEmrServerless } from './use-emr-serverless'
export { useZeppelin } from './use-zeppelin'
export { useNamespace } from './use-namespace'
export { useK8s } from './use-k8s'
diff --git a/dolphinscheduler-ui/src/views/projects/task/components/node/fields/use-emr-serverless.ts b/dolphinscheduler-ui/src/views/projects/task/components/node/fields/use-emr-serverless.ts
new file mode 100644
index 000000000000..ec164275daff
--- /dev/null
+++ b/dolphinscheduler-ui/src/views/projects/task/components/node/fields/use-emr-serverless.ts
@@ -0,0 +1,82 @@
+/*
+ * 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.
+ */
+import { useI18n } from 'vue-i18n'
+import { useCustomParams } from '.'
+import type { IJsonItem } from '../types'
+
+export function useEmrServerless(model: { [field: string]: any }): IJsonItem[] {
+ const { t } = useI18n()
+
+ return [
+ {
+ type: 'input',
+ field: 'applicationId',
+ span: 24,
+ name: t('project.node.emr_serverless_application_id'),
+ props: {
+ placeholder: t('project.node.emr_serverless_application_id_tips')
+ },
+ validate: {
+ required: true,
+ trigger: ['input', 'blur'],
+ message: t('project.node.emr_serverless_application_id_tips')
+ }
+ },
+ {
+ type: 'input',
+ field: 'executionRoleArn',
+ span: 24,
+ name: t('project.node.emr_serverless_execution_role_arn'),
+ props: {
+ placeholder: t('project.node.emr_serverless_execution_role_arn_tips')
+ },
+ validate: {
+ required: true,
+ trigger: ['input', 'blur'],
+ message: t('project.node.emr_serverless_execution_role_arn_tips')
+ }
+ },
+ {
+ type: 'input',
+ field: 'jobName',
+ span: 24,
+ name: t('project.node.emr_serverless_job_name'),
+ props: {
+ placeholder: t('project.node.emr_serverless_job_name_tips')
+ }
+ },
+ {
+ type: 'editor',
+ field: 'startJobRunRequestJson',
+ span: 24,
+ name: t('project.node.emr_serverless_start_job_run_json'),
+ props: {
+ language: 'json'
+ },
+ validate: {
+ trigger: ['input', 'trigger'],
+ required: true,
+ message: t('project.node.emr_serverless_start_job_run_json_tips')
+ }
+ },
+ ...useCustomParams({
+ model,
+ field: 'localParams',
+ isSimple: true
+ })
+ ]
+}
diff --git a/dolphinscheduler-ui/src/views/projects/task/components/node/format-data.ts b/dolphinscheduler-ui/src/views/projects/task/components/node/format-data.ts
index 8fc67906775c..b20460c90279 100644
--- a/dolphinscheduler-ui/src/views/projects/task/components/node/format-data.ts
+++ b/dolphinscheduler-ui/src/views/projects/task/components/node/format-data.ts
@@ -303,6 +303,13 @@ export function formatParams(data: INodeData): {
taskParams.stepsDefineJson = data.stepsDefineJson
}
+ if (data.taskType === 'EMR_SERVERLESS') {
+ taskParams.applicationId = data.applicationId
+ taskParams.executionRoleArn = data.executionRoleArn
+ taskParams.jobName = data.jobName
+ taskParams.startJobRunRequestJson = data.startJobRunRequestJson
+ }
+
if (data.taskType === 'ZEPPELIN') {
taskParams.noteId = data.noteId
taskParams.paragraphId = data.paragraphId
diff --git a/dolphinscheduler-ui/src/views/projects/task/components/node/tasks/index.ts b/dolphinscheduler-ui/src/views/projects/task/components/node/tasks/index.ts
index 3db6bb96ea5b..89fa49e17b4a 100644
--- a/dolphinscheduler-ui/src/views/projects/task/components/node/tasks/index.ts
+++ b/dolphinscheduler-ui/src/views/projects/task/components/node/tasks/index.ts
@@ -33,6 +33,7 @@ import { useConditions } from './use-conditions'
import { useDataX } from './use-datax'
import { useDependent } from './use-dependent'
import { useEmr } from './use-emr'
+import { useEmrServerless } from './use-emr-serverless'
import { useZeppelin } from './use-zeppelin'
import { useK8s } from './use-k8s'
import { useJupyter } from './use-jupyter'
@@ -70,6 +71,7 @@ export default {
DATAX: useDataX,
DEPENDENT: useDependent,
EMR: useEmr,
+ EMR_SERVERLESS: useEmrServerless,
ZEPPELIN: useZeppelin,
K8S: useK8s,
JUPYTER: useJupyter,
diff --git a/dolphinscheduler-ui/src/views/projects/task/components/node/tasks/use-emr-serverless.ts b/dolphinscheduler-ui/src/views/projects/task/components/node/tasks/use-emr-serverless.ts
new file mode 100644
index 000000000000..1b3408262718
--- /dev/null
+++ b/dolphinscheduler-ui/src/views/projects/task/components/node/tasks/use-emr-serverless.ts
@@ -0,0 +1,72 @@
+/*
+ * 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.
+ */
+
+import { reactive } from 'vue'
+import * as Fields from '../fields/index'
+import type { IJsonItem, INodeData } from '../types'
+import { ITaskData } from '../types'
+
+export function useEmrServerless({
+ projectCode,
+ from = 0,
+ readonly,
+ data
+}: {
+ projectCode: number
+ from?: number
+ readonly?: boolean
+ data?: ITaskData
+}) {
+ const model = reactive({
+ name: '',
+ taskType: 'EMR_SERVERLESS',
+ flag: 'YES',
+ description: '',
+ timeoutFlag: false,
+ localParams: [],
+ environmentCode: null,
+ failRetryInterval: 1,
+ failRetryTimes: 0,
+ workerGroup: 'default',
+ delayTime: 0,
+ timeout: 30,
+ applicationId: '',
+ executionRoleArn: '',
+ jobName: '',
+ startJobRunRequestJson: '',
+ timeoutNotifyStrategy: ['WARN']
+ } as INodeData)
+
+ return {
+ json: [
+ Fields.useName(from),
+ ...Fields.useTaskDefinition({ projectCode, from, readonly, data, model }),
+ Fields.useRunFlag(),
+ Fields.useDescription(),
+ Fields.useTaskPriority(),
+ Fields.useWorkerGroup(projectCode),
+ Fields.useEnvironmentName(model, !data?.id),
+ ...Fields.useTaskGroup(model, projectCode),
+ ...Fields.useFailed(),
+ Fields.useDelayTime(model),
+ ...Fields.useTimeoutAlarm(model),
+ ...Fields.useEmrServerless(model),
+ Fields.usePreTasks()
+ ] as IJsonItem[],
+ model
+ }
+}
diff --git a/dolphinscheduler-ui/src/views/projects/task/constants/task-type.ts b/dolphinscheduler-ui/src/views/projects/task/constants/task-type.ts
index ce0db268f4f8..d0eb36b277a8 100644
--- a/dolphinscheduler-ui/src/views/projects/task/constants/task-type.ts
+++ b/dolphinscheduler-ui/src/views/projects/task/constants/task-type.ts
@@ -32,6 +32,7 @@ export type TaskType =
| 'SWITCH'
| 'SEATUNNEL'
| 'EMR'
+ | 'EMR_SERVERLESS'
| 'ZEPPELIN'
| 'K8S'
| 'JUPYTER'
@@ -114,6 +115,10 @@ export const TASK_TYPES_MAP = {
alias: 'AmazonEMR',
helperLinkDisable: true
},
+ EMR_SERVERLESS: {
+ alias: 'AmazonEMRServerless',
+ helperLinkDisable: true
+ },
ZEPPELIN: {
alias: 'ZEPPELIN',
helperLinkDisable: true
From db7f4c76321cef035fefd10647fd0e06fee78441 Mon Sep 17 00:00:00 2001
From: norrishuang <12380647@qq.com>
Date: Sat, 14 Mar 2026 08:37:55 +0000
Subject: [PATCH 02/10] test: add unit tests for EMR Serverless task plugin
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit
11 test cases covering:
- Submit → track → success/failed/cancelled lifecycle
- Full state transition (SUBMITTED→PENDING→SCHEDULED→RUNNING→SUCCESS)
- Submit error handling (SDK exception)
- GetJobRun returns null
- Cancel application (with and without jobRunId)
- Failover recovery via appIds
- Parameter validation (checkParameters)
- Invalid JSON handling
---
.../emrserverless/EmrServerlessTaskTest.java | 288 ++++++++++++++++++
.../emrserverless/StartJobRunRequest.json | 16 +
2 files changed, 304 insertions(+)
create mode 100644 dolphinscheduler-task-plugin/dolphinscheduler-task-emr-serverless/src/test/java/org/apache/dolphinscheduler/plugin/task/emrserverless/EmrServerlessTaskTest.java
create mode 100644 dolphinscheduler-task-plugin/dolphinscheduler-task-emr-serverless/src/test/resources/org/apache/dolphinscheduler/plugin/task/emrserverless/StartJobRunRequest.json
diff --git a/dolphinscheduler-task-plugin/dolphinscheduler-task-emr-serverless/src/test/java/org/apache/dolphinscheduler/plugin/task/emrserverless/EmrServerlessTaskTest.java b/dolphinscheduler-task-plugin/dolphinscheduler-task-emr-serverless/src/test/java/org/apache/dolphinscheduler/plugin/task/emrserverless/EmrServerlessTaskTest.java
new file mode 100644
index 000000000000..eee1b77c13e6
--- /dev/null
+++ b/dolphinscheduler-task-plugin/dolphinscheduler-task-emr-serverless/src/test/java/org/apache/dolphinscheduler/plugin/task/emrserverless/EmrServerlessTaskTest.java
@@ -0,0 +1,288 @@
+/*
+ * 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.plugin.task.emrserverless;
+
+import static org.apache.dolphinscheduler.plugin.task.api.TaskConstants.EXIT_CODE_FAILURE;
+import static org.apache.dolphinscheduler.plugin.task.api.TaskConstants.EXIT_CODE_KILL;
+import static org.apache.dolphinscheduler.plugin.task.api.TaskConstants.EXIT_CODE_SUCCESS;
+import static org.mockito.ArgumentMatchers.any;
+
+import org.apache.dolphinscheduler.common.utils.JSONUtils;
+import org.apache.dolphinscheduler.plugin.task.api.TaskCallBack;
+import org.apache.dolphinscheduler.plugin.task.api.TaskException;
+import org.apache.dolphinscheduler.plugin.task.api.TaskExecutionContext;
+import org.apache.dolphinscheduler.plugin.task.api.model.ApplicationInfo;
+
+import org.apache.commons.io.IOUtils;
+
+import java.io.InputStream;
+import java.nio.charset.StandardCharsets;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.Mockito;
+import org.mockito.junit.jupiter.MockitoExtension;
+
+import com.amazonaws.services.emrserverless.AWSEMRServerless;
+import com.amazonaws.services.emrserverless.model.AWSEMRServerlessException;
+import com.amazonaws.services.emrserverless.model.CancelJobRunResult;
+import com.amazonaws.services.emrserverless.model.GetJobRunResult;
+import com.amazonaws.services.emrserverless.model.JobRun;
+import com.amazonaws.services.emrserverless.model.StartJobRunResult;
+
+@ExtendWith(MockitoExtension.class)
+public class EmrServerlessTaskTest {
+
+ private static final String APPLICATION_ID = "00abcdefgh123456";
+ private static final String JOB_RUN_ID = "00abcdefgh123456-jobrun-001";
+ private static final String EXECUTION_ROLE_ARN = "arn:aws:iam::123456789012:role/EMRServerlessRole";
+
+ private EmrServerlessTask emrServerlessTask;
+ private AWSEMRServerless emrServerlessClient;
+
+ private final TaskCallBack taskCallBack = new TaskCallBack() {
+
+ @Override
+ public void updateRemoteApplicationInfo(int taskInstanceId, ApplicationInfo applicationInfo) {
+ }
+
+ @Override
+ public void updateTaskInstanceInfo(int taskInstanceId) {
+ }
+ };
+
+ @BeforeEach
+ public void before() throws Exception {
+ String taskParams = buildEmrServerlessTaskParameters();
+ TaskExecutionContext taskExecutionContext = Mockito.mock(TaskExecutionContext.class);
+ Mockito.when(taskExecutionContext.getTaskParams()).thenReturn(taskParams);
+ Mockito.lenient().when(taskExecutionContext.getTaskName()).thenReturn("test-emr-serverless-task");
+ Mockito.lenient().when(taskExecutionContext.getTaskInstanceId()).thenReturn(1);
+
+ emrServerlessTask = Mockito.spy(new EmrServerlessTask(taskExecutionContext));
+
+ // mock emrServerlessClient
+ emrServerlessClient = Mockito.mock(AWSEMRServerless.class);
+
+ // mock startJobRun
+ StartJobRunResult startJobRunResult = Mockito.mock(StartJobRunResult.class);
+ Mockito.lenient().when(emrServerlessClient.startJobRun(any())).thenReturn(startJobRunResult);
+ Mockito.lenient().when(startJobRunResult.getJobRunId()).thenReturn(JOB_RUN_ID);
+
+ // inject mock client
+ Mockito.doReturn(emrServerlessClient).when(emrServerlessTask).createEmrServerlessClient();
+
+ emrServerlessTask.init();
+ }
+
+ @Test
+ public void testHandleSuccess() throws Exception {
+ // Job goes: SUBMITTED -> RUNNING -> SUCCESS
+ mockJobRunStates("SUBMITTED", "RUNNING", "SUCCESS");
+
+ emrServerlessTask.handle(taskCallBack);
+ Assertions.assertEquals(EXIT_CODE_SUCCESS, emrServerlessTask.getExitStatusCode());
+ }
+
+ @Test
+ public void testHandleFailed() throws Exception {
+ // Job goes: SUBMITTED -> RUNNING -> FAILED
+ mockJobRunStates("SUBMITTED", "RUNNING", "FAILED");
+
+ emrServerlessTask.handle(taskCallBack);
+ Assertions.assertEquals(EXIT_CODE_FAILURE, emrServerlessTask.getExitStatusCode());
+ }
+
+ @Test
+ public void testHandleCancelled() throws Exception {
+ // Job goes: SUBMITTED -> RUNNING -> CANCELLED
+ mockJobRunStates("SUBMITTED", "RUNNING", "CANCELLED");
+
+ emrServerlessTask.handle(taskCallBack);
+ Assertions.assertEquals(EXIT_CODE_KILL, emrServerlessTask.getExitStatusCode());
+ }
+
+ @Test
+ public void testHandleFullLifecycle() throws Exception {
+ // Job goes through all intermediate states: SUBMITTED -> PENDING -> SCHEDULED -> RUNNING -> SUCCESS
+ mockJobRunStates("SUBMITTED", "PENDING", "SCHEDULED", "RUNNING", "SUCCESS");
+
+ emrServerlessTask.handle(taskCallBack);
+ Assertions.assertEquals(EXIT_CODE_SUCCESS, emrServerlessTask.getExitStatusCode());
+ }
+
+ @Test
+ public void testSubmitError() throws Exception {
+ Mockito.when(emrServerlessClient.startJobRun(any()))
+ .thenThrow(new AWSEMRServerlessException("Access denied"));
+
+ Assertions.assertThrows(TaskException.class, () -> {
+ emrServerlessTask.handle(taskCallBack);
+ });
+ }
+
+ @Test
+ public void testGetJobRunReturnsNull() throws Exception {
+ // First call for submit (need a valid startJobRun response)
+ StartJobRunResult startResult = Mockito.mock(StartJobRunResult.class);
+ Mockito.when(emrServerlessClient.startJobRun(any())).thenReturn(startResult);
+ Mockito.when(startResult.getJobRunId()).thenReturn(JOB_RUN_ID);
+
+ // getJobRun returns null
+ Mockito.when(emrServerlessClient.getJobRun(any())).thenReturn(null);
+
+ emrServerlessTask.handle(taskCallBack);
+ Assertions.assertEquals(EXIT_CODE_FAILURE, emrServerlessTask.getExitStatusCode());
+ }
+
+ @Test
+ public void testCancelApplication() throws Exception {
+ // Submit first so we have a jobRunId
+ mockJobRunStates("SUBMITTED", "RUNNING", "SUCCESS");
+ emrServerlessTask.handle(taskCallBack);
+
+ // Now test cancel
+ CancelJobRunResult cancelResult = Mockito.mock(CancelJobRunResult.class);
+ Mockito.when(emrServerlessClient.cancelJobRun(any())).thenReturn(cancelResult);
+ emrServerlessTask.cancelApplication();
+
+ Mockito.verify(emrServerlessClient).cancelJobRun(any());
+ }
+
+ @Test
+ public void testCancelWithEmptyJobRunId() throws Exception {
+ // Don't submit, so jobRunId is empty — cancel should be a no-op
+ emrServerlessTask.cancelApplication();
+ Mockito.verify(emrServerlessClient, Mockito.never()).cancelJobRun(any());
+ }
+
+ @Test
+ public void testFailoverRecovery() throws Exception {
+ // Simulate failover: submit the job first
+ mockJobRunStates("SUBMITTED", "RUNNING", "SUCCESS");
+ emrServerlessTask.handle(taskCallBack);
+
+ // Now create a new task instance simulating failover
+ String taskParams = buildEmrServerlessTaskParameters();
+ TaskExecutionContext failoverContext = Mockito.mock(TaskExecutionContext.class);
+ Mockito.when(failoverContext.getTaskParams()).thenReturn(taskParams);
+ Mockito.lenient().when(failoverContext.getTaskName()).thenReturn("test-emr-serverless-failover");
+ Mockito.lenient().when(failoverContext.getTaskInstanceId()).thenReturn(2);
+ // Simulate that appIds was persisted from previous run
+ Mockito.when(failoverContext.getAppIds()).thenReturn(APPLICATION_ID + ":" + JOB_RUN_ID);
+
+ EmrServerlessTask failoverTask = Mockito.spy(new EmrServerlessTask(failoverContext));
+ Mockito.doReturn(emrServerlessClient).when(failoverTask).createEmrServerlessClient();
+
+ // Reset getJobRun mock for the recovery path — job already completed
+ mockJobRunStates("SUCCESS");
+
+ failoverTask.init();
+ failoverTask.handle(taskCallBack);
+
+ // Should recover and succeed without calling startJobRun again
+ Assertions.assertEquals(EXIT_CODE_SUCCESS, failoverTask.getExitStatusCode());
+ }
+
+ @Test
+ public void testParametersCheck() {
+ EmrServerlessParameters params = new EmrServerlessParameters();
+
+ // All empty — should fail
+ Assertions.assertFalse(params.checkParameters());
+
+ // Only applicationId — should fail
+ params.setApplicationId(APPLICATION_ID);
+ Assertions.assertFalse(params.checkParameters());
+
+ // applicationId + executionRoleArn — should fail (no JSON)
+ params.setExecutionRoleArn(EXECUTION_ROLE_ARN);
+ Assertions.assertFalse(params.checkParameters());
+
+ // All three — should pass
+ params.setStartJobRunRequestJson("{}");
+ Assertions.assertTrue(params.checkParameters());
+ }
+
+ @Test
+ public void testInvalidJson() throws Exception {
+ // Build params with invalid JSON
+ EmrServerlessParameters params = new EmrServerlessParameters();
+ params.setApplicationId(APPLICATION_ID);
+ params.setExecutionRoleArn(EXECUTION_ROLE_ARN);
+ params.setStartJobRunRequestJson("{invalid json!!!}");
+
+ TaskExecutionContext ctx = Mockito.mock(TaskExecutionContext.class);
+ Mockito.when(ctx.getTaskParams()).thenReturn(JSONUtils.toJsonString(params));
+ Mockito.lenient().when(ctx.getTaskName()).thenReturn("test-bad-json");
+ Mockito.lenient().when(ctx.getTaskInstanceId()).thenReturn(99);
+
+ EmrServerlessTask badJsonTask = Mockito.spy(new EmrServerlessTask(ctx));
+ Mockito.doReturn(emrServerlessClient).when(badJsonTask).createEmrServerlessClient();
+
+ badJsonTask.init();
+ Assertions.assertThrows(TaskException.class, () -> {
+ badJsonTask.handle(taskCallBack);
+ });
+ }
+
+ // --- Helper methods ---
+
+ private void mockJobRunStates(String... states) {
+ if (states.length == 0) {
+ return;
+ }
+
+ GetJobRunResult[] results = new GetJobRunResult[states.length];
+ for (int i = 0; i < states.length; i++) {
+ GetJobRunResult result = Mockito.mock(GetJobRunResult.class);
+ JobRun jobRun = Mockito.mock(JobRun.class);
+ Mockito.when(result.getJobRun()).thenReturn(jobRun);
+ Mockito.when(jobRun.getState()).thenReturn(states[i]);
+ results[i] = result;
+ }
+
+ if (results.length == 1) {
+ Mockito.when(emrServerlessClient.getJobRun(any())).thenReturn(results[0]);
+ } else {
+ GetJobRunResult first = results[0];
+ GetJobRunResult[] rest = new GetJobRunResult[results.length - 1];
+ System.arraycopy(results, 1, rest, 0, rest.length);
+ Mockito.when(emrServerlessClient.getJobRun(any())).thenReturn(first, rest);
+ }
+ }
+
+ private String buildEmrServerlessTaskParameters() {
+ EmrServerlessParameters params = new EmrServerlessParameters();
+ params.setApplicationId(APPLICATION_ID);
+ params.setExecutionRoleArn(EXECUTION_ROLE_ARN);
+
+ String startJobRunRequestJson;
+ try (InputStream is = this.getClass().getResourceAsStream("StartJobRunRequest.json")) {
+ assert is != null;
+ startJobRunRequestJson = IOUtils.toString(is, StandardCharsets.UTF_8);
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ params.setStartJobRunRequestJson(startJobRunRequestJson);
+
+ return JSONUtils.toJsonString(params);
+ }
+}
diff --git a/dolphinscheduler-task-plugin/dolphinscheduler-task-emr-serverless/src/test/resources/org/apache/dolphinscheduler/plugin/task/emrserverless/StartJobRunRequest.json b/dolphinscheduler-task-plugin/dolphinscheduler-task-emr-serverless/src/test/resources/org/apache/dolphinscheduler/plugin/task/emrserverless/StartJobRunRequest.json
new file mode 100644
index 000000000000..36386900a83a
--- /dev/null
+++ b/dolphinscheduler-task-plugin/dolphinscheduler-task-emr-serverless/src/test/resources/org/apache/dolphinscheduler/plugin/task/emrserverless/StartJobRunRequest.json
@@ -0,0 +1,16 @@
+{
+ "JobDriver": {
+ "SparkSubmit": {
+ "EntryPoint": "s3://my-bucket/scripts/spark-etl.py",
+ "EntryPointArguments": ["--input", "s3://data/raw", "--output", "s3://data/processed"],
+ "SparkSubmitParameters": "--conf spark.executor.cores=4 --conf spark.executor.memory=16g --conf spark.executor.instances=10"
+ }
+ },
+ "ConfigurationOverrides": {
+ "MonitoringConfiguration": {
+ "S3MonitoringConfiguration": {
+ "LogUri": "s3://my-bucket/emr-serverless-logs/"
+ }
+ }
+ }
+}
From 1bd61c7981b064c9335fe6edcc6404860d2ad10c Mon Sep 17 00:00:00 2001
From: norrishuang <12380647@qq.com>
Date: Sat, 14 Mar 2026 10:41:44 +0000
Subject: [PATCH 03/10] fix: add shade plugin and TS type defs for EMR
Serverless task
- Add maven-shade-plugin to emr-serverless pom.xml so shade jar is
included in dist assembly
- Add applicationId, executionRoleArn, startJobRunRequestJson fields
to ITaskParams in types.ts to fix TypeScript build
---
.../pom.xml | 17 +++++++++++++++++
.../projects/task/components/node/types.ts | 3 +++
2 files changed, 20 insertions(+)
diff --git a/dolphinscheduler-task-plugin/dolphinscheduler-task-emr-serverless/pom.xml b/dolphinscheduler-task-plugin/dolphinscheduler-task-emr-serverless/pom.xml
index e1e046eda113..81a98a92cf25 100644
--- a/dolphinscheduler-task-plugin/dolphinscheduler-task-emr-serverless/pom.xml
+++ b/dolphinscheduler-task-plugin/dolphinscheduler-task-emr-serverless/pom.xml
@@ -57,4 +57,21 @@