[hudi] branch master updated: Fix usage of SchemaRegistryProvider with Row source (#7601)

[vb...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

vbalaji pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/hudi.git


The following commit(s) were added to refs/heads/master by this push:
     new 24c9a86dae2 Fix usage of SchemaRegistryProvider with Row source (#7601)
24c9a86dae2 is described below

commit 24c9a86dae232f9597bd4ea4c3dc8573fa070322
Author: Pratyaksh Sharma <pr...@gmail.com>
AuthorDate: Thu Feb 23 10:58:19 2023 +0530

    Fix usage of SchemaRegistryProvider with Row source (#7601)
---
 .../org/apache/hudi/utilities/deltastreamer/SourceFormatAdapter.java   | 3 ++-
 1 file changed, 2 insertions(+), 1 deletion(-)

diff --git a/hudi-utilities/src/main/java/org/apache/hudi/utilities/deltastreamer/SourceFormatAdapter.java b/hudi-utilities/src/main/java/org/apache/hudi/utilities/deltastreamer/SourceFormatAdapter.java
index 9fbb11fd828..c8aa2f951c9 100644
--- a/hudi-utilities/src/main/java/org/apache/hudi/utilities/deltastreamer/SourceFormatAdapter.java
+++ b/hudi-utilities/src/main/java/org/apache/hudi/utilities/deltastreamer/SourceFormatAdapter.java
@@ -25,6 +25,7 @@ import org.apache.hudi.exception.HoodieIOException;
 import org.apache.hudi.utilities.UtilHelpers;
 import org.apache.hudi.utilities.schema.FilebasedSchemaProvider;
 import org.apache.hudi.utilities.schema.SchemaProvider;
+import org.apache.hudi.utilities.schema.SchemaRegistryProvider;
 import org.apache.hudi.utilities.sources.InputBatch;
 import org.apache.hudi.utilities.sources.Source;
 import org.apache.hudi.utilities.sources.helpers.AvroConvertor;
@@ -72,7 +73,7 @@ public final class SourceFormatAdapter implements Closeable {
         return new InputBatch<>(Option.ofNullable(r.getBatch().map(
             rdd -> {
               SchemaProvider originalProvider = UtilHelpers.getOriginalSchemaProvider(r.getSchemaProvider());
-              return (originalProvider instanceof FilebasedSchemaProvider)
+              return ((originalProvider instanceof FilebasedSchemaProvider) || (originalProvider instanceof SchemaRegistryProvider))
                   // 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