[FLINK-27084] Fix classloader for per-round operators

yunfengzhou-hub · lindong28 · commit da4bf4b9b946 · 2022-07-01T10:12:16.000+08:00
This closes apache#125.
diff --git a/docs/content/docs/try-flink-ml/build-your-own-project.md b/docs/content/docs/try-flink-ml/build-your-own-project.md
@@ -35,11 +35,7 @@ Learning Model and use it to provide prediction service.
 In order to use Flink ML in a Maven project, add the following dependencies to
 `pom.xml`.
 
-{{< artifact flink-ml-core >}}
-
-{{< artifact flink-ml-iteration >}}
-
-{{< artifact flink-ml-lib >}}
+{{< artifact flink-ml-uber >}}
 
 The example code provided in this document requires additional dependencies on
 the Flink Table API. In order to execute the example code successfully, please
diff --git a/docs/content/docs/try-flink-ml/quick-start.md b/docs/content/docs/try-flink-ml/quick-start.md
@@ -64,16 +64,6 @@ mvn clean package -DskipTests
 cd ./flink-ml-dist/target/flink-ml-*-bin/flink-ml*/
 ```
 
-### Add Flink ML binaries to Flink
-
-You need to copy Flink ML's binary distribution files to Flink's folder for
-proper initialization. Please run the following command from Flink ML's binary
-distribution's folder.
-
-```bash
-cp ./lib/*.jar $FLINK_HOME/lib/
-```
-
 ## Run Flink ML example job
 
 Please start a Flink standalone cluster in your local environment with the
@@ -90,7 +80,7 @@ that the cluster is up and running.
 Then you may submit Flink ML examples to the cluster as follows.
 
 ```
-$FLINK_HOME/bin/flink run -c org.apache.flink.ml.examples.clustering.KMeansExample $FLINK_HOME/lib/flink-ml-examples*.jar
+$FLINK_HOME/bin/flink run -c org.apache.flink.ml.examples.clustering.KMeansExample -C file://`pwd`/lib/flink-ml-uber-2.2-SNAPSHOT.jar -C file://`pwd`/lib/statefun-flink-core-3.2.0.jar lib/flink-ml-examples-2.2-SNAPSHOT.jar
 ```
 
 The command above would submit and execute Flink ML's `KMeansExample` job. There
diff --git a/flink-ml-examples/pom.xml b/flink-ml-examples/pom.xml
@@ -33,72 +33,25 @@ under the License.
     <dependencies>
         <dependency>
             <groupId>org.apache.flink</groupId>
-            <artifactId>flink-ml-core</artifactId>
+            <artifactId>flink-ml-uber</artifactId>
             <version>${project.version}</version>
             <scope>provided</scope>
         </dependency>
 
         <dependency>
             <groupId>org.apache.flink</groupId>
-            <artifactId>flink-ml-iteration</artifactId>
-            <version>${project.version}</version>
-            <scope>provided</scope>
-        </dependency>
-
-        <dependency>
-            <groupId>org.apache.flink</groupId>
-            <artifactId>flink-ml-lib</artifactId>
-            <version>${project.version}</version>
-            <scope>provided</scope>
-        </dependency>
-
-        <dependency>
-            <groupId>org.apache.flink</groupId>
-            <artifactId>flink-clients</artifactId>
-            <version>${flink.version}</version>
-            <scope>provided</scope>
-        </dependency>
-
-        <dependency>
-            <groupId>org.apache.flink</groupId>
-            <artifactId>flink-streaming-java</artifactId>
-            <version>${flink.version}</version>
-            <scope>provided</scope>
-        </dependency>
-
-        <dependency>
-            <groupId>org.apache.flink</groupId>
-            <artifactId>flink-table-api-java</artifactId>
-            <version>${flink.version}</version>
-            <scope>provided</scope>
-        </dependency>
-
-        <dependency>
-            <groupId>org.apache.flink</groupId>
-            <artifactId>flink-table-api-java-bridge</artifactId>
+            <artifactId>flink-table-planner-loader</artifactId>
             <version>${flink.version}</version>
             <scope>provided</scope>
         </dependency>
 
         <dependency>
             <groupId>org.apache.flink</groupId>
             <artifactId>flink-table-runtime</artifactId>
-            <scope>provided</scope>
-        </dependency>
-
-        <dependency>
-            <groupId>org.apache.flink</groupId>
-            <artifactId>flink-table-common</artifactId>
             <version>${flink.version}</version>
             <scope>provided</scope>
         </dependency>
 
-        <dependency>
-            <groupId>org.apache.flink</groupId>
-            <artifactId>flink-table-planner-loader</artifactId>
-            <scope>provided</scope>
-        </dependency>
-
         <dependency>
             <groupId>org.apache.flink</groupId>
             <artifactId>flink-test-utils</artifactId>
diff --git a/flink-ml-iteration/src/main/java/org/apache/flink/iteration/operator/perround/AbstractPerRoundWrapperOperator.java b/flink-ml-iteration/src/main/java/org/apache/flink/iteration/operator/perround/AbstractPerRoundWrapperOperator.java
@@ -158,7 +158,8 @@ private S getWrappedOperator(
         // We need to clone the operator factory to also support SimpleOperatorFactory.
         try {
             StreamOperatorFactory<T> clonedOperatorFactory =
-                    InstantiationUtil.clone(operatorFactory);
+                    InstantiationUtil.clone(
+                            operatorFactory, containingTask.getUserCodeClassLoader());
             wrappedOperator =
                     (S)
                             StreamOperatorFactoryUtil.<T, S>createOperator(
