Apparently reflection utils can't create a class if one of the parameters is null, forced to add another constructor.

Author: volodymyr.burenin

hudi-utilities/src/main/java/org/apache/hudi/utilities/deser/KafkaAvroSchemaDeserializer.java

@@ -22,12 +22,11 @@
 import io.confluent.kafka.serializers.KafkaAvroDeserializer;
 import org.apache.avro.Schema;
 import org.apache.hudi.common.config.TypedProperties;
+import org.apache.hudi.common.util.ReflectionUtils;
 import org.apache.hudi.exception.HoodieException;
-import org.apache.hudi.utilities.UtilHelpers;
 import org.apache.hudi.utilities.schema.SchemaProvider;
 import org.apache.kafka.common.errors.SerializationException;
 
-import java.io.IOException;
 import java.util.Map;
 import java.util.Map.Entry;
 import java.util.Objects;
@@ -47,10 +46,10 @@ public void configure(Map<String, ?> configs, boolean isKey) {
     super.configure(configs, isKey);
     try {
       TypedProperties props = getConvertToTypedProperties(configs);
-      SchemaProvider schemaProvider = UtilHelpers.createSchemaProvider(
-          props.getString(SCHEMA_PROVIDER_CLASS_PROP), props, null);
+      String className = props.getString(SCHEMA_PROVIDER_CLASS_PROP);
+      SchemaProvider schemaProvider = (SchemaProvider) ReflectionUtils.loadClass(className, props);
       sourceSchema = Objects.requireNonNull(schemaProvider).getSourceSchema();
-    } catch (IOException e) {
+    } catch (Throwable e) {
       throw new HoodieException(e);
     }
   }

hudi-utilities/src/main/java/org/apache/hudi/utilities/schema/SchemaRegistryProvider.java

@@ -56,11 +56,8 @@ public static class Config {
   private final String targetRegistryUrl;
   private final boolean noTargetSchema;
 
-  public String fetchSchemaFromRegistry(String registryUrl) throws IOException {
-    URL registry = new URL(registryUrl);
-    ObjectMapper mapper = new ObjectMapper();
-    JsonNode node = mapper.readTree(registry.openStream());
-    return node.get("schema").asText();
+  public SchemaRegistryProvider(TypedProperties props) {
+    this(props, null);
   }
 
   public SchemaRegistryProvider(TypedProperties props, JavaSparkContext jssc) {
@@ -73,6 +70,13 @@ public SchemaRegistryProvider(TypedProperties props, JavaSparkContext jssc) {
     this.noTargetSchema = targetRegistryUrl.equals("null");
   }
 
+  public String fetchSchemaFromRegistry(String registryUrl) throws IOException {
+    URL registry = new URL(registryUrl);
+    ObjectMapper mapper = new ObjectMapper();
+    JsonNode node = mapper.readTree(registry.openStream());
+    return node.get("schema").asText();
+  }
+
   private Schema getSchema(String registryUrl) throws IOException {
     Schema schema = new Schema.Parser().parse(fetchSchemaFromRegistry(registryUrl));
     if (injectKafkaFieldSchema) {

hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/AvroKafkaSource.java

@@ -44,7 +44,7 @@
  */
 public class AvroKafkaSource extends AvroSource {
 
-  private static final String KAFKA_AVRO_VALUE_DESERIALIZER = "hoodie.deltastreamer.source.value.deserializer";
+  private static final String KAFKA_AVRO_VALUE_DESERIALIZER = "hoodie.deltastreamer.source.value.deserializer.class";
   private static final Logger LOG = LogManager.getLogger(AvroKafkaSource.class);
   private final KafkaOffsetGen offsetGen;
   private final HoodieDeltaStreamerMetrics metrics;