[Improvement-17723][TaskPlugin] DmsTask resource leak repair

dolphinscheduler-task-plugin/dolphinscheduler-task-dms/src/main/java/org/apache/dolphinscheduler/plugin/task/dms/DmsHook.java

@@ -141,10 +141,19 @@ public void stopReplicationTask() {
         if (replicationTaskArn == null) {
             return;
         }
-        StopReplicationTaskRequest request = new StopReplicationTaskRequest()
-                .withReplicationTaskArn(replicationTaskArn);
-        client.stopReplicationTask(request);
-        awaitReplicationTaskStatus(STATUS.STOPPED);
+        try {
+            StopReplicationTaskRequest request = new StopReplicationTaskRequest()
+                    .withReplicationTaskArn(replicationTaskArn);
+            client.stopReplicationTask(request);
+            awaitReplicationTaskStatus(STATUS.STOPPED);
+        } catch (Exception e) {
+            log.error("stopReplicationTask error: ", e);
+        } finally {
+            if (client != null) {
+                // shutdown client
+                client.shutdown();
+            }
+        }
     }
 
     public Boolean deleteReplicationTask() {
@@ -268,8 +277,8 @@ public String replaceFileParameters(String parameter) throws IOException {
         }
         if (parameter.startsWith("file://")) {
             String filePath = parameter.substring(7);
-            try {
-                return IOUtils.toString(new FileInputStream(filePath), StandardCharsets.UTF_8);
+            try (FileInputStream fis = new FileInputStream(filePath)) {
+                return IOUtils.toString(fis, StandardCharsets.UTF_8);
             } catch (IOException e) {
                 throw new IOException("Error reading file: " + filePath, e);
             }

dolphinscheduler-task-plugin/dolphinscheduler-task-dms/src/main/java/org/apache/dolphinscheduler/plugin/task/dms/DmsTask.java

@@ -24,9 +24,11 @@
 
 import org.apache.dolphinscheduler.common.utils.JSONUtils;
 import org.apache.dolphinscheduler.plugin.task.api.AbstractRemoteTask;
+import org.apache.dolphinscheduler.plugin.task.api.TaskCallBack;
 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.model.ApplicationInfo;
 import org.apache.dolphinscheduler.plugin.task.api.utils.ParameterUtils;
 
 import org.apache.commons.beanutils.BeanUtils;
@@ -78,6 +80,42 @@ public void init() throws TaskException {
         initDmsHook();
     }
 
+    /**
+     * 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 dmsHook client
+            if (dmsHook.getClient() != null) {
+                dmsHook.getClient().shutdown();
+            }
+        }
+    }
+
     @Override
     public List<String> getApplicationIds() throws TaskException {
         return Collections.emptyList();

dolphinscheduler-task-plugin/dolphinscheduler-task-dms/src/test/java/org/apache/dolphinscheduler/plugin/task/dms/DmsHookTest.java

@@ -23,6 +23,11 @@
 import static org.mockito.Mockito.spy;
 import static org.mockito.Mockito.when;
 
+import org.apache.commons.io.FileUtils;
+
+import java.io.File;
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
 import java.time.Duration;
 import java.util.Arrays;
 
@@ -238,4 +243,63 @@ public void testAwaitReplicationTaskStatus() {
             }
         });
     }
+
+    @Test
+    public void testReplaceFileParametersWithNull() throws IOException {
+        Assertions.assertTimeout(Duration.ofMillis(60000), () -> {
+            try (MockedStatic<DmsHook> mockHook = Mockito.mockStatic(DmsHook.class)) {
+                mockHook.when(DmsHook::createClient).thenReturn(client);
+                DmsHook dmsHook = spy(new DmsHook());
+                String parameter = null;
+                String result = dmsHook.replaceFileParameters(parameter);
+                Assertions.assertNull(result);
+            }
+        });
+    }
+
+    @Test
+    public void testReplaceFileParametersWithNormalString() throws IOException {
+        Assertions.assertTimeout(Duration.ofMillis(60000), () -> {
+            try (MockedStatic<DmsHook> mockHook = Mockito.mockStatic(DmsHook.class)) {
+                mockHook.when(DmsHook::createClient).thenReturn(client);
+                DmsHook dmsHook = spy(new DmsHook());
+                String parameter = "normal string";
+                String result = dmsHook.replaceFileParameters(parameter);
+                Assertions.assertEquals(parameter, result);
+            }
+        });
+    }
+
+    @Test
+    public void testReplaceFileParametersWithExistingFile() throws IOException {
+        Assertions.assertTimeout(Duration.ofMillis(60000), () -> {
+            File tempFile = new File("tempFile.txt");
+            try (MockedStatic<DmsHook> mockHook = Mockito.mockStatic(DmsHook.class)) {
+                mockHook.when(DmsHook::createClient).thenReturn(client);
+                DmsHook dmsHook = spy(new DmsHook());
+                String fileContent = "content of the file";
+                FileUtils.writeStringToFile(tempFile, fileContent, StandardCharsets.UTF_8);
+                String parameter = "file://" + tempFile.getAbsolutePath();
+                String result = dmsHook.replaceFileParameters(parameter);
+                Assertions.assertEquals(fileContent, result);
+            } finally {
+                tempFile.delete();
+            }
+        });
+    }
+
+    @Test
+    public void testReplaceFileParametersWithNonexistentFile() {
+        Assertions.assertTimeout(Duration.ofMillis(60000), () -> {
+            try (MockedStatic<DmsHook> mockHook = Mockito.mockStatic(DmsHook.class)) {
+                mockHook.when(DmsHook::createClient).thenReturn(client);
+                DmsHook dmsHook = spy(new DmsHook());
+                String parameter = "file://nonexistentfile.txt";
+                Assertions.assertThrows(IOException.class, () -> {
+                    dmsHook.replaceFileParameters(parameter);
+                });
+            }
+        });
+    }
+
 }