[#54] Create HBase2 connector (#77)

Avoid to duplicate HBase code as much as possible, this is obtained creating a folder hbase with all the common code and two separate sbt modules hbase1 and hbase2. Both modules include the code and resources placed under the hbase[1|2] folder + the code placed inside hbase folder. Older connector is called the same, while the new is called "darwin-hbase2-connector".

This made necessary changes to make.sh and publish.sh because sbt cannot run two HBase mini cluster in the same JVM without throwing weird errors.

Author: agilelab-tmnd1991

build.sbt

@@ -1,10 +1,13 @@
+import sbt.Keys.baseDirectory
+
 /*
  * Main build definition.
  *
  * See project/Settings.scala for the settings definitions.
  * See project/Dependencies.scala for the dependencies definitions.
  * See project/Versions.scala for the versions definitions.
  */
+
 lazy val Settings.pgpPass = Option(System.getenv().get("PGP_PASS")).map(_.toArray)
 lazy val root = Project("darwin", file("."))
   .settings(Settings.commonSettings: _*)
@@ -29,15 +32,30 @@ lazy val coreCommon = Project("darwin-core-common", file("common"))
   .settings(crossScalaVersions := Versions.crossScalaVersions)
   .enablePlugins(JavaAppPackaging)
 
-lazy val hbaseConnector = Project("darwin-hbase-connector", file("hbase"))
+lazy val hbaseConnector = Project("darwin-hbase-connector", file("hbase1"))
   .settings(Settings.commonSettings: _*)
   .dependsOn(coreCommon)
   .settings(pgpPassphrase := Settings.pgpPass)
   .settings(libraryDependencies ++= Dependencies.hbase_conn_dep)
   .settings(crossScalaVersions := Versions.crossScalaVersions)
+  .settings(Compile / unmanagedSourceDirectories += baseDirectory.value / ".." / "hbase" / "src" / "main" / "scala")
+  .settings(Test / unmanagedSourceDirectories += baseDirectory.value / ".." / "hbase" / "src" / "test" / "scala")
+  .settings(Test / unmanagedResourceDirectories += baseDirectory.value / ".." / "hbase" / "src" / "test" / "resources")
   .settings(Settings.hbaseTestSettings)
   .enablePlugins(JavaAppPackaging)
 
+lazy val hbaseConnector2 = Project("darwin-hbase2-connector", file("hbase2"))
+  .settings(Settings.commonSettings: _*)
+  .dependsOn(coreCommon)
+  .settings(pgpPassphrase := Settings.pgpPass)
+  .settings(libraryDependencies ++= Dependencies.hbase2_conn_dep)
+  .settings(crossScalaVersions := Versions.crossScalaVersions)
+  .settings(Compile / unmanagedSourceDirectories += baseDirectory.value / ".." / "hbase" / "src" / "main" / "scala")
+  .settings(Test / unmanagedSourceDirectories += baseDirectory.value / ".." / "hbase" / "src" / "test" / "scala")
+  .settings(Test / unmanagedResourceDirectories += baseDirectory.value / ".." / "hbase" / "src" / "test" / "resources")
+  .settings(Settings.hbase2TestSettings)
+  .enablePlugins(JavaAppPackaging)
+
 lazy val postgresConnector = Project("darwin-postgres-connector", file("postgres"))
   .settings(Settings.commonSettings: _*)
   .dependsOn(coreCommon)

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

@@ -1,18 +1,18 @@
 package it.agilelab.darwin.connector.hbase
 
 import com.typesafe.config.Config
-import it.agilelab.darwin.common.{using, Connector, Logging}
+import it.agilelab.darwin.common.compat._
+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
 import org.apache.hadoop.conf.Configuration
 import org.apache.hadoop.fs.Path
 import org.apache.hadoop.hbase._
-import org.apache.hadoop.hbase.client.{Connection, ConnectionFactory, Get, Put, Result}
+import org.apache.hadoop.hbase.client._
 import org.apache.hadoop.hbase.security.User
 import org.apache.hadoop.hbase.util.Bytes
 import org.apache.hadoop.security.UserGroupInformation
-import it.agilelab.darwin.common.compat._
 
 object HBaseConnector extends Logging {
 
@@ -49,7 +49,8 @@ case class HBaseConnector(config: Config) extends Connector with Logging {
 
   lazy val TABLE_NAME: TableName = TableName.valueOf(Bytes.toBytes(NAMESPACE_STRING), Bytes.toBytes(TABLE_NAME_STRING))
 
-  val CF: Array[Byte] = Bytes.toBytes("0")
+  val CF_STRING = "0"
+  val CF: Array[Byte] = Bytes.toBytes(CF_STRING)
   val QUALIFIER_SCHEMA: Array[Byte] = Bytes.toBytes("schema")
   val QUALIFIER_NAME: Array[Byte] = Bytes.toBytes("name")
   val QUALIFIER_NAMESPACE: Array[Byte] = Bytes.toBytes("namespace")
@@ -141,11 +142,13 @@ case class HBaseConnector(config: Config) extends Connector with Logging {
       }
       if (!tableExists()) {
         log.info(s"Table $TABLE_NAME does not exists, creating it")
-        admin.createTable(new HTableDescriptor(TABLE_NAME).addFamily(new HColumnDescriptor(CF)))
+        HBaseUtils.createTable(admin, TABLE_NAME, CF)
       }
     }
   }
 
+
+
   override def tableExists(): Boolean = {
     using(connection.getAdmin) { admin =>
       admin.tableExists(TABLE_NAME)
@@ -155,7 +158,7 @@ case class HBaseConnector(config: Config) extends Connector with Logging {
   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
+       |  create '$NAMESPACE_STRING:$TABLE_NAME_STRING', '$CF_STRING'""".stripMargin
   }
 
   override def findSchema(id: Long): Option[Schema] = {

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

@@ -1,19 +1,21 @@
 package it.agilelab.darwin.connector.hbase
 
 import java.nio.file.Files
+import java.util.UUID
 
 import com.typesafe.config.{ConfigFactory, ConfigValueFactory}
 import it.agilelab.darwin.common.Connector
 import org.apache.avro.reflect.ReflectData
 import org.apache.avro.{Schema, SchemaNormalization}
-import org.apache.hadoop.hbase.HBaseTestingUtility
+import org.apache.hadoop.hbase.{HBaseConfiguration, HBaseTestingUtility, MiniHBaseCluster}
 import org.scalatest.BeforeAndAfterAll
 import org.scalatest.flatspec.AnyFlatSpec
 import org.scalatest.matchers.should.Matchers
 
 class HBaseConnectorSuite extends AnyFlatSpec with Matchers with BeforeAndAfterAll {
 
-  var connector: Connector = _
+  private var connector: Connector = _
+  private var minicluster: MiniHBaseCluster = _
 
   "HBaseConnector" should "load all existing schemas" in {
     connector.fullLoad()
@@ -45,34 +47,32 @@ class HBaseConnectorSuite extends AnyFlatSpec with Matchers with BeforeAndAfterA
   }
 
   override def beforeAll(): Unit = {
-
-    connector = new HBaseConnectorCreator().create(HBaseConnectorSuite.config)
-
-    connector.createTable()
-  }
-
-
-}
-
-object HBaseConnectorSuite {
-  private lazy val config = {
-    val util = new HBaseTestingUtility()
-    val minicluster = util.startMiniCluster()
-
-    //Hbase connector can only load configurations from a file path so we need to render the hadoop conf
-    val confFile = Files.createTempFile("prefix", "suffix")
+    val testUUID = UUID.randomUUID().toString
+    val hConf = HBaseConfiguration.create()
+    hConf.set("test.build.data.basedirectory", s"./target/hbase-test-data-$testUUID")
+    val util = new HBaseTestingUtility(hConf)
+    minicluster = util.startMiniCluster(1, true)
+    val confFile = Files.createTempFile(testUUID, ".xml")
+    // Hbase connector can only load configurations from a file path so we need to render the hadoop conf
     val stream = Files.newOutputStream(confFile)
+    // mc.getConfiguration.writeXml(System.out)
     minicluster.getConfiguration.writeXml(stream)
     stream.flush()
     stream.close()
-    val hbaseConfigPath = ConfigValueFactory.fromAnyRef(confFile.toAbsolutePath.toString)
-
-    //HbaseConnector will only load conf if hbase-site and core-site are given,
-    //we give the same file to each.
+    // HbaseConnector will only load conf if hbase-site and core-site are given,
+    // we give the same file to each.
     sys.addShutdownHook(minicluster.shutdown())
-    ConfigFactory.load()
-      .withValue(ConfigurationKeys.HBASE_SITE, hbaseConfigPath)
-      .withValue(ConfigurationKeys.CORE_SITE, hbaseConfigPath)
+    val config = ConfigFactory.load()
+      .withValue(ConfigurationKeys.HBASE_SITE, ConfigValueFactory.fromAnyRef(confFile.toAbsolutePath.toString))
+      .withValue(ConfigurationKeys.CORE_SITE, ConfigValueFactory.fromAnyRef(confFile.toAbsolutePath.toString))
+    connector = new HBaseConnectorCreator().create(config)
+    connector.createTable()
   }
 
+  override def afterAll(): Unit = {
+    minicluster.shutdown()
+    minicluster.waitUntilShutDown()
+  }
+
+
 }

hbase/src/main/resources/META-INF/services/it.agilelab.darwin.common.ConnectorCreator renamed to hbase1/src/main/resources/META-INF/services/it.agilelab.darwin.common.ConnectorCreator

@@ -0,0 +1,10 @@
+package it.agilelab.darwin.connector.hbase
+
+import org.apache.hadoop.hbase.{HColumnDescriptor, HTableDescriptor, TableName}
+import org.apache.hadoop.hbase.client.Admin
+
+object HBaseUtils {
+  def createTable(admin: Admin, tableName: TableName, columnFamily: Array[Byte]): Unit = {
+    admin.createTable(new HTableDescriptor(tableName).addFamily(new HColumnDescriptor(columnFamily)))
+  }
+}

hbase1/src/main/scala/it/agilelab/darwin/connector/hbase/HBaseUtils.scala

@@ -0,0 +1 @@
+it.agilelab.darwin.connector.hbase.HBaseConnectorCreator

hbase2/src/main/resources/META-INF/services/it.agilelab.darwin.common.ConnectorCreator

@@ -0,0 +1,14 @@
+package it.agilelab.darwin.connector.hbase
+
+import org.apache.hadoop.hbase.TableName
+import org.apache.hadoop.hbase.client.{Admin, ColumnFamilyDescriptorBuilder, TableDescriptorBuilder}
+
+object HBaseUtils {
+  def createTable(admin: Admin, tableName: TableName, columnFamily: Array[Byte]): Unit = {
+    admin.createTable(
+      TableDescriptorBuilder.newBuilder(tableName)
+        .setColumnFamily(
+          ColumnFamilyDescriptorBuilder.newBuilder(columnFamily).build()
+        ).build())
+  }
+}

hbase2/src/main/scala/it/agilelab/darwin/connector/hbase/HBaseUtils.scala

@@ -1,2 +1,4 @@
 #!/bin/bash
-sbt clean scalastyle +test +doc
+set -ex
+sbt clean scalastyle +test +doc
+sbt darwin-hbase2-connector/clean darwin-hbase2-connector/scalastyle +darwin-hbase2-connector/test +darwin-hbase2-connector/doc

make.sh

@@ -3,8 +3,8 @@ import Keys._
 
 
 /**
-  * @author andreaL
-  */
+ * @author andreaL
+ */
 object Dependencies {
 
   lazy val scalatest = "org.scalatest" %% "scalatest" % "3.1.1" % "test"
@@ -13,7 +13,9 @@ object Dependencies {
   lazy val avro4s = "com.sksamuel.avro4s" %% "avro4s-core" % "1.8.3"
   lazy val hbase_server = "org.apache.hbase" % "hbase-server" % "1.2.2" % "provided"
   lazy val hbase_common = "org.apache.hbase" % "hbase-common" % "1.2.2" % "provided"
-  lazy val hadoop_common = "org.apache.hadoop" % "hadoop-common" % "2.6.0" % "provided"
+  lazy val hadoop_common = "org.apache.hadoop" % "hadoop-common" % "2.7.7" % "provided"
+  lazy val hbase2_server = "org.apache.hbase" % "hbase-server" % "2.1.10" % "provided"
+  lazy val hbase2_common = "org.apache.hbase" % "hbase-common" % "2.1.10" % "provided"
   lazy val reflections = "org.reflections" % "reflections" % "0.9.11" % Test
   lazy val spark_core = "org.apache.spark" %% "spark-core" % "2.4.5" % "provided"
   lazy val spark_sql = "org.apache.spark" %% "spark-sql" % "2.4.5" % "provided"
@@ -40,30 +42,48 @@ object Dependencies {
   //these jars are not resolved with default ivy behavior, also we need to enable in settings
   //the resolution of transitive dependencies for jars in test scope
   lazy val hbaseTestDependencies = Seq(
-    ("org.apache.hbase" % "hbase-testing-util" % "1.2.2").classifier("tests") % "test",
-    ("org.apache.hadoop" % "hadoop-common" % "2.6.0").classifier("tests") % "test" ,
-    ("org.apache.hbase" % "hbase-server" % "1.2.2").classifier("tests") % "test",
-    ("org.apache.hbase" % "hbase" % "1.2.2") % "test",
-    ("org.apache.hbase" % "hbase-hadoop-compat" % "1.2.2") % "test",
-    ("org.apache.hbase" % "hbase-hadoop-compat" % "1.2.2").classifier("tests") % "test",
-    ("org.apache.hbase" % "hbase-hadoop2-compat" % "1.2.2") % "test",
-    ("org.apache.hbase" % "hbase-hadoop2-compat" % "1.2.2").classifier("tests") % "test",
-    ("org.apache.hbase" % "hbase-common" % "1.2.2").classifier("tests") % "test",
-    ("org.apache.hbase" % "hbase" % "1.2.2").classifier("tests")  % "test" exclude("org.apache.hbase", "hbase"),
-    ("org.apache.hadoop" % "hadoop-hdfs" % "2.6.0").classifier("tests") % "test",
-    "org.apache.hadoop" % "hadoop-hdfs" % "2.6.0"% "test")
+    ("org.apache.hbase" % "hbase-testing-util" % "1.2.2").classifier("tests") % Test,
+    ("org.apache.hadoop" % "hadoop-common" % "2.7.7").classifier("tests") % Test,
+    ("org.apache.hbase" % "hbase-server" % "1.2.2").classifier("tests") % Test,
+    ("org.apache.hbase" % "hbase" % "1.2.2") % Test,
+    ("org.apache.hbase" % "hbase-hadoop-compat" % "1.2.2") % Test,
+    ("org.apache.hbase" % "hbase-hadoop-compat" % "1.2.2").classifier("tests") % Test,
+    ("org.apache.hbase" % "hbase-hadoop2-compat" % "1.2.2") % Test,
+    ("org.apache.hbase" % "hbase-hadoop2-compat" % "1.2.2").classifier("tests") % Test,
+    ("org.apache.hbase" % "hbase-common" % "1.2.2").classifier("tests") % Test,
+    ("org.apache.hbase" % "hbase" % "1.2.2").classifier("tests") % Test exclude("org.apache.hbase", "hbase"),
+    ("org.apache.hadoop" % "hadoop-hdfs" % "2.7.7").classifier("tests") % Test,
+    ("org.apache.hadoop" % "hadoop-hdfs" % "2.7.7") % Test)
+
+  lazy val hbase2TestDependencies = Seq(
+    ("org.apache.hbase" % "hbase-testing-util" % "2.1.10").classifier("tests") % Test,
+    ("org.apache.hadoop" % "hadoop-common" % "2.7.7").classifier("tests") % Test,
+    ("org.apache.hbase" % "hbase-server" % "2.1.10").classifier("tests") % Test,
+    ("org.apache.hbase" % "hbase" % "2.1.10") % Test,
+    ("org.apache.hbase" % "hbase-hadoop-compat" % "2.1.10") % Test,
+    ("org.apache.hbase" % "hbase-hadoop-compat" % "2.1.10").classifier("tests") % Test,
+    ("org.apache.hbase" % "hbase-hadoop2-compat" % "2.1.10") % Test,
+    ("org.apache.hbase" % "hbase-hadoop2-compat" % "2.1.10").classifier("tests") % Test,
+    ("org.apache.hbase" % "hbase-metrics" % "2.1.10") % Test,
+    ("org.apache.hbase" % "hbase-metrics-api" % "2.1.10") % Test,
+    ("org.apache.hbase" % "hbase-http" % "2.1.10") % Test,
+    ("org.apache.hbase" % "hbase-common" % "2.1.10").classifier("tests") % Test,
+    ("org.apache.hbase" % "hbase" % "2.1.10").classifier("tests") % Test exclude("org.apache.hbase", "hbase"),
+    ("org.apache.hadoop" % "hadoop-hdfs" % "2.7.7").classifier("tests") % Test,
+    ("org.apache.hadoop" % "hadoop-hdfs" % "2.7.7") % Test)
 
   lazy val httpClient = "org.scalaj" %% "scalaj-http" % "2.4.2"
 
   lazy val wireMock = Seq("com.github.tomakehurst" % "wiremock-jre8" % "2.21.0" % Test,
-                          "xmlunit" % "xmlunit" % "1.6" % Test)
+    "xmlunit" % "xmlunit" % "1.6" % Test)
 
 
   lazy val restServer = core_deps ++ Seq(logback) ++ akka
   lazy val core_deps = Seq(scalatest, avro, typesafe_config, junit)
   lazy val mock_app_dep = core_deps ++ Seq(reflections, hbase_common)
   lazy val mock_conn = core_deps ++ Seq(reflections)
   lazy val hbase_conn_dep = core_deps ++ Seq(hbase_common, hbase_server, hadoop_common)
+  lazy val hbase2_conn_dep = core_deps ++ Seq(hbase2_common, hbase2_server, hadoop_common)
   lazy val postgres_conn_dep = core_deps :+ postgres_conn :+ postgres_embedded
   lazy val spark_app = mock_app_dep ++ Seq(spark_core, spark_sql, hbase_common)
   lazy val mongo_conn = core_deps ++ Seq(mongo, mongoTest)

project/Dependencies.scala

@@ -95,6 +95,14 @@ object Settings {
     libraryDependencies ++= Dependencies.hbaseTestDependencies
   }
 
+  lazy val hbase2TestSettings: SettingsDefinition = {
+    //enable resolution of transitive dependencies of jars containing tests
+    //needed to run tests over hbase minicluster
+    transitiveClassifiers in Test := Seq(Artifact.TestsClassifier, Artifact.SourceClassifier)
+    libraryDependencies ++= Dependencies.hbase2TestDependencies
+  }
+
+
   lazy val notPublishSettings = Seq(skip in publish := true)
 
   lazy val myCredentials = Credentials(
@@ -116,6 +124,8 @@ object Settings {
 
   lazy val scalastyleSettings = Seq(scalastyleFailOnWarning := true)
 
+//  lazy val testSettings = Seq(parallelExecution in Test := false)
+
   lazy val publishSettings = Seq(
     publishTo := Some("bintray" at "https://api.bintray.com/maven/agile-lab-dev/Darwin/darwin/;publish=1"),
     credentials += myCredentials,