lanyuanxiaoyao/hudi
1
0
Fork 0
Code Activity

[MINOR] Throw an exception when keyGenerator initialization failed (#2307)

Browse Source
This commit is contained in:
wangxianghu
2020-12-10 09:56:19 +08:00
committed by GitHub
parent fce1453fa6
commit 007014c1ef
1 changed files with 5 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files

13
hudi-flink/src/main/java/org/apache/hudi/source/JsonStringToHoodieRecordMapFunction.java
View File

@@ -23,7 +23,9 @@ import org.apache.hudi.avro.HoodieAvroUtils;
import org.apache.hudi.common.config.TypedProperties;
import org.apache.hudi.common.model.HoodieRecord;
import org.apache.hudi.common.model.HoodieRecordPayload;
import org.apache.hudi.exception.HoodieFlinkStreamerException;
import org.apache.hudi.keygen.KeyGenerator;
import org.apache.hudi.keygen.SimpleAvroKeyGenerator;
import org.apache.hudi.schema.FilebasedSchemaProvider;
import org.apache.hudi.util.AvroConvertor;
import org.apache.hudi.util.StreamerUtil;
@@ -31,9 +33,6 @@ import org.apache.hudi.util.StreamerUtil;
import org.apache.avro.generic.GenericRecord;
import org.apache.flink.api.common.functions.MapFunction;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.IOException;
/**
@@ -41,10 +40,7 @@ import java.io.IOException;
*/
public class JsonStringToHoodieRecordMapFunction implements MapFunction<String, HoodieRecord> {
private static Logger LOG = LoggerFactory.getLogger(JsonStringToHoodieRecordMapFunction.class);
private final HoodieFlinkStreamer.Config cfg;
private TypedProperties props;
private KeyGenerator keyGenerator;
private AvroConvertor avroConvertor;
@@ -63,12 +59,13 @@ public class JsonStringToHoodieRecordMapFunction implements MapFunction<String,
}
private void init() {
this.props = StreamerUtil.getProps(cfg);
TypedProperties props = StreamerUtil.getProps(cfg);
avroConvertor = new AvroConvertor(new FilebasedSchemaProvider(props).getSourceSchema());
try {
keyGenerator = StreamerUtil.createKeyGenerator(props);
} catch (IOException e) {
LOG.error("Init keyGenerator failed ", e);
throw new HoodieFlinkStreamerException(String.format("KeyGenerator %s initialization failed",
props.getString("hoodie.datasource.write.keygenerator.class", SimpleAvroKeyGenerator.class.getName())), e);
}
}
}