[#19] WIP Different loading strategies first review

Pirazzini Lorenzo · Pirazzini Lorenzo · commit 96ea26226528 · 2018-11-12T09:05:38.000+01:00
diff --git a/common/src/main/scala/it/agilelab/darwin/common/Connector.scala b/common/src/main/scala/it/agilelab/darwin/common/Connector.scala
@@ -26,4 +26,6 @@ abstract class Connector(config: Config) extends Serializable {
     * @param schemas a sequence of pairs (ID, schema) Schema entities to insert in the storage.
     */
   def insert(schemas: Seq[(Long, Schema)]): Unit
+
+  def findSchema(id: Long): Option[Schema]
 }
diff --git a/core/src/main/scala/it/agilelab/darwin/manager/AvroSchemaCache.scala b/core/src/main/scala/it/agilelab/darwin/manager/AvroSchemaCache.scala
@@ -8,21 +8,13 @@ import org.apache.avro.Schema
   */
 abstract class AvroSchemaCache(schemas: Seq[(Long, Schema)]) {
 
-  /**
-    * Retrieves the ID of a registered schema.
-    *
-    * @param schema the Schema for which an ID is required
-    * @return the ID associated to the input schema
-    */
-  def getId(schema: Schema): Long
-
   /**
     * Retrieves a registered schema for the input ID.
     *
     * @param id the Long ID of the schema
     * @return the Schema associated to the input ID
     */
-  def getSchema(id: Long): Schema
+  def getSchema(id: Long): Option[Schema]
 
   /**
     * Tests if the input schema is contained inside the cache.
diff --git a/core/src/main/scala/it/agilelab/darwin/manager/AvroSchemaCacheFingerprint.scala b/core/src/main/scala/it/agilelab/darwin/manager/AvroSchemaCacheFingerprint.scala
@@ -1,7 +1,5 @@
 package it.agilelab.darwin.manager
 
-import java.security.InvalidKeyException
-
 import it.agilelab.darwin.common.Logging
 import org.apache.avro.{Schema, SchemaNormalization}
 
@@ -15,10 +13,7 @@ case class AvroSchemaCacheFingerprint(schemas: Seq[(Long, Schema)]) extends Avro
   private val _table: Map[Long, Schema] = schemas.toMap
   log.debug("cache initialized")
 
-  override def getId(schema: Schema): Long = SchemaNormalization.parsingFingerprint64(schema)
-
-  override def getSchema(id: Long): Schema = _table.getOrElse(id,
-    throw new InvalidKeyException(s"No schema registered in cache for id $id"))
+  override def getSchema(id: Long): Option[Schema] = _table.get(id)
 
   override def contains(schema: Schema): (Boolean, Long) = {
     val id = SchemaNormalization.parsingFingerprint64(schema)
diff --git a/core/src/main/scala/it/agilelab/darwin/manager/AvroSchemaManager.scala b/core/src/main/scala/it/agilelab/darwin/manager/AvroSchemaManager.scala
@@ -4,8 +4,9 @@ import com.typesafe.config.Config
 import it.agilelab.darwin.common.{Connector, ConnectorFactory, Logging}
 import it.agilelab.darwin.manager.exception.ConnectorNotFoundException
 import jdk.nashorn.internal.runtime.ParserException
-import org.apache.avro.Schema
+import org.apache.avro.{Schema, SchemaNormalization}
 import it.agilelab.darwin.manager.util.ByteArrayUtils._
+import scala.collection.JavaConverters._
 
 trait AvroSchemaManager extends Logging {
   private val V1_HEADER = Array[Byte](0xC3.toByte, 0x01.toByte)
@@ -23,15 +24,15 @@ trait AvroSchemaManager extends Logging {
     * @param schema a Schema with unknown ID
     * @return the ID associated with the input schema
     */
-  def getId(schema: Schema): Long
+  def getId(schema: Schema): Long = SchemaNormalization.parsingFingerprint64(schema)
 
   /**
     * Extracts the Schema from its ID.
     *
     * @param id a Long representing an ID
     * @return the Schema associated to the input ID
     */
-  def getSchema(id: Long): Schema
+  def getSchema(id: Long): Option[Schema]
 
   /**
     * Checks if all the input Schema elements are already in the cache. Then, it performs an insert on the
@@ -49,7 +50,9 @@ trait AvroSchemaManager extends Logging {
     * @param schemas all the Schema that should be registered
     * @return a sequence of pairs of the input schemas associated with their IDs
     */
-  def registerAll(schemas: java.lang.Iterable[Schema]): java.lang.Iterable[IdSchemaPair]
+  def registerAll(schemas: java.lang.Iterable[Schema]): java.lang.Iterable[IdSchemaPair] = {
+    registerAll(schemas.asScala.toSeq).map { case (id, schema) => IdSchemaPair.create(id, schema) }.asJava
+  }
 
   /** Create an array that creates a Single-Object encoded byte array.
     * By specifications the encoded array is obtained concatenating the V1_HEADER, the schema id and the avro-encoded
@@ -83,7 +86,7 @@ trait AvroSchemaManager extends Logging {
     */
   def retrieveSchemaAndAvroPayload(avroSingleObjectEncoded: Array[Byte]): (Schema, Array[Byte]) = {
     if (isAvroSingleObjectEncoded(avroSingleObjectEncoded)) {
-      getSchema(avroSingleObjectEncoded.slice(V1_HEADER.length, HEADER_LENGTH).byteArrayToLong) ->
+      getSchema(avroSingleObjectEncoded.slice(V1_HEADER.length, HEADER_LENGTH).byteArrayToLong).get ->
         avroSingleObjectEncoded.drop(HEADER_LENGTH)
     }
     else {
diff --git a/core/src/main/scala/it/agilelab/darwin/manager/CachedAvroSchemaManager.scala b/core/src/main/scala/it/agilelab/darwin/manager/CachedAvroSchemaManager.scala
@@ -1,11 +1,10 @@
 package it.agilelab.darwin.manager
 
 import java.util.concurrent.atomic.AtomicReference
-import scala.collection.JavaConverters._
 import org.apache.avro.Schema
 
 trait CachedAvroSchemaManager extends AvroSchemaManager {
-  private val _cache: AtomicReference[Option[AvroSchemaCache]] = new AtomicReference[Option[AvroSchemaCache]](None)
+  protected val _cache: AtomicReference[Option[AvroSchemaCache]] = new AtomicReference[Option[AvroSchemaCache]](None)
 
   def cache: AvroSchemaCache = _cache.get
     .getOrElse(throw new IllegalAccessException("Cache not loaded: accesses are allowed only if the cache has been " +
@@ -41,8 +40,4 @@ trait CachedAvroSchemaManager extends AvroSchemaManager {
     log.debug(s"${allSchemas.size} schemas registered")
     allSchemas
   }
-
-  override def registerAll(schemas: java.lang.Iterable[Schema]): java.lang.Iterable[IdSchemaPair] = {
-    registerAll(schemas.asScala.toSeq).map { case (id, schema) => IdSchemaPair.create(id, schema) }.asJava
-  }
 }
diff --git a/core/src/main/scala/it/agilelab/darwin/manager/CachedEagerAvroSchemaManager.scala b/core/src/main/scala/it/agilelab/darwin/manager/CachedEagerAvroSchemaManager.scala
@@ -10,7 +10,5 @@ import org.apache.avro.Schema
   */
 case class CachedEagerAvroSchemaManager(override val config: Config) extends CachedAvroSchemaManager {
 
-  override def getId(schema: Schema): Long = cache.getId(schema)
-
-  override def getSchema(id: Long): Schema = cache.getSchema(id)
+  override def getSchema(id: Long): Option[Schema] = cache.getSchema(id)
 }
diff --git a/core/src/main/scala/it/agilelab/darwin/manager/CachedLazyAvroSchemaManager.scala b/core/src/main/scala/it/agilelab/darwin/manager/CachedLazyAvroSchemaManager.scala
@@ -3,8 +3,12 @@ import com.typesafe.config.Config
 import org.apache.avro.Schema
 
 case class CachedLazyAvroSchemaManager(override val config: Config) extends CachedAvroSchemaManager {
-  //TODO
-  override def getId(schema: Schema): Long = ???
 
-  override def getSchema(id: Long): Schema = ???
+  override def getSchema(id: Long): Option[Schema] = {
+    cache.getSchema(id).orElse{
+      val schema: Option[Schema] = connector.findSchema(id)
+      schema.foreach(s => _cache.set(Some(cache.insert(Seq(getId(s) -> s)))))
+      schema
+    }
+  }
 }
diff --git a/core/src/main/scala/it/agilelab/darwin/manager/LazyAvroSchemaManager.scala b/core/src/main/scala/it/agilelab/darwin/manager/LazyAvroSchemaManager.scala
@@ -1,20 +1,17 @@
 package it.agilelab.darwin.manager
 
-import java.lang
-
 import com.typesafe.config.Config
 import org.apache.avro.Schema
 
 case class LazyAvroSchemaManager(override val config: Config) extends AvroSchemaManager {
-  //TODO
-
-  override def getId(schema: Schema): Long = ???
-
-  override def getSchema(id: Long): Schema = ???
 
-  override def registerAll(schemas: Seq[Schema]): Seq[(Long, Schema)] = ???
+  override def getSchema(id: Long): Option[Schema] = connector.findSchema(id)
 
-  override def registerAll(schemas: lang.Iterable[Schema]): lang.Iterable[IdSchemaPair] = ???
+  override def registerAll(schemas: Seq[Schema]): Seq[(Long, Schema)] = {
+    val schemasWithIds = schemas.map(s => getId(s) -> s)
+    connector.insert(schemasWithIds)
+    schemasWithIds
+  }
 
-  override def reload(): AvroSchemaManager = ???
+  override def reload(): AvroSchemaManager = this
 }
diff --git a/hbase/src/main/scala/it/agilelab/darwin/connector/hbase/HBaseConnector.scala b/hbase/src/main/scala/it/agilelab/darwin/connector/hbase/HBaseConnector.scala
@@ -125,6 +125,8 @@ case class HBaseConnector(config: Config) extends Connector(config) with Logging
     mutator.flush()
     log.debug(s"insertion of schemas into $NAMESPACE_STRING:$TABLE_NAME_STRING successful")
   }
+
+  override def findSchema(id: Long): Option[Schema] = ???
 }
 
 
diff --git a/ignite/src/main/scala/it/agilelab/darwin/connector/ignite/IgniteConnector.scala b/ignite/src/main/scala/it/agilelab/darwin/connector/ignite/IgniteConnector.scala
@@ -8,4 +8,6 @@ class IgniteConnector(config: Config) extends Connector(config) {
   override def fullLoad(): Seq[(Long, Schema)] = ???
 
   override def insert(schemas: Seq[(Long, Schema)]): Unit = ???
+
+  override def findSchema(id: Long): Option[Schema] = ???
 }
diff --git a/mock-application/src/test/java/it/agilelab/darwin/app/mock/JavaApplicationTest.java b/mock-application/src/test/java/it/agilelab/darwin/app/mock/JavaApplicationTest.java
@@ -27,7 +27,8 @@ void mainTest() {
         AvroSchemaManagerFactory.getInstance(ConfigFactory.empty()).registerAll(JavaConversions.asScalaBuffer(schemas));
 
         long id = manager.getId(schemas.get(0));
-        assert (schemas.get(0) == manager.getSchema(id));
+        assert(manager.getSchema(id).isDefined());
+        assert (schemas.get(0) == manager.getSchema(id).get());
     }
 
     void reflectionTest() {
diff --git a/mock-application/src/test/scala/it/agilelab/darwin/app/mock/CachedEagerApplicationSuite.scala b/mock-application/src/test/scala/it/agilelab/darwin/app/mock/CachedEagerApplicationSuite.scala
@@ -0,0 +1,77 @@
+package it.agilelab.darwin.app.mock
+
+import java.lang.reflect.Modifier
+
+import com.typesafe.config.{Config, ConfigFactory}
+import it.agilelab.darwin.annotations.AvroSerde
+import it.agilelab.darwin.app.mock.classes.{MyClass, MyNestedClass, NewClass, OneField}
+import it.agilelab.darwin.manager.{AvroSchemaManager, AvroSchemaManagerFactory}
+import org.apache.avro.{Schema, SchemaNormalization}
+import org.apache.avro.reflect.ReflectData
+import org.reflections.Reflections
+import org.scalatest.{FlatSpec, Matchers}
+
+import scala.collection.JavaConverters._
+
+class CachedEagerApplicationSuite extends FlatSpec with Matchers {
+
+  val config: Config = ConfigFactory.parseMap(Map("type" -> "cached_eager").asJava)
+  val manager: AvroSchemaManager = AvroSchemaManagerFactory.getInstance(config)
+
+  "CachedEagerAvroSchemaManager" should "not fail after the initialization" in {
+    val schemas: Seq[Schema] = Seq(new SchemaGenerator[MyNestedClass].schema)
+    assert(manager.registerAll(schemas).size == 1)
+  }
+
+  it should "load all existing schemas and register a new one" in {
+    val schemas: Seq[Schema] = Seq(new SchemaGenerator[MyNestedClass].schema)
+    manager.getSchema(0L)
+
+    manager.registerAll(schemas)
+
+    val id = manager.getId(schemas.head)
+    assert(manager.getSchema(id).isDefined)
+    assert(schemas.head == manager.getSchema(id).get)
+  }
+
+  it should "get all previously registered schemas" in {
+    val schema: Schema = new SchemaGenerator[MyNestedClass].schema
+    val schema0 = manager.getSchema(0L)
+    val schema1 = manager.getSchema(1L)
+    assert(schema0.isDefined)
+    assert(schema1.isDefined)
+    assert(schema0.get != schema1.get)
+    assert(schema != schema0.get)
+    assert(schema != schema1.get)
+  }
+
+  it should "generate all schemas for all the annotated classes with @AvroSerde" in {
+    val reflections = new Reflections("it.agilelab.darwin.app.mock.classes")
+
+    val oneFieldSchema = ReflectData.get().getSchema(classOf[OneField]).toString
+    val myNestedSchema = ReflectData.get().getSchema(classOf[MyNestedClass]).toString
+    val myClassSchema = ReflectData.get().getSchema(classOf[MyClass]).toString
+
+    val annotationClass: Class[AvroSerde] = classOf[AvroSerde]
+    val classes = reflections.getTypesAnnotatedWith(annotationClass).asScala.toSeq
+      .filter(c => !c.isInterface && !Modifier.isAbstract(c.getModifiers))
+    classes.foreach(println)
+    val schemas = classes.map(c => ReflectData.get().getSchema(Class.forName(c.getName)).toString)
+    schemas.foreach(println)
+    Seq(oneFieldSchema, myClassSchema, myNestedSchema) should contain theSameElementsAs schemas
+  }
+
+  it should "reload all schemas from the connector" in {
+    val newSchema = ReflectData.get().getSchema(classOf[NewClass])
+    val newId = SchemaNormalization.parsingFingerprint64(newSchema)
+    assert(manager.getSchema(newId).isEmpty)
+
+    manager.connector.insert(Seq(newId -> newSchema))
+    assert(manager.getSchema(newId).isEmpty)
+
+    manager.reload()
+    assert(manager.getSchema(newId).isDefined)
+    assert(manager.getSchema(newId).get == newSchema)
+  }
+
+}
diff --git a/mock-application/src/test/scala/it/agilelab/darwin/app/mock/CachedLazyApplicationSuite.scala b/mock-application/src/test/scala/it/agilelab/darwin/app/mock/CachedLazyApplicationSuite.scala
@@ -0,0 +1,73 @@
+package it.agilelab.darwin.app.mock
+
+import java.lang.reflect.Modifier
+
+import com.typesafe.config.{Config, ConfigFactory}
+import it.agilelab.darwin.annotations.AvroSerde
+import it.agilelab.darwin.app.mock.classes.{MyClass, MyNestedClass, NewClass, OneField}
+import it.agilelab.darwin.manager.{AvroSchemaManager, AvroSchemaManagerFactory}
+import org.apache.avro.{Schema, SchemaNormalization}
+import org.apache.avro.reflect.ReflectData
+import org.reflections.Reflections
+import org.scalatest.{FlatSpec, Matchers}
+
+import scala.collection.JavaConverters._
+
+class CachedLazyApplicationSuite extends FlatSpec with Matchers {
+
+  val config: Config = ConfigFactory.parseMap(Map("type" -> "cached_lazy").asJava)
+  val manager: AvroSchemaManager = AvroSchemaManagerFactory.getInstance(config)
+
+  "CachedLazyAvroSchemaManager" should "not fail after the initialization" in {
+    val schemas: Seq[Schema] = Seq(new SchemaGenerator[MyNestedClass].schema)
+    assert(manager.registerAll(schemas).size == 1)
+  }
+
+  it should "load all existing schemas and register a new one" in {
+    val schemas: Seq[Schema] = Seq(new SchemaGenerator[MyNestedClass].schema)
+    manager.getSchema(0L)
+
+    manager.registerAll(schemas)
+
+    val id = manager.getId(schemas.head)
+    assert(manager.getSchema(id).isDefined)
+    assert(schemas.head == manager.getSchema(id).get)
+  }
+
+  it should "get all previously registered schemas" in {
+    val schema: Schema = new SchemaGenerator[MyNestedClass].schema
+    val schema0 = manager.getSchema(0L)
+    val schema1 = manager.getSchema(1L)
+    assert(schema0.isDefined)
+    assert(schema1.isDefined)
+    assert(schema0.get != schema1.get)
+    assert(schema != schema0.get)
+    assert(schema != schema1.get)
+  }
+
+  it should "generate all schemas for all the annotated classes with @AvroSerde" in {
+    val reflections = new Reflections("it.agilelab.darwin.app.mock.classes")
+
+    val oneFieldSchema = ReflectData.get().getSchema(classOf[OneField]).toString
+    val myNestedSchema = ReflectData.get().getSchema(classOf[MyNestedClass]).toString
+    val myClassSchema = ReflectData.get().getSchema(classOf[MyClass]).toString
+
+    val annotationClass: Class[AvroSerde] = classOf[AvroSerde]
+    val classes = reflections.getTypesAnnotatedWith(annotationClass).asScala.toSeq
+      .filter(c => !c.isInterface && !Modifier.isAbstract(c.getModifiers))
+    classes.foreach(println)
+    val schemas = classes.map(c => ReflectData.get().getSchema(Class.forName(c.getName)).toString)
+    schemas.foreach(println)
+    Seq(oneFieldSchema, myClassSchema, myNestedSchema) should contain theSameElementsAs schemas
+  }
+
+  it should "reload all schemas from the connector" in {
+    val newSchema = ReflectData.get().getSchema(classOf[NewClass])
+    val newId = SchemaNormalization.parsingFingerprint64(newSchema)
+    assert(manager.getSchema(newId).isEmpty)
+
+    manager.connector.insert(Seq(newId -> newSchema))
+    assert(manager.getSchema(newId).isDefined)
+    assert(manager.getSchema(newId).get == newSchema)
+  }
+}
diff --git a/mock-application/src/test/scala/it/agilelab/darwin/app/mock/LazyApplicationSuite.scala b/mock-application/src/test/scala/it/agilelab/darwin/app/mock/LazyApplicationSuite.scala
diff --git a/mock-connector/src/main/scala/it/agilelab/darwin/connector/mock/MockConnector.scala b/mock-connector/src/main/scala/it/agilelab/darwin/connector/mock/MockConnector.scala
diff --git a/mongo/src/main/scala/it/agilelab/darwin/connector/mongo/MongoConnector.scala b/mongo/src/main/scala/it/agilelab/darwin/connector/mongo/MongoConnector.scala
diff --git a/postgres/src/main/scala/it/agilelab/darwin/connector/postgres/PostgresConnector.scala b/postgres/src/main/scala/it/agilelab/darwin/connector/postgres/PostgresConnector.scala
diff --git a/rest/src/main/scala/it/agilelab/darwin/connector/rest/RestConnector.scala b/rest/src/main/scala/it/agilelab/darwin/connector/rest/RestConnector.scala

Original file line number	Diff line number	Diff line change
`@@ -26,4 +26,6 @@ abstract class Connector(config: Config) extends Serializable {`
`26`	`26`	`* @param schemas a sequence of pairs (ID, schema) Schema entities to insert in the storage.`
`27`	`27`	`*/`
`28`	`28`	`def insert(schemas: Seq[(Long, Schema)]): Unit`
	`29`	`+`
	`30`	`+ def findSchema(id: Long): Option[Schema]`
`29`	`31`	`}`
Original file line number	Diff line number	Diff line change
`@@ -10,7 +10,5 @@ import org.apache.avro.Schema`
`10`	`10`	`*/`
`11`	`11`	`case class CachedEagerAvroSchemaManager(override val config: Config) extends CachedAvroSchemaManager {`
`12`	`12`
`13`		`- override def getId(schema: Schema): Long = cache.getId(schema)`
`14`		`-`
`15`		`- override def getSchema(id: Long): Schema = cache.getSchema(id)`
	`13`	`+ override def getSchema(id: Long): Option[Schema] = cache.getSchema(id)`
`16`	`14`	`}`
Original file line number	Diff line number	Diff line change
`@@ -125,6 +125,8 @@ case class HBaseConnector(config: Config) extends Connector(config) with Logging`
`125`	`125`	`mutator.flush()`
`126`	`126`	`log.debug(s"insertion of schemas into $NAMESPACE_STRING:$TABLE_NAME_STRING successful")`
`127`	`127`	`}`
	`128`	`+`
	`129`	`+ override def findSchema(id: Long): Option[Schema] = ???`
`128`	`130`	`}`
`129`	`131`
`130`	`132`
Original file line number	Diff line number	Diff line change
`@@ -8,4 +8,6 @@ class IgniteConnector(config: Config) extends Connector(config) {`
`8`	`8`	`override def fullLoad(): Seq[(Long, Schema)] = ???`
`9`	`9`
`10`	`10`	`override def insert(schemas: Seq[(Long, Schema)]): Unit = ???`
	`11`	`+`
	`12`	`+ override def findSchema(id: Long): Option[Schema] = ???`
`11`	`13`	`}`