incubator-gearpump git commit: GEARPUMP-164 configurable hbase user

[ma...@apache.org]

Repository: incubator-gearpump
Updated Branches:
  refs/heads/master 6f153e806 -> 2fe4abd72


GEARPUMP-164 configurable hbase user

Simple change that allows configuring the user for connection to HBase.
Tested on YARN deployment.

(PR replacing messy https://github.com/apache/incubator-gearpump/pull/44)

Author: karol brejna <ka...@intel.com>

Closes #49 from karol-brejna-i/GEARPUMP-164-hbase-user.


Project: http://git-wip-us.apache.org/repos/asf/incubator-gearpump/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-gearpump/commit/2fe4abd7
Tree: http://git-wip-us.apache.org/repos/asf/incubator-gearpump/tree/2fe4abd7
Diff: http://git-wip-us.apache.org/repos/asf/incubator-gearpump/diff/2fe4abd7

Branch: refs/heads/master
Commit: 2fe4abd72793b6c8af77893f47e728e92a6e04a5
Parents: 6f153e8
Author: karol brejna <ka...@intel.com>
Authored: Thu Jun 23 16:40:46 2016 +0800
Committer: manuzhang <ow...@gmail.com>
Committed: Thu Jun 23 16:40:46 2016 +0800

----------------------------------------------------------------------
 external/hbase/README.md                             | 10 +++++++++-
 .../apache/gearpump/external/hbase/HBaseSink.scala   | 15 +++++++++++++--
 2 files changed, 22 insertions(+), 3 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/incubator-gearpump/blob/2fe4abd7/external/hbase/README.md
----------------------------------------------------------------------
diff --git a/external/hbase/README.md b/external/hbase/README.md
index 0e55a5d..9d8e18a 100644
--- a/external/hbase/README.md
+++ b/external/hbase/README.md
@@ -26,8 +26,16 @@ The HBase cluster should run on where Gearpump is deployed.
 Suppose HBase is installed at ```/usr/lib/hbase``` on every node and you already have your application built into a jar file. 
 Then before submitting the application, you need to add HBase lib folder and conf folder into ```gearpump.executor.extraClasspath``` in ```conf/gear.conf```, for example ```/usr/lib/hbase/lib/*:/usr/lib/hbase/conf```. 
 Please note only client side's configuration change is needed. After that, you are able to submit the application.
+
+
+## If you need to supply the HBase user for the connection
+There are HBase configurations that have authorization configured (some users are allowed to read/write to selected namespaces/tables, some are not).
+
+In this cases you may need to configure the user that connects to HBase.
+
+When creating HBase Sink you can pass UserConfig object. If the object contains "hbase.user" property, the value will be used as the user name for HBase connection.
  
-## Working with Secured HBASE
+## Working with Kerberized HBase
 
 When the remote HBase is security enabled, a kerberos keytab and the corresponding principal name need to be
 provided for the gearpump-hbase connector. Specifically, the UserConfig object passed into the HBaseSink should contain

http://git-wip-us.apache.org/repos/asf/incubator-gearpump/blob/2fe4abd7/external/hbase/src/main/scala/org/apache/gearpump/external/hbase/HBaseSink.scala
----------------------------------------------------------------------
diff --git a/external/hbase/src/main/scala/org/apache/gearpump/external/hbase/HBaseSink.scala b/external/hbase/src/main/scala/org/apache/gearpump/external/hbase/HBaseSink.scala
index f3c5209..e4d5633 100644
--- a/external/hbase/src/main/scala/org/apache/gearpump/external/hbase/HBaseSink.scala
+++ b/external/hbase/src/main/scala/org/apache/gearpump/external/hbase/HBaseSink.scala
@@ -23,6 +23,7 @@ import org.apache.hadoop.conf.Configuration
 import org.apache.hadoop.hbase.client.{Connection, ConnectionFactory, Put}
 import org.apache.hadoop.hbase.util.Bytes
 import org.apache.hadoop.hbase.{HBaseConfiguration, TableName}
+import org.apache.hadoop.hbase.security.{User, UserProvider}
 import org.apache.hadoop.security.UserGroupInformation
 
 import org.apache.gearpump.Message
@@ -82,8 +83,8 @@ class HBaseSink(
   }
 
   def close(): Unit = {
-    connection.close()
     table.close()
+    connection.close()
   }
 
   /**
@@ -112,6 +113,7 @@ object HBaseSink {
   val TABLE_NAME = "hbase.table.name"
   val COLUMN_FAMILY = "hbase.table.column.family"
   val COLUMN_NAME = "hbase.table.column.name"
+  val HBASE_USER = "hbase.user"
 
   def apply[T](userconfig: UserConfig, tableName: String): HBaseSink = {
     new HBaseSink(userconfig, tableName)
@@ -142,6 +144,15 @@ object HBaseSink {
       UserGroupInformation.loginUserFromKeytab(principal.get, keytabFile.getAbsolutePath)
       keytabFile.delete()
     }
-    ConnectionFactory.createConnection(configuration)
+
+    val userName = userConfig.getString(HBASE_USER)
+    if (userName.isEmpty) {
+      ConnectionFactory.createConnection(configuration)
+    } else {
+      val user = UserProvider.instantiate(configuration)
+        .create(UserGroupInformation.createRemoteUser(userName.get))
+      ConnectionFactory.createConnection(configuration, user)
+    }
+
   }
 }