lanyuanxiaoyao/hudi
1
0
Fork 0
Code Activity

[HUDI-3413]fix jackson parse error when empty message from JsonKafkaSource Using HoodieDeltaStreamer (#4794)

Browse Source
This commit is contained in:
zhangxiang17
2022-02-12 15:37:29 +08:00
committed by GitHub
parent 89ed6f062e
commit 9518f78610
1 changed files with 2 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/JsonKafkaSource.java
View File

@@ -20,6 +20,7 @@ package org.apache.hudi.utilities.sources;
import org.apache.hudi.common.config.TypedProperties; import org.apache.hudi.common.config.TypedProperties;
import org.apache.hudi.common.util.Option; import org.apache.hudi.common.util.Option;
import org.apache.hudi.common.util.StringUtils;
import org.apache.hudi.utilities.deltastreamer.HoodieDeltaStreamerMetrics; import org.apache.hudi.utilities.deltastreamer.HoodieDeltaStreamerMetrics;
import org.apache.hudi.utilities.exception.HoodieSourceTimeoutException; import org.apache.hudi.utilities.exception.HoodieSourceTimeoutException;
import org.apache.hudi.utilities.schema.SchemaProvider; import org.apache.hudi.utilities.schema.SchemaProvider;
@@ -79,7 +80,7 @@ public class JsonKafkaSource extends JsonSource {
offsetGen.getKafkaParams(), offsetGen.getKafkaParams(),
offsetRanges, offsetRanges,
LocationStrategies.PreferConsistent()) LocationStrategies.PreferConsistent())
.filter(x -> Objects.nonNull(x.value())) .filter(x -> !StringUtils.isNullOrEmpty((String)x.value()))
.map(x -> x.value().toString()); .map(x -> x.value().toString());
} }