[pinpoint-apm#4088] Refactor hbase Connection creation

Xylus · Xylus · commit d8ea8afa2881 · 2018-05-09T16:08:11.000+09:00
diff --git a/collector/src/main/resources/applicationContext-hbase.xml b/collector/src/main/resources/applicationContext-hbase.xml
@@ -45,15 +45,19 @@
         <property name="preStartAllCoreThreads" value="true"/>
     </bean>
 
-    <bean id="connectionFactory" class="com.navercorp.pinpoint.common.hbase.PooledHTableFactory">
+    <bean id="connectionFactory" class="com.navercorp.pinpoint.common.hbase.ConnectionFactoryBean">
         <constructor-arg type="org.apache.hadoop.conf.Configuration" ref="hbaseConfiguration"/>
         <constructor-arg type="java.util.concurrent.ExecutorService" ref="hbaseThreadPool"/>
     </bean>
 
+    <bean id="hbaseTableFactory" class="com.navercorp.pinpoint.common.hbase.HbaseTableFactory">
+        <constructor-arg ref="connectionFactory"/>
+    </bean>
+
     <bean class="org.apache.hadoop.util.ShutdownHookManagerProxy"/>
 
     <bean id="asyncOperation" class="com.navercorp.pinpoint.common.hbase.HBaseAsyncOperationFactory" factory-method="create">
-        <constructor-arg type="org.apache.hadoop.hbase.client.Connection" value="#{connectionFactory.getConnection()}"/>
+        <constructor-arg type="org.apache.hadoop.hbase.client.Connection" ref="connectionFactory"/>
         <constructor-arg type="org.apache.hadoop.conf.Configuration" ref="hbaseConfiguration"/>
     </bean>
 
@@ -63,12 +67,16 @@
 
     <bean id="hbaseTemplate" class="com.navercorp.pinpoint.common.hbase.HbaseTemplate2" destroy-method="destroy">
         <property name="configuration" ref="hbaseConfiguration"/>
-        <property name="tableFactory" ref="connectionFactory"/>
+        <property name="tableFactory" ref="hbaseTableFactory"/>
         <property name="asyncOperation" ref="asyncOperation"/>
     </bean>
 
-    <bean id="hBaseAdminTemplate" class="com.navercorp.pinpoint.common.hbase.HBaseAdminTemplate" destroy-method="close">
-        <constructor-arg ref="hbaseConfiguration" index="0"/>
+    <bean id="hbaseAdminFactory" class="com.navercorp.pinpoint.common.hbase.HbaseAdminFactory">
+        <constructor-arg ref="connectionFactory"/>
+    </bean>
+
+    <bean id="hbaseAdminTemplate" class="com.navercorp.pinpoint.common.hbase.HBaseAdminTemplate">
+        <constructor-arg ref="hbaseAdminFactory"/>
     </bean>
 
     <bean id="applicationTraceIndexDistributor" class="com.sematext.hbase.wd.RowKeyDistributorByHashPrefix">
diff --git a/commons-hbase/src/main/java/com/navercorp/pinpoint/common/hbase/AdminFactory.java b/commons-hbase/src/main/java/com/navercorp/pinpoint/common/hbase/AdminFactory.java
@@ -0,0 +1,30 @@
+/*
+ * Copyright 2018 NAVER Corp.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.navercorp.pinpoint.common.hbase;
+
+import org.apache.hadoop.hbase.client.Admin;
+
+/**
+ * @author HyunGil Jeong
+ */
+public interface AdminFactory {
+
+    Admin getAdmin();
+
+    void releaseAdmin(Admin admin);
+
+}
diff --git a/commons-hbase/src/main/java/com/navercorp/pinpoint/common/hbase/ConnectionFactoryBean.java b/commons-hbase/src/main/java/com/navercorp/pinpoint/common/hbase/ConnectionFactoryBean.java
@@ -0,0 +1,88 @@
+/*
+ * Copyright 2018 NAVER Corp.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.navercorp.pinpoint.common.hbase;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hbase.client.Connection;
+import org.apache.hadoop.hbase.client.ConnectionFactory;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.DisposableBean;
+import org.springframework.beans.factory.FactoryBean;
+
+import java.io.IOException;
+import java.util.Objects;
+import java.util.concurrent.ExecutorService;
+
+/**
+ * @author HyunGil Jeong
+ */
+public class ConnectionFactoryBean implements FactoryBean<Connection>, DisposableBean {
+
+    private final Logger logger = LoggerFactory.getLogger(this.getClass());
+
+    private final Connection connection;
+
+    public ConnectionFactoryBean(Configuration configuration) {
+        Objects.requireNonNull(configuration, " must not be null");
+        try {
+            connection = ConnectionFactory.createConnection(configuration);
+        } catch (IOException e) {
+            throw new HbaseSystemException(e);
+        }
+    }
+
+    public ConnectionFactoryBean(Configuration configuration, ExecutorService executorService) {
+        Objects.requireNonNull(configuration, "configuration must not be null");
+        Objects.requireNonNull(executorService, "executorService must not be null");
+        try {
+            connection = ConnectionFactory.createConnection(configuration, executorService);
+        } catch (IOException e) {
+            throw new HbaseSystemException(e);
+        }
+    }
+
+    @Override
+    public Connection getObject() throws Exception {
+        return connection;
+    }
+
+    @Override
+    public Class<?> getObjectType() {
+        if (connection == null) {
+            return Connection.class;
+        }
+        return connection.getClass();
+    }
+
+    @Override
+    public boolean isSingleton() {
+        return true;
+    }
+
+    @Override
+    public void destroy() throws Exception {
+        logger.info("Hbase Connection destroy()");
+        if (connection != null) {
+            try {
+                connection.close();
+            } catch (IOException e) {
+                logger.warn("Hbase Connection.close() error: " + e.getMessage(), e);
+            }
+        }
+    }
+}
diff --git a/commons-hbase/src/main/java/com/navercorp/pinpoint/common/hbase/HBaseAdminTemplate.java b/commons-hbase/src/main/java/com/navercorp/pinpoint/common/hbase/HBaseAdminTemplate.java
@@ -17,79 +17,76 @@
 package com.navercorp.pinpoint.common.hbase;
 
 import java.io.IOException;
+import java.util.Objects;
 
-import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.hbase.HTableDescriptor;
 import org.apache.hadoop.hbase.TableName;
 import org.apache.hadoop.hbase.client.Admin;
-import org.apache.hadoop.hbase.client.Connection;
-import org.apache.hadoop.hbase.client.ConnectionFactory;
 
 /**
  * @author emeroad
+ * @author HyunGil Jeong
  */
 public class HBaseAdminTemplate {
 
-    private final Admin admin;
-    private final Connection connection;
+    private final AdminFactory adminFactory;
 
-    public HBaseAdminTemplate(Configuration configuration) {
-        try {
-            connection = ConnectionFactory.createConnection(configuration);
-            admin = connection.getAdmin();
-        } catch (Exception e) {
-            throw new HbaseSystemException(e);
-        }
+    public HBaseAdminTemplate(AdminFactory adminFactory) {
+        this.adminFactory = Objects.requireNonNull(adminFactory, "adminFactory must not be null");
     }
 
     public boolean createTableIfNotExist(HTableDescriptor htd) {
+        Admin admin = adminFactory.getAdmin();
         try {
-            if (!admin.tableExists(htd.getTableName())) {
-                this.admin.createTable(htd);
+            TableName tableName = htd.getTableName();
+            if (!admin.tableExists(tableName)) {
+                admin.createTable(htd);
                 return true;
             }
             return false;
         } catch (IOException e) {
             throw new HbaseSystemException(e);
+        } finally {
+            adminFactory.releaseAdmin(admin);
         }
     }
 
     public boolean tableExists(TableName tableName) {
+        Admin admin = adminFactory.getAdmin();
         try {
             return admin.tableExists(tableName);
         } catch (IOException e) {
             throw new HbaseSystemException(e);
+        } finally {
+            adminFactory.releaseAdmin(admin);
         }
     }
 
     public boolean dropTableIfExist(TableName tableName) {
+        Admin admin = adminFactory.getAdmin();
         try {
             if (admin.tableExists(tableName)) {
-                this.admin.disableTable(tableName);
-                this.admin.deleteTable(tableName);
+                admin.disableTable(tableName);
+                admin.deleteTable(tableName);
                 return true;
             }
             return false;
         } catch (IOException e) {
             throw new HbaseSystemException(e);
+        } finally {
+            adminFactory.releaseAdmin(admin);
         }
     }
 
     public void dropTable(TableName tableName) {
+        Admin admin = adminFactory.getAdmin();
         try {
-            this.admin.disableTable(tableName);
-            this.admin.deleteTable(tableName);
-        } catch (IOException e) {
-            throw new HbaseSystemException(e);
-        }
-    }
-
-    public void close() {
-        try {
-            this.admin.close();
-            this.connection.close();
+            admin.disableTable(tableName);
+            admin.deleteTable(tableName);
         } catch (IOException e) {
             throw new HbaseSystemException(e);
+        } finally {
+            adminFactory.releaseAdmin(admin);
         }
     }
 }
diff --git a/commons-hbase/src/main/java/com/navercorp/pinpoint/common/hbase/HbaseAdminFactory.java b/commons-hbase/src/main/java/com/navercorp/pinpoint/common/hbase/HbaseAdminFactory.java
@@ -0,0 +1,56 @@
+/*
+ * Copyright 2018 NAVER Corp.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.navercorp.pinpoint.common.hbase;
+
+import org.apache.hadoop.hbase.client.Admin;
+import org.apache.hadoop.hbase.client.Connection;
+
+import java.io.IOException;
+import java.util.Objects;
+
+/**
+ * @author HyunGil Jeong
+ */
+public class HbaseAdminFactory implements AdminFactory {
+
+    private final Connection connection;
+
+    public HbaseAdminFactory(Connection connection) {
+        this.connection = Objects.requireNonNull(connection, "connection must not be null");
+    }
+
+    @Override
+    public Admin getAdmin() {
+        try {
+            return connection.getAdmin();
+        } catch (IOException e) {
+            throw new HbaseSystemException(e);
+        }
+    }
+
+    @Override
+    public void releaseAdmin(Admin admin) {
+        if (admin == null) {
+            return;
+        }
+        try {
+            admin.close();
+        } catch (IOException e) {
+            throw new HbaseSystemException(e);
+        }
+    }
+}
diff --git a/commons-hbase/src/main/java/com/navercorp/pinpoint/common/hbase/HbaseTableFactory.java b/commons-hbase/src/main/java/com/navercorp/pinpoint/common/hbase/HbaseTableFactory.java
@@ -16,14 +16,9 @@
 
 package com.navercorp.pinpoint.common.hbase;
 
-import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.hbase.TableName;
 import org.apache.hadoop.hbase.client.Connection;
-import org.apache.hadoop.hbase.client.ConnectionFactory;
 import org.apache.hadoop.hbase.client.Table;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-import org.springframework.beans.factory.DisposableBean;
 
 import java.io.IOException;
 import java.util.Objects;
@@ -32,35 +27,21 @@
 /**
  * HTableInterfaceFactory based on HTablePool.
  * @author emeroad
- * @autor minwoo.jung
+ * @author minwoo.jung
+ * @author HyunGil Jeong
  */
-public class PooledHTableFactory implements TableFactory, DisposableBean {
+public class HbaseTableFactory implements TableFactory {
 
-    private final Logger logger = LoggerFactory.getLogger(this.getClass());
-
-    private final ExecutorService executor;
     private final Connection connection;
 
-    public PooledHTableFactory(Configuration config, ExecutorService executor) {
-        Objects.requireNonNull(config, "config must not be null");
-        this.executor = Objects.requireNonNull(executor, "executor must not be null");
-
-        try {
-            this.connection = ConnectionFactory.createConnection(config, executor);
-        } catch (IOException e) {
-            throw new HbaseSystemException(e);
-        }
-    }
-
-    public Connection getConnection() {
-        return connection;
+    public HbaseTableFactory(Connection connection) {
+        this.connection = Objects.requireNonNull(connection, "connection must not be null");
     }
 
-
     @Override
     public Table getTable(TableName tableName) {
         try {
-            return connection.getTable(tableName, executor);
+            return connection.getTable(tableName);
         } catch (IOException e) {
             throw new HbaseSystemException(e);
         }
@@ -87,18 +68,4 @@ public void releaseTable(Table table) {
             throw new HbaseSystemException(ex);
         }
     }
-
-
-    @Override
-    public void destroy() throws Exception {
-        logger.info("PooledHTableFactory.destroy()");
-        
-        if (connection != null) {
-            try {
-                this.connection.close();
-            } catch (IOException ex) {
-                logger.warn("Connection.close() error:" + ex.getMessage(), ex);
-            }
-        }
-    }
 }
diff --git a/commons-hbase/src/test/java/com/navercorp/pinpoint/common/hbase/HbaseTemplate2IT.java b/commons-hbase/src/test/java/com/navercorp/pinpoint/common/hbase/HbaseTemplate2IT.java
diff --git a/flink/src/main/resources/applicationContext-hbase.xml b/flink/src/main/resources/applicationContext-hbase.xml
diff --git a/web/src/main/resources/applicationContext-hbase.xml b/web/src/main/resources/applicationContext-hbase.xml
