add userCodeClassLoader

Author: polyzos

fluss-flink/fluss-flink-common/src/main/java/com/alibaba/fluss/flink/sink/FlinkSink.java

@@ -59,15 +59,19 @@ class FlinkSink<InputT> implements Sink<InputT>, SupportsPreWriteTopology<InputT
     public SinkWriter<InputT> createWriter(InitContext context) throws IOException {
         FlinkSinkWriter<InputT> flinkSinkWriter =
                 builder.createWriter(context.getMailboxExecutor());
-        flinkSinkWriter.initialize(InternalSinkWriterMetricGroup.wrap(context.metricGroup()));
+        flinkSinkWriter.initialize(
+                InternalSinkWriterMetricGroup.wrap(context.metricGroup()),
+                context.getUserCodeClassLoader());
         return flinkSinkWriter;
     }
 
     @Override
     public SinkWriter<InputT> createWriter(WriterInitContext context) throws IOException {
         FlinkSinkWriter<InputT> flinkSinkWriter =
                 builder.createWriter(context.getMailboxExecutor());
-        flinkSinkWriter.initialize(InternalSinkWriterMetricGroup.wrap(context.metricGroup()));
+        flinkSinkWriter.initialize(
+                InternalSinkWriterMetricGroup.wrap(context.metricGroup()),
+                context.getUserCodeClassLoader());
         return flinkSinkWriter;
     }
 

fluss-flink/fluss-flink-common/src/main/java/com/alibaba/fluss/flink/sink/writer/AppendSinkWriter.java

@@ -27,6 +27,7 @@
 import org.apache.flink.metrics.groups.SinkWriterMetricGroup;
 import org.apache.flink.table.types.logical.RowType;
 import org.apache.flink.types.RowKind;
+import org.apache.flink.util.UserCodeClassLoader;
 
 import java.io.IOException;
 import java.util.concurrent.CompletableFuture;
@@ -55,8 +56,9 @@ public AppendSinkWriter(
     }
 
     @Override
-    public void initialize(SinkWriterMetricGroup metricGroup) {
-        super.initialize(metricGroup);
+    public void initialize(
+            SinkWriterMetricGroup metricGroup, UserCodeClassLoader userCodeClassLoader) {
+        super.initialize(metricGroup, userCodeClassLoader);
 
         appendWriter = table.newAppend().createWriter();
         LOG.info("Finished opening Fluss {}.", this.getClass().getSimpleName());

fluss-flink/fluss-flink-common/src/main/java/com/alibaba/fluss/flink/sink/writer/FlinkSinkWriter.java

@@ -82,7 +82,7 @@ public FlinkSinkWriter(
             RowType tableRowType,
             boolean ignoreDelete,
             MailboxExecutor mailboxExecutor,
-            FlussSerializationSchema serializationSchema) {
+            FlussSerializationSchema<InputT> serializationSchema) {
         this(
                 tablePath,
                 flussConfig,
@@ -110,7 +110,8 @@ public FlinkSinkWriter(
         this.serializationSchema = serializationSchema;
     }
 
-    public void initialize(SinkWriterMetricGroup metricGroup) {
+    public void initialize(
+            SinkWriterMetricGroup metricGroup, UserCodeClassLoader userCodeClassLoader) {
         LOG.info(
                 "Opening Fluss {}, database: {} and table: {}",
                 this.getClass().getSimpleName(),
@@ -137,7 +138,7 @@ public MetricGroup getMetricGroup() {
 
                         @Override
                         public UserCodeClassLoader getUserCodeClassLoader() {
-                            return null; // TODO: add user code class loader here
+                            return userCodeClassLoader;
                         }
 
                         @Override

fluss-flink/fluss-flink-common/src/main/java/com/alibaba/fluss/flink/sink/writer/UpsertSinkWriter.java

@@ -28,6 +28,7 @@
 import org.apache.flink.metrics.groups.SinkWriterMetricGroup;
 import org.apache.flink.table.types.logical.RowType;
 import org.apache.flink.types.RowKind;
+import org.apache.flink.util.UserCodeClassLoader;
 
 import javax.annotation.Nullable;
 
@@ -58,8 +59,9 @@ public UpsertSinkWriter(
     }
 
     @Override
-    public void initialize(SinkWriterMetricGroup metricGroup) {
-        super.initialize(metricGroup);
+    public void initialize(
+            SinkWriterMetricGroup metricGroup, UserCodeClassLoader userCodeClassLoader) {
+        super.initialize(metricGroup, userCodeClassLoader);
         Upsert upsert = table.newUpsert();
         if (targetColumnIndexes != null) {
             upsert = upsert.partialUpdate(targetColumnIndexes);

fluss-flink/fluss-flink-common/src/test/java/com/alibaba/fluss/flink/sink/writer/FlinkSinkWriterTest.java

@@ -88,7 +88,9 @@ void testSinkMetrics(String clientId) throws Exception {
         FlinkSinkWriter flinkSinkWriter =
                 createSinkWriter(flussConf, mockWriterInitContext.getMailboxExecutor());
 
-        flinkSinkWriter.initialize(mockWriterInitContext.metricGroup());
+        flinkSinkWriter.initialize(
+                mockWriterInitContext.metricGroup(),
+                mockWriterInitContext.getUserCodeClassLoader());
         flinkSinkWriter.write(
                 GenericRowData.of(1, StringData.fromString("a")), new MockSinkWriterContext());
         flinkSinkWriter.flush(false);
@@ -188,7 +190,9 @@ private void testExceptionWhenFlussUnavailable(
             // test fluss unavailable.
             try (FlinkSinkWriter writer =
                     createSinkWriter(clientConfig, mockWriterInitContext.getMailboxExecutor())) {
-                writer.initialize(mockWriterInitContext.metricGroup());
+                writer.initialize(
+                        mockWriterInitContext.metricGroup(),
+                        mockWriterInitContext.getUserCodeClassLoader());
                 flussClusterExtension.close();
                 writer.write(
                         GenericRowData.of(1, StringData.fromString("a")),