[Improve] YarnClusterDescriptorWrapper minor improvement

apache · Aug 29, 2024 · 9058c9e · 9058c9e
1 parent 33577e0
commit 9058c9e
Show file tree

Hide file tree

Showing 12 changed files with 16 additions and 21 deletions.
diff --git a/...t-core/src/main/scala/org/apache/streampark/flink/client/impl/YarnApplicationClient.scala b/...t-core/src/main/scala/org/apache/streampark/flink/client/impl/YarnApplicationClient.scala
@@ -27,11 +27,8 @@ import org.apache.streampark.flink.packer.pipeline.ShadedBuildResponse
 import org.apache.flink.client.deployment.application.ApplicationConfiguration
 import org.apache.flink.configuration._
 import org.apache.flink.runtime.security.{SecurityConfiguration, SecurityUtils}
-import org.apache.flink.runtime.util.HadoopUtils
 import org.apache.flink.yarn.YarnClusterDescriptor
 import org.apache.flink.yarn.configuration.{YarnConfigOptions, YarnDeploymentTarget}
-import org.apache.hadoop.security.UserGroupInformation
-import org.apache.hadoop.yarn.api.records.ApplicationId
 
 import java.util.Collections
 import java.util.concurrent.Callable

diff --git a/...client-core/src/main/scala/org/apache/streampark/flink/client/impl/YarnPerJobClient.scala b/...client-core/src/main/scala/org/apache/streampark/flink/client/impl/YarnPerJobClient.scala
@@ -24,7 +24,7 @@ import org.apache.streampark.flink.util.FlinkUtils
 
 import org.apache.flink.client.program.PackagedProgram
 import org.apache.flink.configuration.{Configuration, DeploymentOptions}
-import org.apache.flink.yarn.{YarnClusterClientFactory, YarnClusterDescriptor}
+import org.apache.flink.yarn.YarnClusterDescriptor
 import org.apache.flink.yarn.configuration.YarnDeploymentTarget
 import org.apache.flink.yarn.entrypoint.YarnJobClusterEntrypoint
 import org.apache.hadoop.fs.{Path => HadoopPath}

diff --git a/...lient-core/src/main/scala/org/apache/streampark/flink/client/impl/YarnSessionClient.scala b/...lient-core/src/main/scala/org/apache/streampark/flink/client/impl/YarnSessionClient.scala
@@ -25,10 +25,8 @@ import org.apache.commons.lang3.StringUtils
 import org.apache.flink.client.deployment.ClusterSpecification
 import org.apache.flink.client.program.ClusterClient
 import org.apache.flink.configuration._
-import org.apache.flink.runtime.util.HadoopUtils
 import org.apache.flink.yarn.YarnClusterDescriptor
 import org.apache.flink.yarn.configuration.{YarnConfigOptions, YarnDeploymentTarget}
-import org.apache.hadoop.security.UserGroupInformation
 import org.apache.hadoop.yarn.api.records.{ApplicationId, FinalApplicationStatus}
 import org.apache.hadoop.yarn.util.ConverterUtils
 

diff --git a/...k-1.12/src/main/scala/org/apache/streampark/flink/core/YarnClusterDescriptorWrapper.scala b/...k-1.12/src/main/scala/org/apache/streampark/flink/core/YarnClusterDescriptorWrapper.scala
@@ -19,4 +19,4 @@ package org.apache.streampark.flink.core
 import org.apache.flink.yarn.YarnClusterDescriptor
 
 class YarnClusterDescriptorWrapper(yarnClusterDescriptor: YarnClusterDescriptor)
-  extends YarnClusterDescriptorTrait(yarnClusterDescriptor) {}
+  extends YarnClusterDescriptorTrait(yarnClusterDescriptor)
diff --git a/...k-1.13/src/main/scala/org/apache/streampark/flink/core/YarnClusterDescriptorWrapper.scala b/...k-1.13/src/main/scala/org/apache/streampark/flink/core/YarnClusterDescriptorWrapper.scala
@@ -19,4 +19,4 @@ package org.apache.streampark.flink.core
 import org.apache.flink.yarn.YarnClusterDescriptor
 
 class YarnClusterDescriptorWrapper(yarnClusterDescriptor: YarnClusterDescriptor)
-  extends YarnClusterDescriptorTrait(yarnClusterDescriptor) {}
+  extends YarnClusterDescriptorTrait(yarnClusterDescriptor)
diff --git a/...k-1.14/src/main/scala/org/apache/streampark/flink/core/YarnClusterDescriptorWrapper.scala b/...k-1.14/src/main/scala/org/apache/streampark/flink/core/YarnClusterDescriptorWrapper.scala
@@ -19,4 +19,4 @@ package org.apache.streampark.flink.core
 import org.apache.flink.yarn.YarnClusterDescriptor
 
 class YarnClusterDescriptorWrapper(yarnClusterDescriptor: YarnClusterDescriptor)
-  extends YarnClusterDescriptorTrait(yarnClusterDescriptor) {}
+  extends YarnClusterDescriptorTrait(yarnClusterDescriptor)
diff --git a/...k-1.15/src/main/scala/org/apache/streampark/flink/core/YarnClusterDescriptorWrapper.scala b/...k-1.15/src/main/scala/org/apache/streampark/flink/core/YarnClusterDescriptorWrapper.scala
@@ -20,4 +20,4 @@ package org.apache.streampark.flink.core
 import org.apache.flink.yarn.YarnClusterDescriptor
 
 class YarnClusterDescriptorWrapper(yarnClusterDescriptor: YarnClusterDescriptor)
-  extends YarnClusterDescriptorTrait(yarnClusterDescriptor) {}
+  extends YarnClusterDescriptorTrait(yarnClusterDescriptor)
diff --git a/...k-1.16/src/main/scala/org/apache/streampark/flink/core/YarnClusterDescriptorWrapper.scala b/...k-1.16/src/main/scala/org/apache/streampark/flink/core/YarnClusterDescriptorWrapper.scala
@@ -20,4 +20,4 @@ package org.apache.streampark.flink.core
 import org.apache.flink.yarn.YarnClusterDescriptor
 
 class YarnClusterDescriptorWrapper(yarnClusterDescriptor: YarnClusterDescriptor)
-  extends YarnClusterDescriptorTrait(yarnClusterDescriptor) {}
+  extends YarnClusterDescriptorTrait(yarnClusterDescriptor)
diff --git a/...k-1.17/src/main/scala/org/apache/streampark/flink/core/YarnClusterDescriptorWrapper.scala b/...k-1.17/src/main/scala/org/apache/streampark/flink/core/YarnClusterDescriptorWrapper.scala
@@ -20,4 +20,4 @@ package org.apache.streampark.flink.core
 import org.apache.flink.yarn.YarnClusterDescriptor
 
 class YarnClusterDescriptorWrapper(yarnClusterDescriptor: YarnClusterDescriptor)
-  extends YarnClusterDescriptorTrait(yarnClusterDescriptor) {}
+  extends YarnClusterDescriptorTrait(yarnClusterDescriptor)
diff --git a/...k-1.18/src/main/scala/org/apache/streampark/flink/core/YarnClusterDescriptorWrapper.scala b/...k-1.18/src/main/scala/org/apache/streampark/flink/core/YarnClusterDescriptorWrapper.scala
@@ -20,4 +20,4 @@ package org.apache.streampark.flink.core
 import org.apache.flink.yarn.YarnClusterDescriptor
 
 class YarnClusterDescriptorWrapper(yarnClusterDescriptor: YarnClusterDescriptor)
-  extends YarnClusterDescriptorTrait(yarnClusterDescriptor) {}
+  extends YarnClusterDescriptorTrait(yarnClusterDescriptor)
diff --git a/...k-1.19/src/main/scala/org/apache/streampark/flink/core/YarnClusterDescriptorWrapper.scala b/...k-1.19/src/main/scala/org/apache/streampark/flink/core/YarnClusterDescriptorWrapper.scala
@@ -18,7 +18,6 @@
 package org.apache.streampark.flink.core
 
 import collection.JavaConversions._
-import com.google.common.collect.Lists
 import org.apache.flink.yarn.YarnClusterDescriptor
 import org.apache.hadoop.fs.Path
 
@@ -28,8 +27,9 @@ import java.util
 class YarnClusterDescriptorWrapper(yarnClusterDescriptor: YarnClusterDescriptor)
   extends YarnClusterDescriptorTrait(yarnClusterDescriptor) {
 
-  override def addShipFiles(input: util.List[File]) = {
-    val f = input.map(c => new Path(c.toURI))
-    yarnClusterDescriptor.addShipFiles(Lists.newArrayList(f: _*))
+  override def addShipFiles(input: util.List[File]): Unit = {
+    val list = new util.ArrayList[Path]()
+    input.foreach(x => list.add(new Path(x.toURI)))
+    yarnClusterDescriptor.addShipFiles(list)
   }
 }
diff --git a/...k-1.20/src/main/scala/org/apache/streampark/flink/core/YarnClusterDescriptorWrapper.scala b/...k-1.20/src/main/scala/org/apache/streampark/flink/core/YarnClusterDescriptorWrapper.scala
@@ -18,7 +18,6 @@
 package org.apache.streampark.flink.core
 
 import collection.JavaConversions._
-import com.google.common.collect.Lists
 import org.apache.flink.yarn.YarnClusterDescriptor
 import org.apache.hadoop.fs.Path
 
@@ -28,8 +27,9 @@ import java.util
 class YarnClusterDescriptorWrapper(yarnClusterDescriptor: YarnClusterDescriptor)
   extends YarnClusterDescriptorTrait(yarnClusterDescriptor) {
 
-  override def addShipFiles(input: util.List[File]) = {
-    val f = input.map(c => new Path(c.toURI))
-    yarnClusterDescriptor.addShipFiles(Lists.newArrayList(f: _*))
+  override def addShipFiles(input: util.List[File]): Unit = {
+    val list = new util.ArrayList[Path]()
+    input.foreach(x => list.add(new Path(x.toURI)))
+    yarnClusterDescriptor.addShipFiles(list)
   }
 }