lanyuanxiaoyao/hudi
1
0
Fork 0
Code Activity

[HUDI-1953] Fix NPE due to not set the output type of the operator (#3023)

Browse Source 
Co-authored-by: enter58xuan <enter58xuan@zto.com>
This commit is contained in:
taylorliao
2021-06-03 14:20:57 +08:00
committed by GitHub
parent 05a9830e86
commit 86007e9a13
2 changed files with 2 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
hudi-flink/src/main/java/org/apache/hudi/streamer/HoodieFlinkStreamer.java
View File

@@ -103,7 +103,7 @@ public class HoodieFlinkStreamer {
.uid("uid_bucket_assigner") .uid("uid_bucket_assigner")
// shuffle by fileId(bucket id) // shuffle by fileId(bucket id)
.keyBy(record -> record.getCurrentLocation().getFileId()) .keyBy(record -> record.getCurrentLocation().getFileId())
.transform("hoodie_stream_write", null, operatorFactory) .transform("hoodie_stream_write", TypeInformation.of(Object.class), operatorFactory)
.uid("uid_hoodie_stream_write") .uid("uid_hoodie_stream_write")
.setParallelism(numWriteTask); .setParallelism(numWriteTask);
if (StreamerUtil.needsScheduleCompaction(conf)) { if (StreamerUtil.needsScheduleCompaction(conf)) {

2
hudi-flink/src/test/java/org/apache/hudi/sink/StreamWriteITCase.java
View File

@@ -125,7 +125,7 @@ public class StreamWriteITCase extends TestLogger {
.uid("uid_bucket_assigner") .uid("uid_bucket_assigner")
// shuffle by fileId(bucket id) // shuffle by fileId(bucket id)
.keyBy(record -> record.getCurrentLocation().getFileId()) .keyBy(record -> record.getCurrentLocation().getFileId())
.transform("hoodie_stream_write", null, operatorFactory) .transform("hoodie_stream_write", TypeInformation.of(Object.class), operatorFactory)
.uid("uid_hoodie_stream_write"); .uid("uid_hoodie_stream_write");
execEnv.addOperator(dataStream.getTransformation()); execEnv.addOperator(dataStream.getTransformation());