pratyakshsharma commented on a change in pull request #1584:
URL: https://github.com/apache/incubator-hudi/pull/1584#discussion_r419149028
##########
File path:
hudi-utilities/src/main/java/org/apache/hudi/utilities/deltastreamer/SourceFormatAdapter.java
##########
@@ -64,19 +64,22 @@ public SourceFormatAdapter(Source source) {
}
case ROW: {
InputBatch<Dataset<Row>> r = ((RowSource)
source).fetchNext(lastCkptStr, sourceLimit);
- return new InputBatch<>(Option.ofNullable(r.getBatch().map(
- rdd -> (
- (r.getSchemaProvider() instanceof FilebasedSchemaProvider)
- // If the source schema is specified through Avro schema,
- // pass in the schema for the Row-to-Avro conversion
- // to avoid nullability mismatch between Avro schema and
Row schema
- ? AvroConversionUtils.createRdd(
- rdd, r.getSchemaProvider().getSourceSchema(),
- HOODIE_RECORD_STRUCT_NAME,
HOODIE_RECORD_NAMESPACE).toJavaRDD()
- : AvroConversionUtils.createRdd(
- rdd, HOODIE_RECORD_STRUCT_NAME,
HOODIE_RECORD_NAMESPACE).toJavaRDD()
- ))
- .orElse(null)), r.getCheckpointForNextBatch(),
r.getSchemaProvider());
+ if (r.getBatch().isPresent()) {
Review comment:
I think since this method tries to fetch data in avro format, pre
specifying a schema provider is mandatory. So even if you do not get any data,
you should mention RowBasedSchemaProvider as the schema provider in the very
beginning. If that is done, there is no need to do this change I believe. :)
Do you face issues after pre specifying schema provider?
Please let me know your thoughts on this.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
