Merge branch 'develop' into feature/19

Author: Antonio Murgia

common/src/main/scala/it/agilelab/darwin/common/Connector.scala

@@ -9,10 +9,27 @@ import org.apache.avro.Schema
   */
 abstract class Connector(config: Config) extends Serializable {
 
+  /**
+    * Creates the configured table, if the table already exists, does nothing
+    */
+  def createTable(): Unit
+
+  /**
+    * Returns whether or not the configured table exists
+    */
+  def tableExists(): Boolean
+
+  /**
+    *
+    * @return a message indicating the user what he/she should do to create the table him/herself
+    */
+  def tableCreationHint(): String
+
   /**
     * Loads all schemas found on the storage.
     * This method can be invoked multiple times: to initialize the initial values or to update the existing ones with
     * the new data found on the storage.
+    *
     * @return a sequence of all the pairs (ID, schema) found on the storage
     */
   def fullLoad(): Seq[(Long, Schema)]

common/src/main/scala/it/agilelab/darwin/common/package.scala

@@ -0,0 +1,12 @@
+package it.agilelab.darwin
+
+package object common {
+
+  def using[A <: AutoCloseable, B](closeable: A)(f: A => B): B = {
+    try {
+      f(closeable)
+    } finally {
+      closeable.close()
+    }
+  }
+}

core/src/main/resources/reference.conf

@@ -0,0 +1 @@
+createTable: false

core/src/main/scala/it/agilelab/darwin/manager/AvroSchemaManager.scala

@@ -3,6 +3,7 @@ package it.agilelab.darwin.manager
 import com.typesafe.config.Config
 import it.agilelab.darwin.common.{Connector, ConnectorFactory, Logging}
 import it.agilelab.darwin.manager.exception.ConnectorNotFoundException
+import it.agilelab.darwin.manager.util.ConfigurationKeys
 import jdk.nashorn.internal.runtime.ParserException
 import org.apache.avro.{Schema, SchemaNormalization}
 import it.agilelab.darwin.manager.util.ByteArrayUtils._
@@ -20,9 +21,20 @@ trait AvroSchemaManager extends Logging {
 
   protected def config: Config
 
-  protected[darwin] lazy val connector: Connector = ConnectorFactory.creators().headOption.map(_.create(config))
+  protected[darwin] lazy val connector: Connector = {
+    val cnt = ConnectorFactory.creators().headOption.map(_.create(config))
       .getOrElse(throw new ConnectorNotFoundException(config))
 
+    if (config.getBoolean(ConfigurationKeys.CREATE_TABLE)) {
+      cnt.createTable()
+    } else if (!cnt.tableExists()) {
+      log.warn(s"Darwin table does not exists and has not been created (${ConfigurationKeys.CREATE_TABLE} was false)")
+      log.warn(cnt.tableCreationHint())
+    }
+    cnt
+  }
+
+
   /**
     * Extracts the ID from a Schema.
     *

core/src/main/scala/it/agilelab/darwin/manager/util/ConfigurationKeys.scala

@@ -0,0 +1,7 @@
+package it.agilelab.darwin.manager.util
+
+object ConfigurationKeys {
+
+  val CREATE_TABLE = "createTable"
+
+}

hbase/src/main/scala/it/agilelab/darwin/connector/hbase/HBaseConnector.scala

@@ -1,7 +1,7 @@
 package it.agilelab.darwin.connector.hbase
 
 import com.typesafe.config.Config
-import it.agilelab.darwin.common.{Connector, Logging}
+import it.agilelab.darwin.common.{Connector, Logging, using}
 import org.apache.avro.Schema
 import org.apache.avro.Schema.Parser
 import org.apache.commons.io.IOUtils
@@ -10,7 +10,7 @@ import org.apache.hadoop.fs.Path
 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}
+import org.apache.hadoop.hbase._
 import org.apache.hadoop.security.UserGroupInformation
 
 import scala.collection.JavaConverters._
@@ -51,8 +51,8 @@ case class HBaseConnector(config: Config) extends Connector(config) with Logging
 
   lazy val TABLE_NAME: TableName = TableName.valueOf(Bytes.toBytes(NAMESPACE_STRING), Bytes.toBytes(TABLE_NAME_STRING))
 
-  lazy val CF: Array[Byte] = Bytes.toBytes("0")
-  lazy val QUALIFIER: Array[Byte] = Bytes.toBytes("schema")
+  val CF: Array[Byte] = Bytes.toBytes("0")
+  val QUALIFIER: Array[Byte] = Bytes.toBytes("schema")
 
   log.debug("Creating default HBaseConfiguration")
   val configuration: Configuration = HBaseConfiguration.create()
@@ -126,6 +126,31 @@ 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 createTable(): Unit = {
+    using(connection.getAdmin) { admin =>
+      if (!admin.listNamespaceDescriptors().exists(_.getName == NAMESPACE_STRING)) {
+        log.info(s"Namespace $NAMESPACE_STRING does not exists, creating it")
+        admin.createNamespace(NamespaceDescriptor.create(NAMESPACE_STRING).build())
+      }
+      if (!tableExists()) {
+        log.info(s"Table $TABLE_NAME does not exists, creating it")
+        admin.createTable(new HTableDescriptor(TABLE_NAME).addFamily(new HColumnDescriptor(CF)))
+      }
+    }
+  }
+
+  override def tableExists(): Boolean = {
+    using(connection.getAdmin) { admin =>
+      admin.tableExists(TABLE_NAME)
+    }
+  }
+
+  override def tableCreationHint(): String = {
+    s"""To create namespace and table from an HBase shell issue:
+       |  create_namespace '$NAMESPACE_STRING'
+       |  create '$NAMESPACE_STRING:$TABLE_NAME_STRING', '0'""".stripMargin
+  }
+
   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))

hbase/src/test/scala/it/agilelab/darwin/connector/hbase/HBaseConnectorSuite.scala

@@ -2,11 +2,11 @@ package it.agilelab.darwin.connector.hbase
 
 import com.typesafe.config.ConfigFactory
 import it.agilelab.darwin.common.Connector
-import org.apache.avro.{Schema, SchemaNormalization}
 import org.apache.avro.reflect.ReflectData
-import org.scalatest.{FlatSpec, Matchers}
+import org.apache.avro.{Schema, SchemaNormalization}
+import org.scalatest.{BeforeAndAfterAll, FlatSpec, Matchers}
 
-class HBaseConnectorSuite extends FlatSpec with Matchers {
+class HBaseConnectorSuite extends FlatSpec with Matchers with BeforeAndAfterAll {
 
   val connector: Connector = new HBaseConnectorCreator().create(ConfigFactory.load())
 
@@ -28,4 +28,19 @@ class HBaseConnectorSuite extends FlatSpec with Matchers {
     assert(noSchema.isEmpty)
   }
 
+  "connector.tableCreationHint" should "print the correct hint for table creation" in {
+    connector.tableCreationHint() should be(
+      """To create namespace and table from an HBase shell issue:
+        |  create_namespace 'AVRO'
+        |  create 'AVRO:SCHEMA_REPOSITORY', '0'""".stripMargin)
+  }
+
+  "connector.tableExists" should "return true with existent table" in {
+    connector.tableExists() should be(true)
+  }
+
+  override protected def beforeAll(): Unit = {
+    super.beforeAll()
+    connector.createTable()
+  }
 }

ignite/src/main/scala/it/agilelab/darwin/connector/ignite/IgniteConnector.scala

@@ -10,4 +10,10 @@ class IgniteConnector(config: Config) extends Connector(config) {
   override def insert(schemas: Seq[(Long, Schema)]): Unit = ???
 
   override def findSchema(id: Long): Option[Schema] = ???
+
+  override def createTable(): Unit = ???
+
+  override def tableExists(): Boolean = ???
+
+  override def tableCreationHint(): String = ???
 }

mock-connector/src/main/scala/it/agilelab/darwin/connector/mock/MockConnector.scala

@@ -28,4 +28,10 @@ class MockConnector(config: Config) extends Connector(config) {
   }
 
   override def findSchema(id: Long): Option[Schema] = table.get(id)
+
+  override def createTable(): Unit = Unit
+
+  override def tableExists(): Boolean = true
+
+  override def tableCreationHint(): String = ""
 }

mongo/src/main/scala/it/agilelab/darwin/connector/mongo/MongoConnector.scala

@@ -9,5 +9,11 @@ class MongoConnector(config: Config) extends Connector(config) {
 
   override def insert(schemas: Seq[(Long, Schema)]): Unit = ???
 
+  override def createTable(): Unit = ???
+
+  override def tableExists(): Boolean = ???
+
+  override def tableCreationHint(): String = ???
+
   override def findSchema(id: Long): Option[Schema] = ???
 }