lanyuanxiaoyao/hudi
1
0
Fork 0
Code Activity

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

Browse Source
This commit is contained in:
Danny Chan
2021-11-22 11:05:05 +08:00
committed by GitHub
parent 2533a9cc17
commit 8281cbf762
3 changed files with 6 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
hudi-flink/src/main/java/org/apache/hudi/sink/bulk/BulkInsertWriteFunction.java
View File

@@ -111,6 +111,8 @@ public class BulkInsertWriteFunction<I>
@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());

2
hudi-flink/src/main/java/org/apache/hudi/sink/common/AbstractStreamWriteFunction.java
View File

@@ -125,6 +125,8 @@ public abstract class AbstractStreamWriteFunction<I>
@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());

2
hudi-flink/src/main/java/org/apache/hudi/sink/compact/CompactFunction.java
View File

@@ -75,6 +75,8 @@ public class CompactFunction extends ProcessFunction<CompactionPlanEvent, Compac
@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) {