[#19] Cached and lazy strategies

Pirazzini Lorenzo · Pirazzini Lorenzo · commit 2b05e3f8adb6 · 2018-11-23T11:07:40.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
@@ -27,5 +27,11 @@ abstract class Connector(config: Config) extends Serializable {
     */
   def insert(schemas: Seq[(Long, Schema)]): Unit
 
+  /**
+    * Retrieves a single schema using its ID from the storage.
+    *
+    * @param id the ID of the schema
+    * @return an option that is empty if no schema was found for the ID or defined if a schema was found
+    */
   def findSchema(id: Long): Option[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
@@ -8,6 +8,11 @@ import org.apache.avro.{Schema, SchemaNormalization}
 import it.agilelab.darwin.manager.util.ByteArrayUtils._
 import scala.collection.JavaConverters._
 
+/**
+  * The main entry point of the Darwin library.
+  * An instance of AvroSchemaManager should ALWAYS be obtained through the AvroSchemaManagerFactory.
+  * The manager is responsible for schemas registration, retrieval and updates.
+  */
 trait AvroSchemaManager extends Logging {
   private val V1_HEADER = Array[Byte](0xC3.toByte, 0x01.toByte)
   private val ID_SIZE = 8
diff --git a/core/src/main/scala/it/agilelab/darwin/manager/AvroSchemaManagerFactory.scala b/core/src/main/scala/it/agilelab/darwin/manager/AvroSchemaManagerFactory.scala
@@ -5,18 +5,24 @@ import it.agilelab.darwin.common.Logging
 import it.agilelab.darwin.manager.exception.ConnectorNotFoundException
 import it.agilelab.darwin.manager.util.ConfigUtil
 
+/**
+  * Factory used to obtain the desired implementation of AvroSchemaManager.
+  * First of all the initialize method should be called passing the configuration (it will return an instance of
+  * AvroSchemaManager. Then, the same instance can be retrieved using the getInstance method without passing the
+  * configuration anymore.
+  */
 object AvroSchemaManagerFactory extends Logging {
 
   private var _instance: AvroSchemaManager = _
 
   /**
-    * Returns an instance of AvroSchemaManager that can be used to register schemas.
+    * Returns an instance of AvroSchemaManager that can be used to register and retrieve schemas.
     *
     * @param config the Config that is passed to the connector
     * @return an instance of AvroSchemaManager
     */
   @throws[ConnectorNotFoundException]
-  def getInstance(config: Config): AvroSchemaManager = {
+  def initialize(config: Config): AvroSchemaManager = {
     synchronized {
       if (_instance == null) {
         log.debug("creating instance of AvroSchemaManager")
@@ -33,6 +39,12 @@ object AvroSchemaManagerFactory extends Logging {
     }
   }
 
+  /**
+    * Returns the initialized instance of AvroSchemaManager that can be used to register and retrieve schemas.
+    * The instance must be created once using the initialize method passing a configuration before calling this method.
+    *
+    * @return the initialized instance of AvroSchemaManager
+    */
   def getInstance: AvroSchemaManager = {
     synchronized {
       if (_instance == null) {
diff --git a/core/src/main/scala/it/agilelab/darwin/manager/CachedAvroSchemaManager.scala b/core/src/main/scala/it/agilelab/darwin/manager/CachedAvroSchemaManager.scala
@@ -3,6 +3,10 @@ package it.agilelab.darwin.manager
 import java.util.concurrent.atomic.AtomicReference
 import org.apache.avro.Schema
 
+/**
+  * Implementation of AvroSchemaManager that defines a cache where the storage data is loaded, in order to reduce the
+  * number of accesses to the storage.
+  */
 trait CachedAvroSchemaManager extends AvroSchemaManager {
   protected val _cache: AtomicReference[Option[AvroSchemaCache]] = new AtomicReference[Option[AvroSchemaCache]](None)
 
@@ -23,7 +27,7 @@ trait CachedAvroSchemaManager extends AvroSchemaManager {
     * Throws an exception if the cache wasn't already loaded (the getInstance method must always be used to
     * initialize the cache using the required configuration).
     */
-  def reload(): AvroSchemaManager = {
+  override def reload(): AvroSchemaManager = {
     log.debug("reloading cache...")
     _cache.set(Some(AvroSchemaCacheFingerprint(connector.fullLoad())))
     log.debug("cache reloaded")
diff --git a/core/src/main/scala/it/agilelab/darwin/manager/CachedEagerAvroSchemaManager.scala b/core/src/main/scala/it/agilelab/darwin/manager/CachedEagerAvroSchemaManager.scala
@@ -4,11 +4,10 @@ import com.typesafe.config.Config
 import org.apache.avro.Schema
 
 /**
-  * The main entry point to this library.
-  * N.B.: each method all on this object must always be AFTER the initialization, performed invoking the initialize
-  * method.
+  * Implementation of CachedAvroSchemaManager that loads all the schemas into the cache at startup and doesn't
+  * perform any other accesses to the storage: each retrieve is performed onto the cache.
   */
-case class CachedEagerAvroSchemaManager(override val config: Config) extends CachedAvroSchemaManager {
+case class CachedEagerAvroSchemaManager private[darwin](override val config: Config) extends CachedAvroSchemaManager {
 
   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
@@ -2,7 +2,11 @@ package it.agilelab.darwin.manager
 import com.typesafe.config.Config
 import org.apache.avro.Schema
 
-case class CachedLazyAvroSchemaManager(override val config: Config) extends CachedAvroSchemaManager {
+/**
+  * Implementation of CachedAvroSchemaManager that loads all the schemas into the cache at startup and perform
+  * all the retrieves onto the cache; an access to the storage is performed only if there is a cache miss.
+  */
+case class CachedLazyAvroSchemaManager private[darwin](override val config: Config) extends CachedAvroSchemaManager {
 
   override def getSchema(id: Long): Option[Schema] = {
     cache.getSchema(id).orElse{
diff --git a/core/src/main/scala/it/agilelab/darwin/manager/LazyAvroSchemaManager.scala b/core/src/main/scala/it/agilelab/darwin/manager/LazyAvroSchemaManager.scala
@@ -3,7 +3,11 @@ package it.agilelab.darwin.manager
 import com.typesafe.config.Config
 import org.apache.avro.Schema
 
-case class LazyAvroSchemaManager(override val config: Config) extends AvroSchemaManager {
+/**
+  * Implementation of AvroSchemaManager that performs all the operations directly on the storage (retrievals and
+  * insertions).
+  */
+case class LazyAvroSchemaManager private[darwin](override val config: Config) extends AvroSchemaManager {
 
   override def getSchema(id: Long): Option[Schema] = connector.findSchema(id)
 
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
@@ -7,7 +7,7 @@ import org.apache.avro.Schema.Parser
 import org.apache.commons.io.IOUtils
 import org.apache.hadoop.conf.Configuration
 import org.apache.hadoop.fs.Path
-import org.apache.hadoop.hbase.client.{Connection, ConnectionFactory, Put, Result}
+import org.apache.hadoop.hbase.client.{Connection, ConnectionFactory, Get, Put, Result}
 import org.apache.hadoop.hbase.security.User
 import org.apache.hadoop.hbase.util.Bytes
 import org.apache.hadoop.hbase.{HBaseConfiguration, TableName}
@@ -126,8 +126,14 @@ case class HBaseConnector(config: Config) extends Connector(config) with Logging
     log.debug(s"insertion of schemas into $NAMESPACE_STRING:$TABLE_NAME_STRING successful")
   }
 
-  override def findSchema(id: Long): Option[Schema] = ???
+  override def findSchema(id: Long): Option[Schema] = {
+    log.debug(s"loading a schema with id = $id from table $NAMESPACE_STRING:$TABLE_NAME_STRING")
+    val get: Get = new Get(Bytes.toBytes(id))
+    get.addColumn(CF, QUALIFIER)
+    val result: Result = connection.getTable(TABLE_NAME).get(get)
+    val value: Option[Array[Byte]] = Option(result.getValue(CF, QUALIFIER))
+    val schema: Option[Schema] = value.map(v => parser.parse(Bytes.toString(v)))
+    log.debug(s"$schema loaded from HBase for id = $id")
+    schema
+  }
 }
-
-
-
diff --git a/hbase/src/test/scala/it/agilelab/darwin/connector/hbase/HBaseConnectorSuite.scala b/hbase/src/test/scala/it/agilelab/darwin/connector/hbase/HBaseConnectorSuite.scala
@@ -21,6 +21,11 @@ class HBaseConnectorSuite extends FlatSpec with Matchers {
     val loaded: Seq[(Long, Schema)] = connector.fullLoad()
     assert(loaded.size == schemas.size)
     assert(loaded.forall(schemas.contains))
+    val schema = connector.findSchema(loaded.head._1)
+    assert(schema.isDefined)
+    assert(schema.get == loaded.head._2)
+    val noSchema = connector.findSchema(-1L)
+    assert(noSchema.isEmpty)
   }
 
 }
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
@@ -7,30 +7,33 @@
 import it.agilelab.darwin.manager.AvroSchemaManagerFactory;
 import org.apache.avro.Schema;
 import org.apache.avro.reflect.ReflectData;
+import org.junit.jupiter.api.Test;
 import org.reflections.Reflections;
 import scala.collection.JavaConversions;
 
-import java.util.ArrayList;
-import java.util.List;
-import java.util.Set;
+import java.util.*;
 
-class JavaApplicationTest {
 
-    AvroSchemaManager manager = AvroSchemaManagerFactory.getInstance(ConfigFactory.empty());
+class JavaApplicationTest {
 
+    @Test
     void mainTest() {
+        Map<String, Object> configMap = new HashMap<>();
+        configMap.put("type", "cached_eager");
+        AvroSchemaManager manager = AvroSchemaManagerFactory.initialize(ConfigFactory.parseMap(configMap));
+
         List<Schema> schemas = new ArrayList<>();
         Schema s = ReflectData.get().getSchema(OneField.class);
         schemas.add(s);
         manager.getSchema(0L);
-
-        AvroSchemaManagerFactory.getInstance(ConfigFactory.empty()).registerAll(JavaConversions.asScalaBuffer(schemas));
+        AvroSchemaManagerFactory.getInstance().registerAll(JavaConversions.asScalaBuffer(schemas));
 
         long id = manager.getId(schemas.get(0));
         assert(manager.getSchema(id).isDefined());
         assert (schemas.get(0) == manager.getSchema(id).get());
     }
 
+    @Test
     void reflectionTest() {
         Reflections reflections = new Reflections("it.agilelab.darwin.app.mock.classes");
 
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
@@ -16,7 +16,7 @@ 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)
+  val manager: AvroSchemaManager = AvroSchemaManagerFactory.initialize(config)
 
   "CachedEagerAvroSchemaManager" should "not fail after the initialization" in {
     val schemas: Seq[Schema] = Seq(new SchemaGenerator[MyNestedClass].schema)
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
@@ -16,7 +16,7 @@ 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)
+  val manager: AvroSchemaManager = AvroSchemaManagerFactory.initialize(config)
 
   "CachedLazyAvroSchemaManager" should "not fail after the initialization" in {
     val schemas: Seq[Schema] = Seq(new SchemaGenerator[MyNestedClass].schema)
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
@@ -16,7 +16,7 @@ import scala.collection.JavaConverters._
 class LazyApplicationSuite extends FlatSpec with Matchers {
 
   val config: Config = ConfigFactory.parseMap(Map("type" -> "lazy").asJava)
-  val manager: AvroSchemaManager = AvroSchemaManagerFactory.getInstance(config)
+  val manager: AvroSchemaManager = AvroSchemaManagerFactory.initialize(config)
 
   "LazyAvroSchemaManager" should "not fail after the initialization" in {
     val schemas: Seq[Schema] = Seq(new SchemaGenerator[MyNestedClass].schema)
diff --git a/mock-application/src/test/scala/it/agilelab/darwin/app/mock/ManagerUtilsSuite.scala b/mock-application/src/test/scala/it/agilelab/darwin/app/mock/ManagerUtilsSuite.scala
@@ -7,11 +7,14 @@ import it.agilelab.darwin.manager.util.ByteArrayUtils._
 import org.scalatest.{FlatSpec, Matchers}
 import scala.util.Random
 
+import scala.collection.JavaConverters._
+
 class ManagerUtilsSuite extends FlatSpec with Matchers {
 
   "AvroSchemaManager utilities" should "create a Single-Object encoded byte array" in {
     val originalSchema = new SchemaGenerator[OneField].schema
-    val manager = AvroSchemaManagerFactory.getInstance(ConfigFactory.empty)
+    val config = ConfigFactory.parseMap(Map("type" -> "cached_eager").asJava)
+    val manager = AvroSchemaManagerFactory.initialize(config)
     manager.registerAll(Seq(originalSchema))
     val originalPayload = new Array[Byte](10)
     Random.nextBytes(originalPayload)
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
@@ -32,7 +32,7 @@ class PostgresConnector(config: Config) extends Connector(config) with PostgresC
       val schema = parser.parse(resultSet.getString("schema"))
       schemas = schemas :+ (id -> schema)
     }
-    connection.close
+    connection.close()
     schemas
   }
 
@@ -47,17 +47,29 @@ class PostgresConnector(config: Config) extends Connector(config) with PostgresC
         insertSchemaPS.executeUpdate()
         insertSchemaPS.close()
       }
-      connection.commit
+      connection.commit()
     } catch {
-      case e: Exception => {
-        connection.rollback
+      case e: Exception =>
+        connection.rollback()
         // e.printStackTrace
         throw e // should re-throw?
-      }
     } finally {
-      connection.close
+      connection.close()
     }
   }
 
-  override def findSchema(id: Long): Option[Schema] = ???
+  override def findSchema(id: Long): Option[Schema] = {
+    val connection = getConnection
+    val statement = connection.prepareStatement(s"select * from $TABLE_NAME where id = ?")
+    statement.setLong(1, id)
+    val resultSet: ResultSet = statement.executeQuery()
+
+    val schema = if(resultSet.next()) {
+      Option(resultSet.getString("schema")).map(v => parser.parse(v))
+    } else {
+      None
+    }
+    connection.close()
+    schema
+  }
 }
diff --git a/spark-application/src/main/scala/it/agilelab/darwin/app/spark/SchemaManagerSparkApp.scala b/spark-application/src/main/scala/it/agilelab/darwin/app/spark/SchemaManagerSparkApp.scala
@@ -25,9 +25,9 @@ object SchemaManagerSparkApp extends GenericMainClass with SparkManager {
     val schemas = Seq(ReflectData.get().getSchema(classOf[Menu]), ReflectData.get().getSchema(classOf[MenuItem]),
       ReflectData.get().getSchema(classOf[Food]), ReflectData.get().getSchema(classOf[Order]),
       ReflectData.get().getSchema(classOf[Price]))
-    val manager = AvroSchemaManagerFactory.getInstance(ConfigFactory.load())
+    val manager = AvroSchemaManagerFactory.initialize(ConfigFactory.load())
     val registeredIDs: Seq[Long] = manager.registerAll(schemas).map(_._1)
-    ds.foreach(_ => AvroSchemaManagerFactory.getInstance(ConfigFactory.load()))
+    ds.foreach(_ => AvroSchemaManagerFactory.initialize(ConfigFactory.load()))
     mainLogger.info("Schemas registered")
 
     mainLogger.info("Getting ID for a schema")

Original file line number	Diff line number	Diff line change
`@@ -21,6 +21,11 @@ class HBaseConnectorSuite extends FlatSpec with Matchers {`
`21`	`21`	`val loaded: Seq[(Long, Schema)] = connector.fullLoad()`
`22`	`22`	`assert(loaded.size == schemas.size)`
`23`	`23`	`assert(loaded.forall(schemas.contains))`
	`24`	`+ val schema = connector.findSchema(loaded.head._1)`
	`25`	`+ assert(schema.isDefined)`
	`26`	`+ assert(schema.get == loaded.head._2)`
	`27`	`+ val noSchema = connector.findSchema(-1L)`
	`28`	`+ assert(noSchema.isEmpty)`
`24`	`29`	`}`
`25`	`30`
`26`	`31`	`}`