[HUDI-2799] Fix the classloader of flink write task (apache#4042)

vinothchandar · Nov 22, 2021 · 8281cbf · 8281cbf
1 parent 2533a9c
commit 8281cbf
Show file tree

Hide file tree

Showing 3 changed files with 6 additions and 0 deletions.
diff --git a/hudi-flink/src/main/java/org/apache/hudi/sink/bulk/BulkInsertWriteFunction.java b/hudi-flink/src/main/java/org/apache/hudi/sink/bulk/BulkInsertWriteFunction.java
@@ -111,6 +111,8 @@ public BulkInsertWriteFunction(Configuration config, RowType rowType) {
 
   @Override
   public void open(Configuration parameters) throws IOException {
+    // always use the user classloader
+    Thread.currentThread().setContextClassLoader(getRuntimeContext().getUserCodeClassLoader());
     this.taskID = getRuntimeContext().getIndexOfThisSubtask();
     this.metaClient = StreamerUtil.createMetaClient(this.config);
     this.writeClient = StreamerUtil.createWriteClient(this.config, getRuntimeContext());

diff --git a/hudi-flink/src/main/java/org/apache/hudi/sink/common/AbstractStreamWriteFunction.java b/hudi-flink/src/main/java/org/apache/hudi/sink/common/AbstractStreamWriteFunction.java
@@ -125,6 +125,8 @@ public AbstractStreamWriteFunction(Configuration config) {
 
   @Override
   public void initializeState(FunctionInitializationContext context) throws Exception {
+    // always use the user classloader
+    Thread.currentThread().setContextClassLoader(getRuntimeContext().getUserCodeClassLoader());
     this.taskID = getRuntimeContext().getIndexOfThisSubtask();
     this.metaClient = StreamerUtil.createMetaClient(this.config);
     this.writeClient = StreamerUtil.createWriteClient(this.config, getRuntimeContext());

diff --git a/hudi-flink/src/main/java/org/apache/hudi/sink/compact/CompactFunction.java b/hudi-flink/src/main/java/org/apache/hudi/sink/compact/CompactFunction.java
@@ -75,6 +75,8 @@ public CompactFunction(Configuration conf) {
 
   @Override
   public void open(Configuration parameters) throws Exception {
+    // always use the user classloader
+    Thread.currentThread().setContextClassLoader(getRuntimeContext().getUserCodeClassLoader());
     this.taskID = getRuntimeContext().getIndexOfThisSubtask();
     this.writeClient = StreamerUtil.createWriteClient(conf, getRuntimeContext());
     if (this.asyncCompaction) {