[Feature] Pyflink remote mode support maven dependency (#3311)

* [Feature] Pyflink remote mode support maven dependency
apache · Nov 4, 2023 · 44d2bbf · 44d2bbf
1 parent 34a899d
commit 44d2bbf
Showing 1 changed file with 9 additions and 4 deletions.
diff --git a/...lient-core/src/main/scala/org/apache/streampark/flink/client/trait/FlinkClientTrait.scala b/...lient-core/src/main/scala/org/apache/streampark/flink/client/trait/FlinkClientTrait.scala
@@ -22,7 +22,7 @@ import org.apache.streampark.common.conf.ConfigKeys._
 import org.apache.streampark.common.conf.Workspace
 import org.apache.streampark.common.enums.{ApplicationType, FlinkDevelopmentMode, FlinkExecutionMode, FlinkRestoreMode}
 import org.apache.streampark.common.fs.FsOperator
-import org.apache.streampark.common.util.{DeflaterUtils, ExceptionUtils, FileUtils, Logger, PropertiesUtils, SystemPropertyUtils, Utils}
+import org.apache.streampark.common.util._
 import org.apache.streampark.flink.client.bean._
 import org.apache.streampark.flink.core.FlinkClusterClient
 import org.apache.streampark.flink.core.conf.FlinkRunOption
@@ -42,11 +42,10 @@ import org.apache.flink.runtime.jobgraph.{JobGraph, SavepointConfigOptions}
 import org.apache.flink.util.FlinkException
 import org.apache.flink.util.Preconditions.checkNotNull
 
+import java.io.File
 import java.util
 import java.util.{Collections, List => JavaList, Map => JavaMap}
 
-import scala.annotation.tailrec
-import scala.collection.JavaConverters._
 import scala.collection.convert.ImplicitConversions._
 import scala.collection.mutable
 import scala.collection.mutable.ArrayBuffer
@@ -277,7 +276,13 @@ trait FlinkClientTrait extends Logger {
           throw new RuntimeException(s"$pythonVenv File does not exist")
         }
         // including $app/lib
-        includingPipelineJars(submitRequest, flinkConfig)
+        val localLib: String = s"${Workspace.local.APP_WORKSPACE}/${submitRequest.id}/lib"
+        if (FileUtils.exists(localLib) && FileUtils.directoryNotBlank(localLib)) {
+          val localLibUrl = new File(localLib).listFiles().map(_.toURI.toURL).toList
+          pkgBuilder.setUserClassPaths(
+            Lists.newArrayList(localLibUrl: _*)
+          )
+        }
         flinkConfig
           // python.archives
           .safeSet(PythonOptions.PYTHON_ARCHIVES, pythonVenv)