[Improvement-17724][TaskPlugin] EmrTask resource leak repair

dolphinscheduler-task-plugin/dolphinscheduler-task-emr/src/main/java/org/apache/dolphinscheduler/plugin/task/emr/AbstractEmrTask.java

@@ -27,9 +27,14 @@
 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.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.dolphinscheduler.plugin.task.api.parameters.AbstractParameters;
 
+import org.apache.commons.lang3.StringUtils;
+
 import java.util.Map;
 
 import lombok.extern.slf4j.Slf4j;
@@ -86,6 +91,42 @@ public void init() {
         emrClient = createEmrClient();
     }
 
+    /**
+     * If appIds is empty, submit a new remote application; otherwise, just track application status.
+     *
+     * @param taskCallBack
+     * @throws TaskException
+     */
+    @Override
+    public void handle(TaskCallBack taskCallBack) throws TaskException {
+        try {
+            // if appIds is not empty, just track application status, avoid resubmitting remote task
+            if (StringUtils.isNotEmpty(taskRequest.getAppIds())) {
+                setAppIds(taskRequest.getAppIds());
+                trackApplicationStatus();
+                return;
+            }
+
+            // submit a remote application
+            submitApplication();
+
+            if (StringUtils.isNotEmpty(getAppIds())) {
+                taskRequest.setAppIds(getAppIds());
+                // callback to update remote application info
+                taskCallBack.updateRemoteApplicationInfo(taskRequest.getTaskInstanceId(),
+                        new ApplicationInfo(getAppIds()));
+            }
+
+            // keep tracking application status
+            trackApplicationStatus();
+        } finally {
+            // shutdown emrclient
+            if (emrClient != null) {
+                emrClient.shutdown();
+            }
+        }
+    }
+
     @Override
     public AbstractParameters getParameters() {
         return emrParameters;

dolphinscheduler-task-plugin/dolphinscheduler-task-emr/src/main/java/org/apache/dolphinscheduler/plugin/task/emr/EmrAddStepsTask.java

@@ -51,7 +51,7 @@
 @Slf4j
 public class EmrAddStepsTask extends AbstractEmrTask {
 
-    private String stepId;
+    String stepId;
 
     private final HashSet<String> waitingStateSet = Sets.newHashSet(
             StepState.PENDING.toString(),
@@ -189,7 +189,8 @@ private StepStatus getStepStatus() {
     public void cancelApplication() throws TaskException {
         log.info("trying cancel emr step, taskId:{}, clusterId:{}, stepId:{}",
                 this.taskExecutionContext.getTaskInstanceId(), clusterId, stepId);
-        CancelStepsRequest cancelStepsRequest = new CancelStepsRequest().withClusterId(clusterId).withStepIds(stepId);
+        CancelStepsRequest cancelStepsRequest =
+                new CancelStepsRequest().withClusterId(clusterId).withStepIds(stepId);
         CancelStepsResult cancelStepsResult = emrClient.cancelSteps(cancelStepsRequest);
 
         if (cancelStepsResult == null) {

dolphinscheduler-task-plugin/dolphinscheduler-task-emr/src/main/java/org/apache/dolphinscheduler/plugin/task/emr/EmrJobFlowTask.java

@@ -180,9 +180,11 @@ private ClusterStatus getClusterStatus() {
 
     @Override
     public void cancelApplication() throws TaskException {
-        log.info("trying terminate job flow, taskId:{}, clusterId:{}", this.taskExecutionContext.getTaskInstanceId(),
+        log.info("trying terminate job flow, taskId:{}, clusterId:{}",
+                this.taskExecutionContext.getTaskInstanceId(),
                 clusterId);
-        TerminateJobFlowsRequest terminateJobFlowsRequest = new TerminateJobFlowsRequest().withJobFlowIds(clusterId);
+        TerminateJobFlowsRequest terminateJobFlowsRequest =
+                new TerminateJobFlowsRequest().withJobFlowIds(clusterId);
         TerminateJobFlowsResult terminateJobFlowsResult = emrClient.terminateJobFlows(terminateJobFlowsRequest);
         log.info("the result of terminate job flow is:{}", terminateJobFlowsResult);
     }