The YarnClientTrait add clusterClient close

apache · Jul 28, 2023 · 8e31e06 · 8e31e06
1 parent af0df56
commit 8e31e06
Showing 1 changed file with 6 additions and 6 deletions.
diff --git a/...client-core/src/main/scala/org/apache/streampark/flink/client/trait/YarnClientTrait.scala b/...client-core/src/main/scala/org/apache/streampark/flink/client/trait/YarnClientTrait.scala
@@ -33,8 +33,6 @@ import org.apache.hadoop.yarn.api.records.ApplicationId
 import java.lang.{Boolean => JavaBool}
 import java.lang.reflect.Method
 
-import scala.util.Try
-
 /** yarn application mode submit */
 trait YarnClientTrait extends FlinkClientTrait {
 
@@ -54,14 +52,16 @@ trait YarnClientTrait extends FlinkClientTrait {
       val clusterDescriptor = clusterClientFactory.createClusterDescriptor(flinkConf)
       clusterDescriptor.retrieve(applicationId).getClusterClient
     }
-    Try {
+    try {
       actionFunc(jobID, clusterClient)
-    }.recover {
-      case e =>
+    } catch {
+      case e: Exception =>
         throw new FlinkException(
           s"[StreamPark] Do ${request.getClass.getSimpleName} for the job ${request.jobId} failed. " +
             s"detail: ${Utils.stringifyException(e)}");
-    }.get
+    } finally {
+      if (clusterClient != null) clusterClient.close()
+    }
   }
 
   override def doTriggerSavepoint(