[#18] Make HBaseConnector and PostgreSql connector create tables automatically if set

Author: agilelab-tmnd1991

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/scala/it/agilelab/darwin/manager/AvroSchemaManager.scala

@@ -6,8 +6,10 @@ import java.util.concurrent.atomic.AtomicReference
 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
+
 import scala.collection.JavaConverters._
 
 object AvroSchemaManager extends Logging {
@@ -162,6 +164,12 @@ case class AvroSchemaManager(config: Config) extends Logging {
       .getOrElse(throw new ConnectorNotFoundException(config))
 
   private def initialize(): Unit = {
+    if (config.getBoolean(ConfigurationKeys.CREATE_TABLE)) {
+      connector.createTable()
+    } else if (!connector.tableExists()) {
+      log.warn(s"Darwin table does not exists and has not been created (${ConfigurationKeys.CREATE_TABLE} was false)")
+      log.warn(connector.tableCreationHint())
+    }
     log.debug("cache initialization...")
     AvroSchemaManager._cache.compareAndSet(None, Some(AvroSchemaCacheFingerprint(connector.fullLoad())))
     log.debug("cache initialized")

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, 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()
@@ -125,6 +125,31 @@ 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 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
+  }
 }
 
 

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())
 
@@ -23,4 +23,19 @@ class HBaseConnectorSuite extends FlatSpec with Matchers {
     assert(loaded.forall(schemas.contains))
   }
 
+  "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

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

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

@@ -26,4 +26,10 @@ class MockConnector(config: Config) extends Connector(config) {
     val p = new Schema.Parser()
     p.parse(getClass.getClassLoader.getResourceAsStream(path))
   }
+
+  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

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

postgres/src/main/scala/it/agilelab/darwin/connector/postgres/PostgresConnector.scala

@@ -3,7 +3,7 @@ package it.agilelab.darwin.connector.postgres
 import java.sql.ResultSet
 
 import com.typesafe.config.Config
-import it.agilelab.darwin.common.Connector
+import it.agilelab.darwin.common.{Connector, using}
 import org.apache.avro.Schema
 import org.apache.avro.Schema.Parser
 
@@ -21,6 +21,12 @@ class PostgresConnector(config: Config) extends Connector(config) with PostgresC
 
   setConnectionConfig(config)
 
+  private val CREATE_TABLE_STMT =
+    s"""CREATE TABLE IF NOT EXISTS $TABLE_NAME (
+       |id bigint NOT NULL PRIMARY KEY,
+       |schema text NOT NULL
+       |)""".stripMargin
+
   override def fullLoad(): Seq[(Long, Schema)] = {
     val connection = getConnection
     var schemas: Seq[(Long, Schema)] = Seq.empty[(Long, Schema)]
@@ -58,4 +64,18 @@ class PostgresConnector(config: Config) extends Connector(config) with PostgresC
       connection.close
     }
   }
+
+  override def createTable(): Unit = {
+    using(getConnection) { conn =>
+      conn.createStatement().executeUpdate(CREATE_TABLE_STMT)
+    }
+  }
+
+  override def tableExists(): Boolean = ???
+
+  override def tableCreationHint(): String = {
+    s"""To create table perform the following sql query:
+       |$CREATE_TABLE_STMT
+     """.stripMargin
+  }
 }