Abris #246 fix profile for Spark 2.3

Abris #246 code review changes II

Author: cerveada

pom.xml

@@ -69,6 +69,7 @@
 
         <!--Platforms-->
         <spark.version>${spark-24.version}</spark.version>
+        <spark.avro.version>${spark.version}</spark.avro.version>
         <kafka.spark.version>0-10</kafka.spark.version>
         <confluent.version>5.3.4</confluent.version>
 
@@ -159,7 +160,7 @@
         <dependency>
             <groupId>org.apache.spark</groupId>
             <artifactId>spark-avro_${scala.compat.version}</artifactId>
-            <version>${spark.version}</version>
+            <version>${spark.avro.version}</version>
         </dependency>
 
         <!-- Avro -->
@@ -334,6 +335,9 @@
             <id>spark-2.3</id>
             <properties>
                 <spark.version>${spark-23.version}</spark.version>
+                <spark.avro.version>${spark-24.version}</spark.avro.version>
+                <avro.version>1.8.2</avro.version>
+                <confluent.version>5.3.4</confluent.version>
             </properties>
         </profile>
         <profile>

src/main/scala/org/apache/spark/sql/avro/AbrisAvroDeserializer.scala

@@ -32,10 +32,10 @@ class AbrisAvroDeserializer(rootAvroType: Schema, rootCatalystType: DataType) {
     val clazz = classOf[AvroDeserializer]
     Try {
       clazz.getConstructor(classOf[Schema], classOf[DataType])
-        .newInstance(rootAvroType, rootCatalystType)
+        .newInstance(rootAvroType, rootCatalystType) // Spark 2.4 -
     }.recover { case _: NoSuchMethodException =>
       clazz.getConstructor(classOf[Schema], classOf[DataType], classOf[String])
-        .newInstance(rootAvroType, rootCatalystType, "LEGACY")
+        .newInstance(rootAvroType, rootCatalystType, "LEGACY") // Spark 3.0 +
     }
       .get
       .asInstanceOf[AvroDeserializer]
@@ -49,8 +49,8 @@ class AbrisAvroDeserializer(rootAvroType: Schema, rootCatalystType: DataType) {
 
   def deserialize(data: Any): Any = {
     deserializeMethod(data) match {
-      case Some(x) => x
-      case x => x
+      case Some(x) => x // Spark 3.1 +
+      case x => x // Spark 3.0 -
     }
   }
 

src/main/scala/za/co/absa/abris/avro/read/confluent/SchemaManagerFactory.scala

@@ -18,11 +18,12 @@
 package za.co.absa.abris.avro.read.confluent
 
 import org.apache.spark.internal.Logging
-import za.co.absa.abris.avro.registry.{AbrisRegistryClient, ConfluentRegistryClient}
+import za.co.absa.abris.avro.registry.{AbrisRegistryClient, ConfluentMockRegistryClient, ConfluentRegistryClient}
 import za.co.absa.abris.config.AbrisConfig
 import za.co.absa.commons.annotation.DeveloperApi
 
 import scala.collection.concurrent
+import scala.util.Try
 import scala.util.control.NonFatal
 
 /**
@@ -53,16 +54,21 @@ object SchemaManagerFactory extends Logging {
         try {
           val clazz = Class.forName(configs(AbrisConfig.REGISTRY_CLIENT_CLASS))
           logInfo(msg = s"Configuring new Schema Registry client of type '${clazz.getCanonicalName}'")
-          clazz
-            .getDeclaredConstructor(classOf[Map[String, String]])
-            .newInstance(configs)
+          Try(clazz.getConstructor(classOf[Map[String, String]]).newInstance(configs))
+            .recover { case _: NoSuchMethodException =>
+              clazz.getConstructor().newInstance()
+            }
+            .get
             .asInstanceOf[AbrisRegistryClient]
         } catch {
           case e if NonFatal(e) =>
             throw new IllegalArgumentException("Custom registry client must implement AbrisRegistryClient " +
-              "and have constructor accepting Map[String, String]", e)
+              "and have parameterless or Map[String, String] accepting constructor", e)
         }
-      } else {
+    } else if (configs(AbrisConfig.SCHEMA_REGISTRY_URL).startsWith("mock://")) {
+        logInfo(msg = s"Configuring new Schema Registry client of type ConfluentMockRegistryClient")
+        new ConfluentMockRegistryClient()
+    } else {
         logInfo(msg = s"Configuring new Schema Registry client of type ConfluentRegistryClient")
         new ConfluentRegistryClient(configs)
       }

src/main/scala/za/co/absa/abris/avro/registry/ConfluentMockRegistryClient.scala

@@ -17,17 +17,16 @@
 package za.co.absa.abris.avro.registry
 
 import io.confluent.kafka.schemaregistry.client.rest.exceptions.RestClientException
-import io.confluent.kafka.schemaregistry.client.{MockSchemaRegistryClient, SchemaMetadata}
+import io.confluent.kafka.schemaregistry.client.{MockSchemaRegistryClient, SchemaMetadata, SchemaRegistryClient}
 import org.apache.avro.Schema
 
 import java.io.IOException
 import java.util
 
 
-class ConfluentMockRegistryClient extends AbrisRegistryClient {
-
-  private val client = new MockSchemaRegistryClient()
+class ConfluentMockRegistryClient(client: SchemaRegistryClient) extends AbrisRegistryClient {
 
+  def this() = this(new MockSchemaRegistryClient())
 
   override def getAllVersions(subject: String): util.List[Integer] =
     client.getAllVersions(subject)
@@ -44,10 +43,10 @@ class ConfluentMockRegistryClient extends AbrisRegistryClient {
   @throws[IOException]
   @throws[RestClientException]
   override def getLatestSchemaMetadata(subject: String): SchemaMetadata = {
-    try (client.getLatestSchemaMetadata(subject))
+    try client.getLatestSchemaMetadata(subject)
     catch {
-      case e: IOException if e.getMessage == "No schema registered under subject!"
-      => throw new RestClientException("No schema registered under subject!", 404, 40401)
+      case e: IOException if e.getMessage == "No schema registered under subject!" =>
+        throw new RestClientException("No schema registered under subject!", 404, 40401)
     }
   }
 

src/test/scala/za/co/absa/abris/avro/read/confluent/SchemaManagerFactorySpec.scala

@@ -17,7 +17,7 @@
 package za.co.absa.abris.avro.read.confluent
 
 import org.scalatest.{BeforeAndAfterEach, FlatSpec}
-import za.co.absa.abris.avro.registry.{AbrisRegistryClient, ConfluentRegistryClient, TestRegistryClient}
+import za.co.absa.abris.avro.registry.{AbrisRegistryClient, ConfluentMockRegistryClient, ConfluentRegistryClient, TestRegistryClient}
 import za.co.absa.abris.config.AbrisConfig
 
 import scala.reflect.runtime.{universe => ru}
@@ -78,4 +78,16 @@ class SchemaManagerFactorySpec extends FlatSpec with BeforeAndAfterEach {
     assert(res1.isInstanceOf[ConfluentRegistryClient])
     assert(res3.isInstanceOf[TestRegistryClient])
   }
+
+  it should "create mock client when url starts with mock://" in {
+    val config = Map(AbrisConfig.SCHEMA_REGISTRY_URL -> "mock://dummy_sr_2")
+
+    val schemaManagerRef1 = SchemaManagerFactory.create(config)
+
+    val m = ru.runtimeMirror(schemaManagerRef1.getClass.getClassLoader)
+    val fieldTerm = ru.typeOf[SchemaManager].decl(ru.TermName("schemaRegistryClient")).asTerm
+
+    val res1 = m.reflect(schemaManagerRef1).reflectField(fieldTerm).get.asInstanceOf[AbrisRegistryClient]
+    assert(res1.isInstanceOf[ConfluentMockRegistryClient])
+  }
 }