diff --git a/hbase-client/pom.xml b/hbase-client/pom.xml index 425bd05..401e28e 100644 --- a/hbase-client/pom.xml +++ b/hbase-client/pom.xml @@ -189,6 +189,10 @@ log4j test + + com.yammer.metrics + metrics-core + diff --git a/hbase-client/src/main/java/org/apache/hadoop/hbase/client/ClusterConnection.java b/hbase-client/src/main/java/org/apache/hadoop/hbase/client/ClusterConnection.java index b3d99ae..99071fa 100644 --- a/hbase-client/src/main/java/org/apache/hadoop/hbase/client/ClusterConnection.java +++ b/hbase-client/src/main/java/org/apache/hadoop/hbase/client/ClusterConnection.java @@ -297,4 +297,9 @@ public interface ClusterConnection extends HConnection { */ ClientBackoffPolicy getBackoffPolicy(); + /** + * @return the MetricsConnection instance associated with this connection. + */ + public MetricsConnection getConnectionMetrics(); + } diff --git a/hbase-client/src/main/java/org/apache/hadoop/hbase/client/Connection.java b/hbase-client/src/main/java/org/apache/hadoop/hbase/client/Connection.java index dab4905..a3f6fe6 100644 --- a/hbase-client/src/main/java/org/apache/hadoop/hbase/client/Connection.java +++ b/hbase-client/src/main/java/org/apache/hadoop/hbase/client/Connection.java @@ -174,5 +174,4 @@ public interface Connection extends Abortable, Closeable { * @return true if this connection is closed */ boolean isClosed(); - } diff --git a/hbase-client/src/main/java/org/apache/hadoop/hbase/client/ConnectionImplementation.java b/hbase-client/src/main/java/org/apache/hadoop/hbase/client/ConnectionImplementation.java index ade32a8..92afbe5 100644 --- a/hbase-client/src/main/java/org/apache/hadoop/hbase/client/ConnectionImplementation.java +++ b/hbase-client/src/main/java/org/apache/hadoop/hbase/client/ConnectionImplementation.java @@ -121,6 +121,7 @@ class ConnectionImplementation implements ClusterConnection, Closeable { private final AsyncProcess asyncProcess; // single tracker per connection private final ServerStatisticTracker stats; + private final MetricsConnection metrics; private volatile boolean closed; private volatile boolean aborted; @@ -157,11 +158,11 @@ class ConnectionImplementation implements ClusterConnection, Closeable { // Client rpc instance. private RpcClient rpcClient; - private MetaCache metaCache = new MetaCache(); + private final MetaCache metaCache; private int refCount; - private User user; + protected User user; private RpcRetryingCallerFactory rpcCallerFactory; @@ -188,7 +189,7 @@ class ConnectionImplementation implements ClusterConnection, Closeable { this.registry = setupRegistry(); retrieveClusterId(); - this.rpcClient = RpcClientFactory.createClient(this.conf, this.clusterId); + this.rpcClient = RpcClientFactory.createClient(this.conf, this.clusterId, this.metrics); this.rpcControllerFactory = RpcControllerFactory.instantiate(conf); // Do we publish the status? @@ -239,11 +240,13 @@ class ConnectionImplementation implements ClusterConnection, Closeable { } else { nonceGenerator = new NoNonceGenerator(); } - stats = ServerStatisticTracker.create(conf); + this.stats = ServerStatisticTracker.create(conf); this.asyncProcess = createAsyncProcess(this.conf); this.interceptor = (new RetryingCallerInterceptorFactory(conf)).build(); this.rpcCallerFactory = RpcRetryingCallerFactory.instantiate(conf, interceptor, this.stats); this.backoffPolicy = ClientBackoffPolicyFactory.create(conf); + this.metrics = new MetricsConnection(this); + this.metaCache = new MetaCache(this.metrics); } /** @@ -365,6 +368,11 @@ class ConnectionImplementation implements ClusterConnection, Closeable { return new HBaseAdmin(this); } + @Override + public MetricsConnection getConnectionMetrics() { + return this.metrics; + } + private ExecutorService getBatchPool() { if (batchPool == null) { synchronized (this) { @@ -1320,7 +1328,7 @@ class ConnectionImplementation implements ClusterConnection, Closeable { // InetSocketAddress, and this way we will rightfully get a new stubKey. // Also, include the hostname in the key so as to take care of those cases where the // DNS name is different but IP address remains the same. - InetAddress i = new InetSocketAddress(rsHostname, port).getAddress(); + InetAddress i = new InetSocketAddress(rsHostname, port).getAddress(); String address = rsHostname; if (i != null) { address = i.getHostAddress() + "-" + rsHostname; diff --git a/hbase-client/src/main/java/org/apache/hadoop/hbase/client/MetaCache.java b/hbase-client/src/main/java/org/apache/hadoop/hbase/client/MetaCache.java index 8e1c93c..e763dd9 100644 --- a/hbase-client/src/main/java/org/apache/hadoop/hbase/client/MetaCache.java +++ b/hbase-client/src/main/java/org/apache/hadoop/hbase/client/MetaCache.java @@ -59,6 +59,12 @@ public class MetaCache { // The access to this attribute must be protected by a lock on cachedRegionLocations private final Set cachedServers = new ConcurrentSkipListSet(); + private final MetricsConnection metrics; + + public MetaCache(MetricsConnection metrics) { + this.metrics = metrics; + } + /** * Search the cache for a location that fits our table and row key. * Return null if no suitable region is located. @@ -74,6 +80,7 @@ public class MetaCache { Entry e = tableLocations.floorEntry(row); if (e == null) { + metrics.incrMetaCacheMiss(); return null; } RegionLocations possibleRegion = e.getValue(); @@ -94,10 +101,12 @@ public class MetaCache { // HConstants.EMPTY_END_ROW) check itself will pass. if (Bytes.equals(endKey, HConstants.EMPTY_END_ROW) || Bytes.compareTo(endKey, 0, endKey.length, row, 0, row.length) > 0) { + metrics.incrMetaCacheHit(); return possibleRegion; } // Passed all the way through, so we got nothing - complete cache miss + metrics.incrMetaCacheMiss(); return null; } diff --git a/hbase-client/src/main/java/org/apache/hadoop/hbase/client/MetricsConnection.java b/hbase-client/src/main/java/org/apache/hadoop/hbase/client/MetricsConnection.java new file mode 100644 index 0000000..62c118a --- /dev/null +++ b/hbase-client/src/main/java/org/apache/hadoop/hbase/client/MetricsConnection.java @@ -0,0 +1,145 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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 org.apache.hadoop.hbase.client; + +import com.google.protobuf.Descriptors; +import com.yammer.metrics.Metrics; +import com.yammer.metrics.core.Counter; +import com.yammer.metrics.core.Gauge; +import com.yammer.metrics.core.MetricsRegistry; +import com.yammer.metrics.reporting.JmxReporter; +import com.yammer.metrics.util.PercentGauge; +import com.yammer.metrics.util.RatioGauge; +import org.apache.hadoop.hbase.ServerName; +import org.apache.hadoop.hbase.classification.InterfaceAudience; + +import java.util.concurrent.ThreadPoolExecutor; +import java.util.concurrent.TimeUnit; + +/** + * This class is for maintaining the various connection statistics and publishing them through + * the metrics interfaces. + */ +@InterfaceAudience.Private +public class MetricsConnection { + + /** A container class for collecting details about the RPC call as it percolates. */ + public static class CallStats { + private long requestSizeBytes = 0; + private long responseSizeBytes = 0; + private long startTime = 0; + private long callTimeMs = 0; + + public long getRequestSizeBytes() { + return requestSizeBytes; + } + + public void setRequestSizeBytes(long requestSizeBytes) { + this.requestSizeBytes = requestSizeBytes; + } + + public long getResponseSizeBytes() { + return responseSizeBytes; + } + + public void setResponseSizeBytes(long responseSizeBytes) { + this.responseSizeBytes = responseSizeBytes; + } + + public long getStartTime() { + return startTime; + } + + public void setStartTime(long startTime) { + this.startTime = startTime; + } + + public long getCallTimeMs() { + return callTimeMs; + } + + public void setCallTimeMs(long callTimeMs) { + this.callTimeMs = callTimeMs; + } + } + + private final MetricsRegistry registry = Metrics.defaultRegistry(); + private final String scope; + + private final Counter metaCacheHits; + private final Counter metaCacheMisses; + + public MetricsConnection(final ConnectionImplementation conn) { + this.scope = conn.toString(); + final ThreadPoolExecutor batchPool = (ThreadPoolExecutor) conn.getCurrentBatchPool(); + final ThreadPoolExecutor metaPool = (ThreadPoolExecutor) conn.getCurrentMetaLookupPool(); + + this.registry.newGauge(this.getClass(), "executorPoolActiveThreads", scope, + new RatioGauge() { + @Override protected double getNumerator() { + return batchPool.getActiveCount(); + } + @Override protected double getDenominator() { + return batchPool.getMaximumPoolSize(); + } + }); + this.registry.newGauge(this.getClass(), "metaPoolActiveThreads", scope, + new RatioGauge() { + @Override protected double getNumerator() { + return metaPool.getActiveCount(); + } + @Override protected double getDenominator() { + return metaPool.getMaximumPoolSize(); + } + }); + this.metaCacheHits = registry.newCounter(this.getClass(), "metaCacheHits", scope); + this.metaCacheMisses = registry.newCounter(this.getClass(), "metaCacheMisses", scope); + JmxReporter.startDefault(this.registry); + } + + /** Produce an instance of {@link CallStats} for clients to attach to RPCs. */ + public static CallStats newCallStats() { + // TODO: instance pool to reduce GC? + return new CallStats(); + } + + /** Increment the number of meta cache hits. */ + public void incrMetaCacheHit() { + metaCacheHits.inc(); + } + + /** Increment the number of meta cache misses. */ + public void incrMetaCacheMiss() { + metaCacheMisses.inc(); + } + + /** Report RPC context to metrics system. */ + public void updateRpc(Descriptors.MethodDescriptor method, ServerName serverName, + CallStats stats) { + final String methodName = method.getService().getName() + "_" + method.getName(); + final String sn = serverName.toShortString().replace(':', '-'); + + // TODO: manage our own cache of references to these Histograms to avoid excess allocations + registry.newTimer(this.getClass(), "rpcCallDurationMs_" + methodName, scope) + .update(stats.getCallTimeMs(), TimeUnit.MILLISECONDS); + registry.newHistogram(this.getClass(), "rpcCallRequestSizeBytes_" + methodName, scope) + .update(stats.getRequestSizeBytes()); + registry.newHistogram(this.getClass(), "rpcCallResponseSizeBytes_" + methodName, scope) + .update(stats.getResponseSizeBytes()); + } +} diff --git a/hbase-client/src/main/java/org/apache/hadoop/hbase/ipc/AbstractRpcClient.java b/hbase-client/src/main/java/org/apache/hadoop/hbase/ipc/AbstractRpcClient.java index 9be370d..95cfc84 100644 --- a/hbase-client/src/main/java/org/apache/hadoop/hbase/ipc/AbstractRpcClient.java +++ b/hbase-client/src/main/java/org/apache/hadoop/hbase/ipc/AbstractRpcClient.java @@ -31,6 +31,7 @@ import org.apache.hadoop.hbase.CellScanner; import org.apache.hadoop.hbase.HConstants; import org.apache.hadoop.hbase.ServerName; import org.apache.hadoop.hbase.classification.InterfaceAudience; +import org.apache.hadoop.hbase.client.MetricsConnection; import org.apache.hadoop.hbase.codec.Codec; import org.apache.hadoop.hbase.codec.KeyValueCodec; import org.apache.hadoop.hbase.security.User; @@ -55,6 +56,7 @@ public abstract class AbstractRpcClient implements RpcClient { protected final Configuration conf; protected String clusterId; protected final SocketAddress localAddr; + protected final MetricsConnection metrics; protected UserProvider userProvider; protected final IPCUtil ipcUtil; @@ -79,8 +81,10 @@ public abstract class AbstractRpcClient implements RpcClient { * @param conf configuration * @param clusterId the cluster id * @param localAddr client socket bind address. + * @param metrics the connection metrics */ - public AbstractRpcClient(Configuration conf, String clusterId, SocketAddress localAddr) { + public AbstractRpcClient(Configuration conf, String clusterId, SocketAddress localAddr, + MetricsConnection metrics) { this.userProvider = UserProvider.instantiate(conf); this.localAddr = localAddr; this.tcpKeepAlive = conf.getBoolean("hbase.ipc.client.tcpkeepalive", true); @@ -100,6 +104,7 @@ public abstract class AbstractRpcClient implements RpcClient { this.connectTO = conf.getInt(SOCKET_TIMEOUT_CONNECT, DEFAULT_SOCKET_TIMEOUT_CONNECT); this.readTO = conf.getInt(SOCKET_TIMEOUT_READ, DEFAULT_SOCKET_TIMEOUT_READ); this.writeTO = conf.getInt(SOCKET_TIMEOUT_WRITE, DEFAULT_SOCKET_TIMEOUT_WRITE); + this.metrics = metrics; // login the server principal (if using secure Hadoop) if (LOG.isDebugEnabled()) { @@ -199,25 +204,27 @@ public abstract class AbstractRpcClient implements RpcClient { * @return A pair with the Message response and the Cell data (if any). */ Message callBlockingMethod(Descriptors.MethodDescriptor md, PayloadCarryingRpcController pcrc, - Message param, Message returnType, final User ticket, final InetSocketAddress isa) + Message param, Message returnType, final User ticket, final InetSocketAddress isa, + final ServerName serverName) throws ServiceException { if (pcrc == null) { pcrc = new PayloadCarryingRpcController(); } - long startTime = 0; - if (LOG.isTraceEnabled()) { - startTime = EnvironmentEdgeManager.currentTime(); - } Pair val; try { - val = call(pcrc, md, param, returnType, ticket, isa); + final MetricsConnection.CallStats cs = MetricsConnection.newCallStats(); + cs.setStartTime(EnvironmentEdgeManager.currentTime()); + val = call(pcrc, md, param, returnType, ticket, isa, cs); // Shove the results into controller so can be carried across the proxy/pb service void. pcrc.setCellScanner(val.getSecond()); + cs.setCallTimeMs(EnvironmentEdgeManager.currentTime() - cs.getStartTime()); + if (metrics != null) { + metrics.updateRpc(md, serverName, cs); + } if (LOG.isTraceEnabled()) { - long callTime = EnvironmentEdgeManager.currentTime() - startTime; - LOG.trace("Call: " + md.getName() + ", callTime: " + callTime + "ms"); + LOG.trace("Call: " + md.getName() + ", callTime: " + cs.getCallTimeMs() + "ms"); } return val.getFirst(); } catch (Throwable e) { @@ -242,7 +249,8 @@ public abstract class AbstractRpcClient implements RpcClient { */ protected abstract Pair call(PayloadCarryingRpcController pcrc, Descriptors.MethodDescriptor md, Message param, Message returnType, User ticket, - InetSocketAddress isa) throws IOException, InterruptedException; + InetSocketAddress isa, MetricsConnection.CallStats callStats) + throws IOException, InterruptedException; @Override public BlockingRpcChannel createBlockingRpcChannel(final ServerName sn, final User ticket, @@ -255,6 +263,7 @@ public abstract class AbstractRpcClient implements RpcClient { */ @VisibleForTesting public static class BlockingRpcChannelImplementation implements BlockingRpcChannel { + private final ServerName sn; private final InetSocketAddress isa; private final AbstractRpcClient rpcClient; private final User ticket; @@ -265,6 +274,7 @@ public abstract class AbstractRpcClient implements RpcClient { */ protected BlockingRpcChannelImplementation(final AbstractRpcClient rpcClient, final ServerName sn, final User ticket, int channelOperationTimeout) { + this.sn = sn; this.isa = new InetSocketAddress(sn.getHostname(), sn.getPort()); this.rpcClient = rpcClient; this.ticket = ticket; @@ -285,7 +295,8 @@ public abstract class AbstractRpcClient implements RpcClient { pcrc.setCallTimeout(channelOperationTimeout); } - return this.rpcClient.callBlockingMethod(md, pcrc, param, returnType, this.ticket, this.isa); + return this.rpcClient.callBlockingMethod(md, pcrc, param, returnType, this.ticket, this.isa, + this.sn); } } } diff --git a/hbase-client/src/main/java/org/apache/hadoop/hbase/ipc/AsyncCall.java b/hbase-client/src/main/java/org/apache/hadoop/hbase/ipc/AsyncCall.java index 431c669..a5da0dc 100644 --- a/hbase-client/src/main/java/org/apache/hadoop/hbase/ipc/AsyncCall.java +++ b/hbase-client/src/main/java/org/apache/hadoop/hbase/ipc/AsyncCall.java @@ -25,6 +25,7 @@ import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; import org.apache.hadoop.hbase.CellScanner; import org.apache.hadoop.hbase.classification.InterfaceAudience; +import org.apache.hadoop.hbase.client.MetricsConnection; import org.apache.hadoop.hbase.protobuf.ProtobufUtil; import org.apache.hadoop.hbase.util.EnvironmentEdgeManager; import org.apache.hadoop.hbase.util.ExceptionUtil; @@ -49,6 +50,7 @@ public class AsyncCall extends DefaultPromise { final Message responseDefaultType; final long startTime; final long rpcTimeout; + final MetricsConnection.CallStats callStats; /** * Constructor @@ -61,7 +63,8 @@ public class AsyncCall extends DefaultPromise { * @param responseDefaultType the default response type */ public AsyncCall(EventLoop eventLoop, int connectId, Descriptors.MethodDescriptor md, Message - param, PayloadCarryingRpcController controller, Message responseDefaultType) { + param, PayloadCarryingRpcController controller, Message responseDefaultType, + MetricsConnection.CallStats callStats) { super(eventLoop); this.id = connectId; @@ -73,6 +76,7 @@ public class AsyncCall extends DefaultPromise { this.startTime = EnvironmentEdgeManager.currentTime(); this.rpcTimeout = controller.hasCallTimeout() ? controller.getCallTimeout() : 0; + this.callStats = callStats; } /** diff --git a/hbase-client/src/main/java/org/apache/hadoop/hbase/ipc/AsyncRpcChannel.java b/hbase-client/src/main/java/org/apache/hadoop/hbase/ipc/AsyncRpcChannel.java index 43d75f9..44e8322 100644 --- a/hbase-client/src/main/java/org/apache/hadoop/hbase/ipc/AsyncRpcChannel.java +++ b/hbase-client/src/main/java/org/apache/hadoop/hbase/ipc/AsyncRpcChannel.java @@ -49,6 +49,7 @@ import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; import org.apache.hadoop.hbase.HConstants; import org.apache.hadoop.hbase.classification.InterfaceAudience; +import org.apache.hadoop.hbase.client.MetricsConnection; import org.apache.hadoop.hbase.exceptions.ConnectionClosingException; import org.apache.hadoop.hbase.protobuf.ProtobufUtil; import org.apache.hadoop.hbase.protobuf.generated.AuthenticationProtos; @@ -310,10 +311,10 @@ public class AsyncRpcChannel { */ public Promise callMethod(final Descriptors.MethodDescriptor method, final PayloadCarryingRpcController controller, final Message request, - final Message responsePrototype) { + final Message responsePrototype, MetricsConnection.CallStats callStats) { final AsyncCall call = new AsyncCall(channel.eventLoop(), client.callIdCnt.getAndIncrement(), method, request, - controller, responsePrototype); + controller, responsePrototype, callStats); controller.notifyOnCancel(new RpcCallback() { @Override public void run(Object parameter) { @@ -433,7 +434,7 @@ public class AsyncRpcChannel { ByteBuf b = channel.alloc().directBuffer(4 + totalSize); try(ByteBufOutputStream out = new ByteBufOutputStream(b)) { - IPCUtil.write(out, rh, call.param, cellBlock); + call.callStats.setRequestSizeBytes(IPCUtil.write(out, rh, call.param, cellBlock)); } channel.writeAndFlush(b).addListener(new CallWriteListener(this, call.id)); @@ -579,8 +580,6 @@ public class AsyncRpcChannel { /** * Clean up calls. - * - * @param cleanAll true if all calls should be cleaned, false for only the timed out calls */ private void cleanupCalls() { List toCleanup = new ArrayList(); diff --git a/hbase-client/src/main/java/org/apache/hadoop/hbase/ipc/AsyncRpcClient.java b/hbase-client/src/main/java/org/apache/hadoop/hbase/ipc/AsyncRpcClient.java index e1662f3..4085ad0 100644 --- a/hbase-client/src/main/java/org/apache/hadoop/hbase/ipc/AsyncRpcClient.java +++ b/hbase-client/src/main/java/org/apache/hadoop/hbase/ipc/AsyncRpcClient.java @@ -52,7 +52,9 @@ import org.apache.hadoop.hbase.HBaseInterfaceAudience; import org.apache.hadoop.hbase.HConstants; import org.apache.hadoop.hbase.ServerName; import org.apache.hadoop.hbase.classification.InterfaceAudience; +import org.apache.hadoop.hbase.client.MetricsConnection; import org.apache.hadoop.hbase.security.User; +import org.apache.hadoop.hbase.util.EnvironmentEdgeManager; import org.apache.hadoop.hbase.util.JVM; import org.apache.hadoop.hbase.util.Pair; import org.apache.hadoop.hbase.util.PoolMap; @@ -146,12 +148,13 @@ public class AsyncRpcClient extends AbstractRpcClient { * @param configuration to HBase * @param clusterId for the cluster * @param localAddress local address to connect to + * @param metrics the connection metrics * @param channelInitializer for custom channel handlers */ - @VisibleForTesting - AsyncRpcClient(Configuration configuration, String clusterId, SocketAddress localAddress, + protected AsyncRpcClient(Configuration configuration, String clusterId, + SocketAddress localAddress, MetricsConnection metrics, ChannelInitializer channelInitializer) { - super(configuration, clusterId, localAddress); + super(configuration, clusterId, localAddress, metrics); if (LOG.isDebugEnabled()) { LOG.debug("Starting async Hbase RPC client"); @@ -191,15 +194,28 @@ public class AsyncRpcClient extends AbstractRpcClient { } } + /** Used in test only. */ + AsyncRpcClient(Configuration configuration) { + this(configuration, HConstants.CLUSTER_ID_DEFAULT, null, null); + } + + /** Used in test only. */ + AsyncRpcClient(Configuration configuration, + ChannelInitializer channelInitializer) { + this(configuration, HConstants.CLUSTER_ID_DEFAULT, null, null, channelInitializer); + } + /** * Constructor * * @param configuration to HBase * @param clusterId for the cluster * @param localAddress local address to connect to + * @param metrics the connection metrics */ - public AsyncRpcClient(Configuration configuration, String clusterId, SocketAddress localAddress) { - this(configuration, clusterId, localAddress, null); + public AsyncRpcClient(Configuration configuration, String clusterId, SocketAddress localAddress, + MetricsConnection metrics) { + this(configuration, clusterId, localAddress, metrics, null); } /** @@ -219,13 +235,14 @@ public class AsyncRpcClient extends AbstractRpcClient { @Override protected Pair call(PayloadCarryingRpcController pcrc, Descriptors.MethodDescriptor md, Message param, Message returnType, User ticket, - InetSocketAddress addr) throws IOException, InterruptedException { + InetSocketAddress addr, MetricsConnection.CallStats callStats) + throws IOException, InterruptedException { if (pcrc == null) { pcrc = new PayloadCarryingRpcController(); } final AsyncRpcChannel connection = createRpcChannel(md.getService().getName(), addr, ticket); - Promise promise = connection.callMethod(md, pcrc, param, returnType); + Promise promise = connection.callMethod(md, pcrc, param, returnType, callStats); long timeout = pcrc.hasCallTimeout() ? pcrc.getCallTimeout() : 0; try { Message response = timeout > 0 ? promise.get(timeout, TimeUnit.MILLISECONDS) : promise.get(); @@ -244,40 +261,49 @@ public class AsyncRpcClient extends AbstractRpcClient { /** * Call method async */ - private void callMethod(Descriptors.MethodDescriptor md, final PayloadCarryingRpcController pcrc, - Message param, Message returnType, User ticket, InetSocketAddress addr, - final RpcCallback done) { + private void callMethod(final Descriptors.MethodDescriptor md, + final PayloadCarryingRpcController pcrc, Message param, Message returnType, User ticket, + InetSocketAddress addr, final ServerName serverName, final RpcCallback done) { final AsyncRpcChannel connection; try { connection = createRpcChannel(md.getService().getName(), addr, ticket); - - connection.callMethod(md, pcrc, param, returnType).addListener( + final MetricsConnection.CallStats cs = MetricsConnection.newCallStats(); + GenericFutureListener> listener = new GenericFutureListener>() { - @Override - public void operationComplete(Future future) throws Exception { - if(!future.isSuccess()){ - Throwable cause = future.cause(); - if (cause instanceof IOException) { - pcrc.setFailed((IOException) cause); - }else{ - pcrc.setFailed(new IOException(cause)); - } - }else{ - try { - done.run(future.get()); - }catch (ExecutionException e){ - Throwable cause = e.getCause(); - if (cause instanceof IOException) { - pcrc.setFailed((IOException) cause); - }else{ - pcrc.setFailed(new IOException(cause)); + @Override + public void operationComplete(Future future) throws Exception { + cs.setCallTimeMs(EnvironmentEdgeManager.currentTime() - cs.getStartTime()); + if (metrics != null) { + metrics.updateRpc(md, serverName, cs); + } + if (LOG.isTraceEnabled()) { + LOG.trace("Call: " + md.getName() + ", callTime: " + cs.getCallTimeMs() + "ms"); + } + if (!future.isSuccess()) { + Throwable cause = future.cause(); + if (cause instanceof IOException) { + pcrc.setFailed((IOException) cause); + } else { + pcrc.setFailed(new IOException(cause)); + } + } else { + try { + done.run(future.get()); + } catch (ExecutionException e) { + Throwable cause = e.getCause(); + if (cause instanceof IOException) { + pcrc.setFailed((IOException) cause); + } else { + pcrc.setFailed(new IOException(cause)); + } + } catch (InterruptedException e) { + pcrc.setFailed(new IOException(e)); + } } - }catch (InterruptedException e){ - pcrc.setFailed(new IOException(e)); } - } - } - }); + }; + cs.setStartTime(EnvironmentEdgeManager.currentTime()); + connection.callMethod(md, pcrc, param, returnType, cs).addListener(listener); } catch (StoppedRpcClientException|FailedServerException e) { pcrc.setFailed(e); } @@ -433,6 +459,7 @@ public class AsyncRpcClient extends AbstractRpcClient { @VisibleForTesting public static class RpcChannelImplementation implements RpcChannel { private final InetSocketAddress isa; + private final ServerName serverName; private final AsyncRpcClient rpcClient; private final User ticket; private final int channelOperationTimeout; @@ -442,6 +469,7 @@ public class AsyncRpcClient extends AbstractRpcClient { */ protected RpcChannelImplementation(final AsyncRpcClient rpcClient, final ServerName sn, final User ticket, int channelOperationTimeout) { + this.serverName = sn; this.isa = new InetSocketAddress(sn.getHostname(), sn.getPort()); this.rpcClient = rpcClient; this.ticket = ticket; @@ -462,7 +490,8 @@ public class AsyncRpcClient extends AbstractRpcClient { pcrc.setCallTimeout(channelOperationTimeout); } - this.rpcClient.callMethod(md, pcrc, param, returnType, this.ticket, this.isa, done); + this.rpcClient.callMethod(md, pcrc, param, returnType, this.ticket, this.isa, + this.serverName, done); } } diff --git a/hbase-client/src/main/java/org/apache/hadoop/hbase/ipc/AsyncServerResponseHandler.java b/hbase-client/src/main/java/org/apache/hadoop/hbase/ipc/AsyncServerResponseHandler.java index f7aa8a9..8f6c85b 100644 --- a/hbase-client/src/main/java/org/apache/hadoop/hbase/ipc/AsyncServerResponseHandler.java +++ b/hbase-client/src/main/java/org/apache/hadoop/hbase/ipc/AsyncServerResponseHandler.java @@ -24,8 +24,6 @@ import io.netty.channel.ChannelInboundHandlerAdapter; import java.io.IOException; -import org.apache.commons.logging.Log; -import org.apache.commons.logging.LogFactory; import org.apache.hadoop.hbase.CellScanner; import org.apache.hadoop.hbase.classification.InterfaceAudience; import org.apache.hadoop.hbase.protobuf.ProtobufUtil; @@ -39,8 +37,6 @@ import com.google.protobuf.Message; */ @InterfaceAudience.Private public class AsyncServerResponseHandler extends ChannelInboundHandlerAdapter { - private static final Log LOG = LogFactory.getLog(AsyncServerResponseHandler.class.getName()); - private final AsyncRpcChannel channel; /** @@ -102,6 +98,7 @@ public class AsyncServerResponseHandler extends ChannelInboundHandlerAdapter { cellBlockScanner = channel.client.createCellScanner(cellBlock); } call.setSuccess(value, cellBlockScanner); + call.callStats.setResponseSizeBytes(totalSize); } } catch (IOException e) { // Treat this as a fatal condition and close this connection diff --git a/hbase-client/src/main/java/org/apache/hadoop/hbase/ipc/Call.java b/hbase-client/src/main/java/org/apache/hadoop/hbase/ipc/Call.java index df32730..5f90837 100644 --- a/hbase-client/src/main/java/org/apache/hadoop/hbase/ipc/Call.java +++ b/hbase-client/src/main/java/org/apache/hadoop/hbase/ipc/Call.java @@ -21,6 +21,7 @@ import com.google.protobuf.Descriptors; import com.google.protobuf.Message; import org.apache.hadoop.hbase.CellScanner; import org.apache.hadoop.hbase.classification.InterfaceAudience; +import org.apache.hadoop.hbase.client.MetricsConnection; import org.apache.hadoop.hbase.protobuf.ProtobufUtil; import org.apache.hadoop.hbase.util.EnvironmentEdgeManager; @@ -41,16 +42,18 @@ public class Call { Message responseDefaultType; IOException error; // exception, null if value volatile boolean done; // true when call is done - long startTime; final Descriptors.MethodDescriptor md; final int timeout; // timeout in millisecond for this call; 0 means infinite. + final MetricsConnection.CallStats callStats; protected Call(int id, final Descriptors.MethodDescriptor md, Message param, - final CellScanner cells, final Message responseDefaultType, int timeout) { + final CellScanner cells, final Message responseDefaultType, int timeout, + MetricsConnection.CallStats callStats) { this.param = param; this.md = md; this.cells = cells; - this.startTime = EnvironmentEdgeManager.currentTime(); + this.callStats = callStats; + this.callStats.setStartTime(EnvironmentEdgeManager.currentTime()); this.responseDefaultType = responseDefaultType; this.id = id; this.timeout = timeout; @@ -122,6 +125,6 @@ public class Call { } public long getStartTime() { - return this.startTime; + return this.callStats.getStartTime(); } } \ No newline at end of file diff --git a/hbase-client/src/main/java/org/apache/hadoop/hbase/ipc/RpcClientFactory.java b/hbase-client/src/main/java/org/apache/hadoop/hbase/ipc/RpcClientFactory.java index 10ddc56..822daca 100644 --- a/hbase-client/src/main/java/org/apache/hadoop/hbase/ipc/RpcClientFactory.java +++ b/hbase-client/src/main/java/org/apache/hadoop/hbase/ipc/RpcClientFactory.java @@ -17,8 +17,10 @@ */ package org.apache.hadoop.hbase.ipc; +import com.google.common.annotations.VisibleForTesting; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.hbase.classification.InterfaceAudience; +import org.apache.hadoop.hbase.client.MetricsConnection; import org.apache.hadoop.hbase.util.ReflectionUtils; import java.net.SocketAddress; @@ -37,15 +39,23 @@ public final class RpcClientFactory { private RpcClientFactory() { } + /** Helper method for tests only. Creates an {@code RpcClient} without metrics. */ + @VisibleForTesting + public static RpcClient createClient(Configuration conf, String clusterId) { + return createClient(conf, clusterId, null); + } + /** * Creates a new RpcClient by the class defined in the configuration or falls back to * RpcClientImpl * @param conf configuration * @param clusterId the cluster id + * @param metrics the connection metrics * @return newly created RpcClient */ - public static RpcClient createClient(Configuration conf, String clusterId) { - return createClient(conf, clusterId, null); + public static RpcClient createClient(Configuration conf, String clusterId, + MetricsConnection metrics) { + return createClient(conf, clusterId, null, metrics); } /** @@ -54,16 +64,18 @@ public final class RpcClientFactory { * @param conf configuration * @param clusterId the cluster id * @param localAddr client socket bind address. + * @param metrics the connection metrics * @return newly created RpcClient */ public static RpcClient createClient(Configuration conf, String clusterId, - SocketAddress localAddr) { + SocketAddress localAddr, MetricsConnection metrics) { String rpcClientClass = conf.get(CUSTOM_RPC_CLIENT_IMPL_CONF_KEY, AsyncRpcClient.class.getName()); return ReflectionUtils.instantiateWithCustomCtor( rpcClientClass, - new Class[] { Configuration.class, String.class, SocketAddress.class }, - new Object[] { conf, clusterId, localAddr } + new Class[] { Configuration.class, String.class, SocketAddress.class, + MetricsConnection.class }, + new Object[] { conf, clusterId, localAddr, metrics } ); } } \ No newline at end of file diff --git a/hbase-client/src/main/java/org/apache/hadoop/hbase/ipc/RpcClientImpl.java b/hbase-client/src/main/java/org/apache/hadoop/hbase/ipc/RpcClientImpl.java index cb18952..76fa538 100644 --- a/hbase-client/src/main/java/org/apache/hadoop/hbase/ipc/RpcClientImpl.java +++ b/hbase-client/src/main/java/org/apache/hadoop/hbase/ipc/RpcClientImpl.java @@ -19,6 +19,7 @@ package org.apache.hadoop.hbase.ipc; +import com.google.common.annotations.VisibleForTesting; import com.google.protobuf.Descriptors.MethodDescriptor; import com.google.protobuf.Message; import com.google.protobuf.Message.Builder; @@ -32,6 +33,7 @@ import org.apache.hadoop.hbase.DoNotRetryIOException; import org.apache.hadoop.hbase.HConstants; import org.apache.hadoop.hbase.ServerName; import org.apache.hadoop.hbase.classification.InterfaceAudience; +import org.apache.hadoop.hbase.client.MetricsConnection; import org.apache.hadoop.hbase.codec.Codec; import org.apache.hadoop.hbase.exceptions.ConnectionClosingException; import org.apache.hadoop.hbase.protobuf.ProtobufUtil; @@ -911,7 +913,8 @@ public class RpcClientImpl extends AbstractRpcClient { checkIsOpen(); // Now we're checking that it didn't became idle in between. try { - IPCUtil.write(this.out, header, call.param, cellBlock); + call.callStats.setRequestSizeBytes( + IPCUtil.write(this.out, header, call.param, cellBlock)); } catch (IOException e) { // We set the value inside the synchronized block, this way the next in line // won't even try to write. Otherwise we might miss a call in the calls map? @@ -964,12 +967,20 @@ public class RpcClientImpl extends AbstractRpcClient { int readSoFar = IPCUtil.getTotalSizeWhenWrittenDelimited(responseHeader); int whatIsLeftToRead = totalSize - readSoFar; IOUtils.skipFully(in, whatIsLeftToRead); + if (call != null) { + call.callStats.setResponseSizeBytes(totalSize); + call.callStats.setCallTimeMs( + EnvironmentEdgeManager.currentTime() - call.callStats.getStartTime()); + } return; } if (responseHeader.hasException()) { ExceptionResponse exceptionResponse = responseHeader.getException(); RemoteException re = createRemoteException(exceptionResponse); call.setException(re); + call.callStats.setResponseSizeBytes(totalSize); + call.callStats.setCallTimeMs( + EnvironmentEdgeManager.currentTime() - call.callStats.getStartTime()); if (isFatalConnectionException(exceptionResponse)) { markClosed(re); } @@ -988,6 +999,9 @@ public class RpcClientImpl extends AbstractRpcClient { cellBlockScanner = ipcUtil.createCellScanner(this.codec, this.compressor, cellBlock); } call.setResponse(value, cellBlockScanner); + call.callStats.setResponseSizeBytes(totalSize); + call.callStats.setCallTimeMs( + EnvironmentEdgeManager.currentTime() - call.callStats.getStartTime()); } } catch (IOException e) { if (expectedCall) call.setException(e); @@ -1075,13 +1089,15 @@ public class RpcClientImpl extends AbstractRpcClient { } /** - * Construct an IPC cluster client whose values are of the {@link Message} class. + * Used in test only. Construct an IPC cluster client whose values are of the + * {@link Message} class. * @param conf configuration * @param clusterId the cluster id * @param factory socket factory */ + @VisibleForTesting RpcClientImpl(Configuration conf, String clusterId, SocketFactory factory) { - this(conf, clusterId, factory, null); + this(conf, clusterId, factory, null, null); } /** @@ -1090,10 +1106,11 @@ public class RpcClientImpl extends AbstractRpcClient { * @param clusterId the cluster id * @param factory socket factory * @param localAddr client socket bind address + * @param metrics the connection metrics */ RpcClientImpl(Configuration conf, String clusterId, SocketFactory factory, - SocketAddress localAddr) { - super(conf, clusterId, localAddr); + SocketAddress localAddr, MetricsConnection metrics) { + super(conf, clusterId, localAddr, metrics); this.socketFactory = factory; this.connections = new PoolMap(getPoolType(conf), getPoolSize(conf)); @@ -1101,25 +1118,36 @@ public class RpcClientImpl extends AbstractRpcClient { } /** - * Construct an IPC client for the cluster clusterId with the default SocketFactory - * @param conf configuration - * @param clusterId the cluster id + * Used in test only. Construct an IPC client for the cluster {@code clusterId} with + * the default SocketFactory + */ + @VisibleForTesting + RpcClientImpl(Configuration conf, String clusterId) { + this(conf, clusterId, NetUtils.getDefaultSocketFactory(conf), null, null); + } + + /** + * Used in test only. Construct an IPC client for the cluster {@code clusterId} with + * the default SocketFactory */ - public RpcClientImpl(Configuration conf, String clusterId) { - this(conf, clusterId, NetUtils.getDefaultSocketFactory(conf), null); + @VisibleForTesting + public RpcClientImpl(Configuration conf, String clusterId, SocketAddress localAddr) { + this(conf, clusterId, localAddr, null); } /** - * Construct an IPC client for the cluster clusterId with the default SocketFactory + * Construct an IPC client for the cluster {@code clusterId} with the default SocketFactory * * This method is called with reflection by the RpcClientFactory to create an instance * * @param conf configuration * @param clusterId the cluster id * @param localAddr client socket bind address. + * @param metrics the connection metrics */ - public RpcClientImpl(Configuration conf, String clusterId, SocketAddress localAddr) { - this(conf, clusterId, NetUtils.getDefaultSocketFactory(conf), localAddr); + public RpcClientImpl(Configuration conf, String clusterId, SocketAddress localAddr, + MetricsConnection metrics) { + this(conf, clusterId, NetUtils.getDefaultSocketFactory(conf), localAddr, metrics); } /** Stop all threads related to this client. No further calls may be made @@ -1182,7 +1210,8 @@ public class RpcClientImpl extends AbstractRpcClient { */ @Override protected Pair call(PayloadCarryingRpcController pcrc, MethodDescriptor md, - Message param, Message returnType, User ticket, InetSocketAddress addr) + Message param, Message returnType, User ticket, InetSocketAddress addr, + MetricsConnection.CallStats callStats) throws IOException, InterruptedException { if (pcrc == null) { pcrc = new PayloadCarryingRpcController(); @@ -1190,7 +1219,7 @@ public class RpcClientImpl extends AbstractRpcClient { CellScanner cells = pcrc.cellScanner(); final Call call = new Call(this.callIdCnt.getAndIncrement(), md, param, cells, returnType, - pcrc.getCallTimeout()); + pcrc.getCallTimeout(), MetricsConnection.newCallStats()); final Connection connection = getConnection(ticket, call, addr); diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HRegionServer.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HRegionServer.java index 59d13fa..1f71dbd 100644 --- a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HRegionServer.java +++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HRegionServer.java @@ -851,7 +851,7 @@ public class HRegionServer extends HasThread implements // Setup RPC client for master communication rpcClient = RpcClientFactory.createClient(conf, clusterId, new InetSocketAddress( - rpcServices.isa.getAddress(), 0)); + rpcServices.isa.getAddress(), 0), clusterConnection.getConnectionMetrics()); boolean onlyMetaRefresh = false; int storefileRefreshPeriod = conf.getInt( diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/MetricsRegionServer.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/MetricsRegionServer.java index b2cb772..91f494a 100644 --- a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/MetricsRegionServer.java +++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/MetricsRegionServer.java @@ -17,6 +17,7 @@ */ package org.apache.hadoop.hbase.regionserver; +import com.google.common.annotations.VisibleForTesting; import org.apache.hadoop.hbase.classification.InterfaceAudience; import org.apache.hadoop.hbase.classification.InterfaceStability; import org.apache.hadoop.hbase.CompatibilitySingletonFactory; @@ -48,7 +49,7 @@ public class MetricsRegionServer { this.serverSource = serverSource; } - // for unit-test usage + @VisibleForTesting public MetricsRegionServerSource getMetricsSource() { return serverSource; } diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/client/TestClientTimeouts.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/client/TestClientTimeouts.java index 418a0ac..0ae16ce 100644 --- a/hbase-server/src/test/java/org/apache/hadoop/hbase/client/TestClientTimeouts.java +++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/client/TestClientTimeouts.java @@ -134,7 +134,7 @@ public class TestClientTimeouts { /** * Rpc Channel implementation with RandomTimeoutBlockingRpcChannel */ - public static class RandomTimeoutRpcClient extends RpcClientImpl{ + public static class RandomTimeoutRpcClient extends RpcClientImpl { public RandomTimeoutRpcClient(Configuration conf, String clusterId, SocketAddress localAddr) { super(conf, clusterId, localAddr); } diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/ipc/AbstractTestIPC.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/ipc/AbstractTestIPC.java index 32eb9f6..d427419 100644 --- a/hbase-server/src/test/java/org/apache/hadoop/hbase/ipc/AbstractTestIPC.java +++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/ipc/AbstractTestIPC.java @@ -38,6 +38,7 @@ import org.apache.hadoop.hbase.CellScanner; import org.apache.hadoop.hbase.CellUtil; import org.apache.hadoop.hbase.HBaseConfiguration; import org.apache.hadoop.hbase.KeyValue; +import org.apache.hadoop.hbase.client.MetricsConnection; import org.apache.hadoop.hbase.ipc.protobuf.generated.TestProtos.EchoRequestProto; import org.apache.hadoop.hbase.ipc.protobuf.generated.TestProtos.EchoResponseProto; import org.apache.hadoop.hbase.ipc.protobuf.generated.TestProtos.EmptyRequestProto; @@ -163,7 +164,8 @@ public abstract class AbstractTestIPC { final String message = "hello"; EchoRequestProto param = EchoRequestProto.newBuilder().setMessage(message).build(); Pair r = - client.call(null, md, param, md.getOutputType().toProto(), User.getCurrent(), address); + client.call(null, md, param, md.getOutputType().toProto(), User.getCurrent(), address, + new MetricsConnection.CallStats()); assertTrue(r.getSecond() == null); // Silly assertion that the message is in the returned pb. assertTrue(r.getFirst().toString().contains(message)); @@ -205,7 +207,8 @@ public abstract class AbstractTestIPC { PayloadCarryingRpcController pcrc = new PayloadCarryingRpcController(CellUtil.createCellScanner(cells)); Pair r = - client.call(pcrc, md, param, md.getOutputType().toProto(), User.getCurrent(), address); + client.call(pcrc, md, param, md.getOutputType().toProto(), User.getCurrent(), address, + new MetricsConnection.CallStats()); int index = 0; while (r.getSecond().advance()) { assertTrue(CELL.equals(r.getSecond().current())); @@ -231,7 +234,8 @@ public abstract class AbstractTestIPC { InetSocketAddress address = rpcServer.getListenerAddress(); MethodDescriptor md = SERVICE.getDescriptorForType().findMethodByName("echo"); EchoRequestProto param = EchoRequestProto.newBuilder().setMessage("hello").build(); - client.call(null, md, param, null, User.getCurrent(), address); + client.call(null, md, param, null, User.getCurrent(), address, + new MetricsConnection.CallStats()); fail("Expected an exception to have been thrown!"); } catch (Exception e) { LOG.info("Caught expected exception: " + e.toString()); @@ -255,10 +259,10 @@ public abstract class AbstractTestIPC { MethodDescriptor md = SERVICE.getDescriptorForType().findMethodByName("echo"); EchoRequestProto param = EchoRequestProto.newBuilder().setMessage("hello").build(); for (int i = 0; i < 10; i++) { - client.call( - new PayloadCarryingRpcController( - CellUtil.createCellScanner(ImmutableList. of(CELL))), md, param, md - .getOutputType().toProto(), User.getCurrent(), rpcServer.getListenerAddress()); + client.call(new PayloadCarryingRpcController( + CellUtil.createCellScanner(ImmutableList. of(CELL))), md, param, + md.getOutputType().toProto(), User.getCurrent(), rpcServer.getListenerAddress(), + new MetricsConnection.CallStats()); } verify(scheduler, times(10)).dispatch((CallRunner) anyObject()); } finally { diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/ipc/TestAsyncIPC.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/ipc/TestAsyncIPC.java index 18e3798..d761ae9 100644 --- a/hbase-server/src/test/java/org/apache/hadoop/hbase/ipc/TestAsyncIPC.java +++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/ipc/TestAsyncIPC.java @@ -46,6 +46,7 @@ import org.apache.hadoop.hbase.HRegionInfo; import org.apache.hadoop.hbase.KeyValue; import org.apache.hadoop.hbase.ServerName; import org.apache.hadoop.hbase.Waiter; +import org.apache.hadoop.hbase.client.MetricsConnection; import org.apache.hadoop.hbase.client.Put; import org.apache.hadoop.hbase.client.RowMutations; import org.apache.hadoop.hbase.codec.Codec; @@ -116,7 +117,7 @@ public class TestAsyncIPC extends AbstractTestIPC { @Override protected AsyncRpcClient createRpcClientNoCodec(Configuration conf) { setConf(conf); - return new AsyncRpcClient(conf, HConstants.CLUSTER_ID_DEFAULT, null) { + return new AsyncRpcClient(conf) { @Override Codec getCodec() { @@ -129,15 +130,13 @@ public class TestAsyncIPC extends AbstractTestIPC { @Override protected AsyncRpcClient createRpcClient(Configuration conf) { setConf(conf); - return new AsyncRpcClient(conf, HConstants.CLUSTER_ID_DEFAULT, null); + return new AsyncRpcClient(conf); } @Override protected AsyncRpcClient createRpcClientRTEDuringConnectionSetup(Configuration conf) { setConf(conf); - return new AsyncRpcClient(conf, HConstants.CLUSTER_ID_DEFAULT, null, - new ChannelInitializer() { - + return new AsyncRpcClient(conf, new ChannelInitializer() { @Override protected void initChannel(SocketChannel ch) throws Exception { ch.pipeline().addFirst(new ChannelOutboundHandlerAdapter() { @@ -248,7 +247,7 @@ public class TestAsyncIPC extends AbstractTestIPC { TestRpcServer rpcServer = new TestRpcServer(); MethodDescriptor md = SERVICE.getDescriptorForType().findMethodByName("echo"); EchoRequestProto param = EchoRequestProto.newBuilder().setMessage("hello").build(); - AsyncRpcClient client = new AsyncRpcClient(conf, HConstants.CLUSTER_ID_DEFAULT, null); + AsyncRpcClient client = new AsyncRpcClient(conf); KeyValue kv = BIG_CELL; Put p = new Put(CellUtil.cloneRow(kv)); for (int i = 0; i < cellcount; i++) { @@ -282,7 +281,8 @@ public class TestAsyncIPC extends AbstractTestIPC { PayloadCarryingRpcController pcrc = new PayloadCarryingRpcController(CellUtil.createCellScanner(cells)); // Pair response = - client.call(pcrc, md, builder.build(), param, user, address); + client.call(pcrc, md, builder.build(), param, user, address, + new MetricsConnection.CallStats()); /* * int count = 0; while (p.getSecond().advance()) { count++; } assertEquals(cells.size(), * count); diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/ipc/TestGlobalEventLoopGroup.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/ipc/TestGlobalEventLoopGroup.java index 60dbd1b..52c77e7 100644 --- a/hbase-server/src/test/java/org/apache/hadoop/hbase/ipc/TestGlobalEventLoopGroup.java +++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/ipc/TestGlobalEventLoopGroup.java @@ -37,15 +37,15 @@ public class TestGlobalEventLoopGroup { public void test() { Configuration conf = HBaseConfiguration.create(); conf.setBoolean(AsyncRpcClient.USE_GLOBAL_EVENT_LOOP_GROUP, true); - AsyncRpcClient client = new AsyncRpcClient(conf, HConstants.CLUSTER_ID_DEFAULT, null); + AsyncRpcClient client = new AsyncRpcClient(conf); assertNotNull(AsyncRpcClient.GLOBAL_EVENT_LOOP_GROUP); - AsyncRpcClient client1 = new AsyncRpcClient(conf, HConstants.CLUSTER_ID_DEFAULT, null); + AsyncRpcClient client1 = new AsyncRpcClient(conf); assertSame(client.bootstrap.group(), client1.bootstrap.group()); client1.close(); assertFalse(client.bootstrap.group().isShuttingDown()); conf.setBoolean(AsyncRpcClient.USE_GLOBAL_EVENT_LOOP_GROUP, false); - AsyncRpcClient client2 = new AsyncRpcClient(conf, HConstants.CLUSTER_ID_DEFAULT, null); + AsyncRpcClient client2 = new AsyncRpcClient(conf); assertNotSame(client.bootstrap.group(), client2.bootstrap.group()); client2.close(); diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/ipc/TestIPC.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/ipc/TestIPC.java index 67e4e4f..d1b8202 100644 --- a/hbase-server/src/test/java/org/apache/hadoop/hbase/ipc/TestIPC.java +++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/ipc/TestIPC.java @@ -39,6 +39,7 @@ import org.apache.hadoop.hbase.HBaseConfiguration; import org.apache.hadoop.hbase.HConstants; import org.apache.hadoop.hbase.HRegionInfo; import org.apache.hadoop.hbase.KeyValue; +import org.apache.hadoop.hbase.client.MetricsConnection; import org.apache.hadoop.hbase.client.Put; import org.apache.hadoop.hbase.client.RowMutations; import org.apache.hadoop.hbase.codec.Codec; @@ -145,7 +146,8 @@ public class TestIPC extends AbstractTestIPC { PayloadCarryingRpcController pcrc = new PayloadCarryingRpcController(CellUtil.createCellScanner(cells)); // Pair response = - client.call(pcrc, md, builder.build(), param, user, address); + client.call(pcrc, md, builder.build(), param, user, address, + new MetricsConnection.CallStats()); /* * int count = 0; while (p.getSecond().advance()) { count++; } assertEquals(cells.size(), * count); diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/ipc/TestRpcClientLeaks.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/ipc/TestRpcClientLeaks.java index 2965055..596b8ab 100644 --- a/hbase-server/src/test/java/org/apache/hadoop/hbase/ipc/TestRpcClientLeaks.java +++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/ipc/TestRpcClientLeaks.java @@ -32,6 +32,7 @@ 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.Get; +import org.apache.hadoop.hbase.client.MetricsConnection; import org.apache.hadoop.hbase.client.RetriesExhaustedException; import org.apache.hadoop.hbase.client.Table; import org.apache.hadoop.hbase.codec.Codec; @@ -58,8 +59,9 @@ public class TestRpcClientLeaks { super(conf, clusterId); } - public MyRpcClientImpl(Configuration conf, String clusterId, SocketAddress address) { - super(conf, clusterId, address); + public MyRpcClientImpl(Configuration conf, String clusterId, SocketAddress address, + MetricsConnection metrics) { + super(conf, clusterId, address, metrics); } @Override diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/ipc/TestRpcHandlerException.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/ipc/TestRpcHandlerException.java index 8a6b092..a4e55d9 100644 --- a/hbase-server/src/test/java/org/apache/hadoop/hbase/ipc/TestRpcHandlerException.java +++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/ipc/TestRpcHandlerException.java @@ -34,6 +34,7 @@ import org.apache.hadoop.hbase.CellUtil; import org.apache.hadoop.hbase.HBaseConfiguration; import org.apache.hadoop.hbase.HConstants; import org.apache.hadoop.hbase.KeyValue; +import org.apache.hadoop.hbase.client.MetricsConnection; import org.apache.hadoop.hbase.ipc.protobuf.generated.TestProtos.EchoRequestProto; import org.apache.hadoop.hbase.ipc.protobuf.generated.TestProtos.EchoResponseProto; import org.apache.hadoop.hbase.ipc.protobuf.generated.TestProtos.EmptyRequestProto; @@ -181,7 +182,7 @@ public class TestRpcHandlerException { new PayloadCarryingRpcController(CellUtil.createCellScanner(ImmutableList.of(CELL))); client.call(controller, md, param, md.getOutputType().toProto(), User.getCurrent(), - rpcServer.getListenerAddress()); + rpcServer.getListenerAddress(), new MetricsConnection.CallStats()); } catch (Throwable e) { assert(abortable.isAborted() == true); } finally {