vli02 commented on code in PR #4874:
URL: https://github.com/apache/hbase/pull/4874#discussion_r1020625492
##########
hbase-client/src/main/java/org/apache/hadoop/hbase/client/AsyncConnection.java:
##########
@@ -34,6 +34,16 @@
@InterfaceAudience.Public
public interface AsyncConnection extends Closeable {
+ /**
+ * Returns the clusterId of this connection.
+ */
+ String getClusterId2();
+
+ /**
+ * Returns this connection identity.
+ */
+ String getIdentity();
Review Comment:
I use connScope now.
##########
hbase-client/src/main/java/org/apache/hadoop/hbase/client/MetricsConnection.java:
##########
@@ -334,27 +381,48 @@ public Counter newMetric(Class<?> clazz, String name,
String scope) {
MetricsConnection(String scope, Supplier<ThreadPoolExecutor> batchPool,
Supplier<ThreadPoolExecutor> metaPool) {
this.scope = scope;
+ this.batchPools.add(batchPool);
Review Comment:
You are right, the caller in AsyncConnectionImpl.java always pass a null:
```
this.metrics =
Optional.of(MetricsConnection.getMetricsConnection(this, () -> null,
() -> null));
```
##########
hbase-client/src/main/java/org/apache/hadoop/hbase/client/ConnectionFactory.java:
##########
@@ -225,7 +256,7 @@ public static Connection createConnection(Configuration
conf, ExecutorService po
clazz.getDeclaredConstructor(Configuration.class,
ExecutorService.class, User.class);
constructor.setAccessible(true);
return user.runAs((PrivilegedExceptionAction<
- Connection>) () -> (Connection) constructor.newInstance(conf, pool,
user));
+ Connection>) () -> (Connection) constructor.newInstance(conf, pool,
user, identity));
Review Comment:
Your are right, there should no need to change here. Updated.
##########
hbase-client/src/main/java/org/apache/hadoop/hbase/client/MetricsConnection.java:
##########
@@ -295,8 +337,13 @@ public Counter newMetric(Class<?> clazz, String name,
String scope) {
}
};
+ // List of thread pool per connection of the metrics.
+ private List batchPools = new ArrayList<Supplier>();
+ private List metaPools = new ArrayList<Supplier>();
Review Comment:
Good suggestion, updated, thanks!
##########
hbase-client/src/main/java/org/apache/hadoop/hbase/client/AsyncConnection.java:
##########
@@ -34,6 +34,16 @@
@InterfaceAudience.Public
public interface AsyncConnection extends Closeable {
+ /**
+ * Returns the clusterId of this connection.
+ */
+ String getClusterId2();
Review Comment:
Updated.
##########
hbase-client/src/main/java/org/apache/hadoop/hbase/client/MetricsConnection.java:
##########
@@ -457,6 +525,27 @@ public void incrementServerOverloadedBackoffTime(long
time, TimeUnit timeUnit) {
overloadedBackoffTimer.update(time, timeUnit);
}
+ /** Return the connection count of the metrics within a scope */
+ private long getConnectionCount() {
+ return connectionCount.getCount();
Review Comment:
will do. thanks
##########
hbase-client/src/main/java/org/apache/hadoop/hbase/client/MetricsConnection.java:
##########
@@ -334,27 +381,48 @@ public Counter newMetric(Class<?> clazz, String name,
String scope) {
MetricsConnection(String scope, Supplier<ThreadPoolExecutor> batchPool,
Supplier<ThreadPoolExecutor> metaPool) {
this.scope = scope;
+ this.batchPools.add(batchPool);
+ this.metaPools.add(metaPool);
this.registry = new MetricRegistry();
this.registry.register(getExecutorPoolName(), new RatioGauge() {
@Override
protected Ratio getRatio() {
- ThreadPoolExecutor pool = batchPool.get();
- if (pool == null) {
- return Ratio.of(0, 0);
+ int numerator = 0;
Review Comment:
comments added, I am returning the max ratio of the pool among pools in all
connections.
##########
hbase-client/src/main/java/org/apache/hadoop/hbase/client/ConnectionFactory.java:
##########
@@ -216,6 +216,37 @@ public static Connection createConnection(Configuration
conf, User user) throws
*/
public static Connection createConnection(Configuration conf,
ExecutorService pool,
final User user) throws IOException {
+ return createConnection(conf, pool, user, null);
+ }
+
+ /**
+ * Create a new Connection instance using the passed <code>conf</code>
instance. Connection
+ * encapsulates all housekeeping for a connection to the cluster. All tables
and interfaces
+ * created from returned connection share zookeeper connection, meta cache,
and connections to
+ * region servers and masters. <br>
+ * The caller is responsible for calling {@link Connection#close()} on the
returned connection
+ * instance. Typical usage:
+ *
+ * <pre>
+ * Connection connection = ConnectionFactory.createConnection(conf);
+ * Table table = connection.getTable(TableName.valueOf("table1"));
+ * try {
+ * table.get(...);
+ * ...
+ * } finally {
+ * table.close();
+ * connection.close();
+ * }
+ * </pre>
+ *
+ * @param conf configuration
+ * @param user the user the connection is for
+ * @param pool the thread pool to use for batch operations
+ * @param identity the identity of the metrics for this connection.
Review Comment:
updated with connScope.
##########
hbase-client/src/main/java/org/apache/hadoop/hbase/client/MetricsConnection.java:
##########
@@ -334,27 +381,48 @@ public Counter newMetric(Class<?> clazz, String name,
String scope) {
MetricsConnection(String scope, Supplier<ThreadPoolExecutor> batchPool,
Supplier<ThreadPoolExecutor> metaPool) {
this.scope = scope;
+ this.batchPools.add(batchPool);
+ this.metaPools.add(metaPool);
this.registry = new MetricRegistry();
this.registry.register(getExecutorPoolName(), new RatioGauge() {
@Override
protected Ratio getRatio() {
- ThreadPoolExecutor pool = batchPool.get();
- if (pool == null) {
- return Ratio.of(0, 0);
+ int numerator = 0;
+ int denominator = 0;
+ for (int i = 0; i < batchPools.size(); i++) {
+ ThreadPoolExecutor pool = (ThreadPoolExecutor) ((Supplier)
batchPools.get(i)).get();
Review Comment:
updated, thanks!
##########
hbase-client/src/main/java/org/apache/hadoop/hbase/client/MetricsConnection.java:
##########
@@ -54,6 +58,38 @@
@InterfaceAudience.Private
public class MetricsConnection implements StatisticTrackable {
+ static final Map<String, MetricsConnection> METRICS_INSTANCES =
+ new HashMap<String, MetricsConnection>();
+
+ static MetricsConnection getMetricsConnection(final AsyncConnection conn,
+ Supplier<ThreadPoolExecutor> batchPool, Supplier<ThreadPoolExecutor>
metaPool) {
+ String scope = getScope(conn);
+ MetricsConnection metrics;
+ synchronized (METRICS_INSTANCES) {
+ metrics = METRICS_INSTANCES.get(scope);
+ if (metrics == null) {
+ metrics = new MetricsConnection(scope, batchPool, metaPool);
+ METRICS_INSTANCES.put(scope, metrics);
+ } else {
+ metrics.addThreadPools(batchPool, metaPool);
+ }
+ metrics.incrConnectionCount();
+ }
+ return metrics;
+ }
+
+ static void deleteMetricsConnection(final AsyncConnection conn) {
+ String scope = getScope(conn);
+ synchronized (METRICS_INSTANCES) {
+ MetricsConnection metrics = METRICS_INSTANCES.get(scope);
+ metrics.decrConnectionCount();
Review Comment:
Updated, no logs as the entire code in this class does not log anything.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
