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

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

@@ -133,6 +133,8 @@ public Boolean checkFinishedReplicationTask() {
         log.info("checkFinishedReplicationTask ......");
         awaitReplicationTaskStatus(STATUS.STOPPED);
         String stopReason = describeReplicationTasks().getStopReason();
+        // shutdown client
+        client.shutdown();
         return stopReason.endsWith(STATUS.FINISH_END_TOKEN);
     }
 
@@ -145,6 +147,8 @@ public void stopReplicationTask() {
                 .withReplicationTaskArn(replicationTaskArn);
         client.stopReplicationTask(request);
         awaitReplicationTaskStatus(STATUS.STOPPED);
+        // shutdown client
+        client.shutdown();
     }
 
     public Boolean deleteReplicationTask() {
@@ -158,6 +162,8 @@ public Boolean deleteReplicationTask() {
         } catch (ResourceNotFoundException e) {
             isDeleteSuccessfully = true;
         }
+        // shutdown client
+        client.shutdown();
         return isDeleteSuccessfully;
     }
 
@@ -268,8 +274,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/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 @@
             }
         });
     }
+
+    @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);
+                });
+            }
+        });
+    }
+
 }