You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hbase.apache.org by zh...@apache.org on 2017/03/16 15:00:59 UTC
[06/10] hbase git commit: HBASE-16584 Backport the new ipc
implementation in HBASE-16432 to branch-1
http://git-wip-us.apache.org/repos/asf/hbase/blob/094e9a31/hbase-client/src/main/java/org/apache/hadoop/hbase/ipc/RpcClientImpl.java
----------------------------------------------------------------------
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
deleted file mode 100644
index 58e577f..0000000
--- a/hbase-client/src/main/java/org/apache/hadoop/hbase/ipc/RpcClientImpl.java
+++ /dev/null
@@ -1,1326 +0,0 @@
-/**
- *
- * 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.ipc;
-
-import java.io.BufferedInputStream;
-import java.io.BufferedOutputStream;
-import java.io.Closeable;
-import java.io.DataInputStream;
-import java.io.DataOutputStream;
-import java.io.IOException;
-import java.io.InputStream;
-import java.io.InterruptedIOException;
-import java.io.OutputStream;
-import java.net.InetSocketAddress;
-import java.net.Socket;
-import java.net.SocketAddress;
-import java.net.SocketTimeoutException;
-import java.net.UnknownHostException;
-import java.nio.ByteBuffer;
-import java.security.PrivilegedExceptionAction;
-import java.util.HashMap;
-import java.util.HashSet;
-import java.util.Iterator;
-import java.util.Locale;
-import java.util.Map;
-import java.util.Map.Entry;
-import java.util.Random;
-import java.util.Set;
-import java.util.concurrent.ArrayBlockingQueue;
-import java.util.concurrent.BlockingQueue;
-import java.util.concurrent.ConcurrentSkipListMap;
-import java.util.concurrent.atomic.AtomicBoolean;
-import java.util.concurrent.atomic.AtomicInteger;
-
-import javax.net.SocketFactory;
-import javax.security.sasl.SaslException;
-
-import org.apache.commons.logging.Log;
-import org.apache.commons.logging.LogFactory;
-import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.hbase.CellScanner;
-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;
-import org.apache.hadoop.hbase.protobuf.generated.AuthenticationProtos;
-import org.apache.hadoop.hbase.protobuf.generated.RPCProtos.CellBlockMeta;
-import org.apache.hadoop.hbase.protobuf.generated.RPCProtos.ConnectionHeader;
-import org.apache.hadoop.hbase.protobuf.generated.RPCProtos.ExceptionResponse;
-import org.apache.hadoop.hbase.protobuf.generated.RPCProtos.RequestHeader;
-import org.apache.hadoop.hbase.protobuf.generated.RPCProtos.ResponseHeader;
-import org.apache.hadoop.hbase.protobuf.generated.RPCProtos.UserInformation;
-import org.apache.hadoop.hbase.protobuf.generated.TracingProtos.RPCTInfo;
-import org.apache.hadoop.hbase.security.AuthMethod;
-import org.apache.hadoop.hbase.security.HBaseSaslRpcClient;
-import org.apache.hadoop.hbase.security.SaslUtil.QualityOfProtection;
-import org.apache.hadoop.hbase.security.SecurityInfo;
-import org.apache.hadoop.hbase.security.User;
-import org.apache.hadoop.hbase.security.UserProvider;
-import org.apache.hadoop.hbase.security.token.AuthenticationTokenSelector;
-import org.apache.hadoop.hbase.util.EnvironmentEdgeManager;
-import org.apache.hadoop.hbase.util.ExceptionUtil;
-import org.apache.hadoop.hbase.util.Pair;
-import org.apache.hadoop.hbase.util.PoolMap;
-import org.apache.hadoop.io.IOUtils;
-import org.apache.hadoop.io.Text;
-import org.apache.hadoop.io.compress.CompressionCodec;
-import org.apache.hadoop.ipc.RemoteException;
-import org.apache.hadoop.net.NetUtils;
-import org.apache.hadoop.security.SecurityUtil;
-import org.apache.hadoop.security.UserGroupInformation;
-import org.apache.hadoop.security.token.Token;
-import org.apache.hadoop.security.token.TokenIdentifier;
-import org.apache.hadoop.security.token.TokenSelector;
-import org.apache.htrace.Span;
-import org.apache.htrace.Trace;
-import org.apache.htrace.TraceScope;
-
-import com.google.common.annotations.VisibleForTesting;
-import com.google.protobuf.Descriptors.MethodDescriptor;
-import com.google.protobuf.Message;
-import com.google.protobuf.Message.Builder;
-import com.google.protobuf.RpcCallback;
-
-/**
- * Does RPC against a cluster. Manages connections per regionserver in the cluster.
- * <p>See HBaseServer
- */
-@InterfaceAudience.Private
-public class RpcClientImpl extends AbstractRpcClient {
- private static final Log LOG = LogFactory.getLog(RpcClientImpl.class);
- protected final AtomicInteger callIdCnt = new AtomicInteger();
-
- protected final PoolMap<ConnectionId, Connection> connections;
-
- protected final AtomicBoolean running = new AtomicBoolean(true); // if client runs
-
- protected final FailedServers failedServers;
-
- protected final SocketFactory socketFactory; // how to create sockets
-
- protected final static Map<AuthenticationProtos.TokenIdentifier.Kind,
- TokenSelector<? extends TokenIdentifier>> tokenHandlers =
- new HashMap<AuthenticationProtos.TokenIdentifier.Kind,
- TokenSelector<? extends TokenIdentifier>>();
- static {
- tokenHandlers.put(AuthenticationProtos.TokenIdentifier.Kind.HBASE_AUTH_TOKEN,
- new AuthenticationTokenSelector());
- }
-
- /**
- * Creates a connection. Can be overridden by a subclass for testing.
- * @param remoteId - the ConnectionId to use for the connection creation.
- */
- protected Connection createConnection(ConnectionId remoteId, final Codec codec,
- final CompressionCodec compressor)
- throws IOException {
- return new Connection(remoteId, codec, compressor);
- }
-
- /**
- * see {@link RpcClientImpl.Connection.CallSender}
- */
- private static class CallFuture {
- final Call call;
- final int priority;
- final Span span;
-
- // We will use this to stop the writer
- final static CallFuture DEATH_PILL = new CallFuture(null, -1, null);
-
- CallFuture(Call call, int priority, Span span) {
- this.call = call;
- this.priority = priority;
- this.span = span;
- }
- }
-
- /** Thread that reads responses and notifies callers. Each connection owns a
- * socket connected to a remote address. Calls are multiplexed through this
- * socket: responses may be delivered out of order. */
- protected class Connection extends Thread {
- private ConnectionHeader header; // connection header
- protected ConnectionId remoteId;
- protected Socket socket = null; // connected socket
- protected DataInputStream in;
- protected DataOutputStream out;
- private Object outLock = new Object();
- private InetSocketAddress server; // server ip:port
- private String serverPrincipal; // server's krb5 principal name
- private AuthMethod authMethod; // authentication method
- private boolean useSasl;
- private Token<? extends TokenIdentifier> token;
- private HBaseSaslRpcClient saslRpcClient;
- private int reloginMaxBackoff; // max pause before relogin on sasl failure
- private final Codec codec;
- private final CompressionCodec compressor;
-
- // currently active calls
- protected final ConcurrentSkipListMap<Integer, Call> calls =
- new ConcurrentSkipListMap<Integer, Call>();
-
- protected final AtomicBoolean shouldCloseConnection = new AtomicBoolean();
- protected final CallSender callSender;
-
-
- /**
- * If the client wants to interrupt its calls easily (i.e. call Thread#interrupt),
- * it gets into a java issue: an interruption during a write closes the socket/channel.
- * A way to avoid this is to use a different thread for writing. This way, on interruptions,
- * we either cancel the writes or ignore the answer if the write is already done, but we
- * don't stop the write in the middle.
- * This adds a thread per region server in the client, so it's kept as an option.
- * <p>
- * The implementation is simple: the client threads adds their call to the queue, and then
- * wait for an answer. The CallSender blocks on the queue, and writes the calls one
- * after the other. On interruption, the client cancels its call. The CallSender checks that
- * the call has not been canceled before writing it.
- * </p>
- * When the connection closes, all the calls not yet sent are dismissed. The client thread
- * is notified with an appropriate exception, as if the call was already sent but the answer
- * not yet received.
- * </p>
- */
- private class CallSender extends Thread implements Closeable {
- protected final BlockingQueue<CallFuture> callsToWrite;
-
-
- public CallFuture sendCall(Call call, int priority, Span span)
- throws InterruptedException, IOException {
- CallFuture cts = new CallFuture(call, priority, span);
- if (!callsToWrite.offer(cts)) {
- throw new IOException("Can't add the call " + call.id +
- " to the write queue. callsToWrite.size()=" + callsToWrite.size());
- }
- checkIsOpen(); // We check after the put, to be sure that the call we added won't stay
- // in the list while the cleanup was already done.
- return cts;
- }
-
- @Override
- public void close(){
- assert shouldCloseConnection.get();
- callsToWrite.offer(CallFuture.DEATH_PILL);
- // We don't care if we can't add the death pill to the queue: the writer
- // won't be blocked in the 'take', as its queue is full.
- }
-
- CallSender(String name, Configuration conf) {
- int queueSize = conf.getInt("hbase.ipc.client.write.queueSize", 1000);
- callsToWrite = new ArrayBlockingQueue<CallFuture>(queueSize);
- setDaemon(true);
- setName(name + " - writer");
- }
-
- public void remove(CallFuture cts){
- callsToWrite.remove(cts);
-
- // By removing the call from the expected call list, we make the list smaller, but
- // it means as well that we don't know how many calls we cancelled.
- calls.remove(cts.call.id);
- cts.call.callComplete();
- }
-
- /**
- * Reads the call from the queue, write them on the socket.
- */
- @Override
- public void run() {
- while (!shouldCloseConnection.get()) {
- CallFuture cts = null;
- try {
- cts = callsToWrite.take();
- } catch (InterruptedException e) {
- markClosed(new InterruptedIOException());
- }
-
- if (cts == null || cts == CallFuture.DEATH_PILL) {
- assert shouldCloseConnection.get();
- break;
- }
-
- if (cts.call.done) {
- continue;
- }
-
- if (cts.call.checkAndSetTimeout()) {
- continue;
- }
-
- try {
- Connection.this.tracedWriteRequest(cts.call, cts.priority, cts.span);
- } catch (IOException e) {
- if (LOG.isDebugEnabled()) {
- LOG.debug("call write error for call #" + cts.call.id
- + ", message =" + e.getMessage());
- }
- cts.call.setException(e);
- markClosed(e);
- }
- }
-
- cleanup();
- }
-
- /**
- * Cleans the call not yet sent when we finish.
- */
- private void cleanup() {
- assert shouldCloseConnection.get();
-
- IOException ie = new ConnectionClosingException("Connection to " + server + " is closing.");
- while (true) {
- CallFuture cts = callsToWrite.poll();
- if (cts == null) {
- break;
- }
- if (cts.call != null && !cts.call.done) {
- cts.call.setException(ie);
- }
- }
- }
- }
-
- Connection(ConnectionId remoteId, final Codec codec, final CompressionCodec compressor)
- throws IOException {
- if (remoteId.getAddress().isUnresolved()) {
- throw new UnknownHostException("unknown host: " + remoteId.getAddress().getHostName());
- }
- this.server = remoteId.getAddress();
- this.codec = codec;
- this.compressor = compressor;
-
- UserGroupInformation ticket = remoteId.getTicket().getUGI();
- SecurityInfo securityInfo = SecurityInfo.getInfo(remoteId.getServiceName());
- this.useSasl = userProvider.isHBaseSecurityEnabled();
- if (useSasl && securityInfo != null) {
- AuthenticationProtos.TokenIdentifier.Kind tokenKind = securityInfo.getTokenKind();
- if (tokenKind != null) {
- TokenSelector<? extends TokenIdentifier> tokenSelector =
- tokenHandlers.get(tokenKind);
- if (tokenSelector != null) {
- token = tokenSelector.selectToken(new Text(clusterId),
- ticket.getTokens());
- } else if (LOG.isDebugEnabled()) {
- LOG.debug("No token selector found for type "+tokenKind);
- }
- }
- String serverKey = securityInfo.getServerPrincipal();
- if (serverKey == null) {
- throw new IOException(
- "Can't obtain server Kerberos config key from SecurityInfo");
- }
- serverPrincipal = SecurityUtil.getServerPrincipal(
- conf.get(serverKey), server.getAddress().getCanonicalHostName().toLowerCase(Locale.ROOT));
- if (LOG.isDebugEnabled()) {
- LOG.debug("RPC Server Kerberos principal name for service="
- + remoteId.getServiceName() + " is " + serverPrincipal);
- }
- }
-
- if (!useSasl) {
- authMethod = AuthMethod.SIMPLE;
- } else if (token != null) {
- authMethod = AuthMethod.DIGEST;
- } else {
- authMethod = AuthMethod.KERBEROS;
- }
-
- if (LOG.isDebugEnabled()) {
- LOG.debug("Use " + authMethod + " authentication for service " + remoteId.serviceName +
- ", sasl=" + useSasl);
- }
- reloginMaxBackoff = conf.getInt("hbase.security.relogin.maxbackoff", 5000);
- this.remoteId = remoteId;
-
- ConnectionHeader.Builder builder = ConnectionHeader.newBuilder();
- builder.setServiceName(remoteId.getServiceName());
- UserInformation userInfoPB = getUserInfo(ticket);
- if (userInfoPB != null) {
- builder.setUserInfo(userInfoPB);
- }
- if (this.codec != null) {
- builder.setCellBlockCodecClass(this.codec.getClass().getCanonicalName());
- }
- if (this.compressor != null) {
- builder.setCellBlockCompressorClass(this.compressor.getClass().getCanonicalName());
- }
- builder.setVersionInfo(ProtobufUtil.getVersionInfo());
- this.header = builder.build();
-
- this.setName("IPC Client (" + socketFactory.hashCode() +") connection to " +
- remoteId.getAddress().toString() +
- ((ticket==null)?" from an unknown user": (" from "
- + ticket.getUserName())));
- this.setDaemon(true);
-
- if (conf.getBoolean(SPECIFIC_WRITE_THREAD, false)) {
- callSender = new CallSender(getName(), conf);
- callSender.start();
- } else {
- callSender = null;
- }
- }
-
- private synchronized UserInformation getUserInfo(UserGroupInformation ugi) {
- if (ugi == null || authMethod == AuthMethod.DIGEST) {
- // Don't send user for token auth
- return null;
- }
- UserInformation.Builder userInfoPB = UserInformation.newBuilder();
- if (authMethod == AuthMethod.KERBEROS) {
- // Send effective user for Kerberos auth
- userInfoPB.setEffectiveUser(ugi.getUserName());
- } else if (authMethod == AuthMethod.SIMPLE) {
- //Send both effective user and real user for simple auth
- userInfoPB.setEffectiveUser(ugi.getUserName());
- if (ugi.getRealUser() != null) {
- userInfoPB.setRealUser(ugi.getRealUser().getUserName());
- }
- }
- return userInfoPB.build();
- }
-
- protected synchronized void setupConnection() throws IOException {
- short ioFailures = 0;
- short timeoutFailures = 0;
- while (true) {
- try {
- this.socket = socketFactory.createSocket();
- this.socket.setTcpNoDelay(tcpNoDelay);
- this.socket.setKeepAlive(tcpKeepAlive);
- if (localAddr != null) {
- this.socket.bind(localAddr);
- }
- NetUtils.connect(this.socket, remoteId.getAddress(), connectTO);
- this.socket.setSoTimeout(readTO);
- return;
- } catch (SocketTimeoutException toe) {
- /* The max number of retries is 45,
- * which amounts to 20s*45 = 15 minutes retries.
- */
- handleConnectionFailure(timeoutFailures++, maxRetries, toe);
- } catch (IOException ie) {
- handleConnectionFailure(ioFailures++, maxRetries, ie);
- }
- }
- }
-
- protected synchronized void closeConnection() {
- if (socket == null) {
- return;
- }
-
- // close the current connection
- try {
- if (socket.getOutputStream() != null) {
- socket.getOutputStream().close();
- }
- } catch (IOException ignored) { // Can happen if the socket is already closed
- if (LOG.isTraceEnabled()) LOG.trace("ignored", ignored);
- }
- try {
- if (socket.getInputStream() != null) {
- socket.getInputStream().close();
- }
- } catch (IOException ignored) { // Can happen if the socket is already closed
- if (LOG.isTraceEnabled()) LOG.trace("ignored", ignored);
- }
- try {
- if (socket.getChannel() != null) {
- socket.getChannel().close();
- }
- } catch (IOException ignored) { // Can happen if the socket is already closed
- if (LOG.isTraceEnabled()) LOG.trace("ignored", ignored);
- }
- try {
- socket.close();
- } catch (IOException e) {
- LOG.warn("Not able to close a socket", e);
- }
-
- // set socket to null so that the next call to setupIOstreams
- // can start the process of connect all over again.
- socket = null;
- }
-
- /**
- * Handle connection failures
- *
- * If the current number of retries is equal to the max number of retries,
- * stop retrying and throw the exception; Otherwise backoff N seconds and
- * try connecting again.
- *
- * This Method is only called from inside setupIOstreams(), which is
- * synchronized. Hence the sleep is synchronized; the locks will be retained.
- *
- * @param curRetries current number of retries
- * @param maxRetries max number of retries allowed
- * @param ioe failure reason
- * @throws IOException if max number of retries is reached
- */
- private void handleConnectionFailure(int curRetries, int maxRetries, IOException ioe)
- throws IOException {
- closeConnection();
-
- // throw the exception if the maximum number of retries is reached
- if (curRetries >= maxRetries || ExceptionUtil.isInterrupt(ioe)) {
- throw ioe;
- }
-
- // otherwise back off and retry
- try {
- Thread.sleep(failureSleep);
- } catch (InterruptedException ie) {
- ExceptionUtil.rethrowIfInterrupt(ie);
- }
-
- LOG.info("Retrying connect to server: " + remoteId.getAddress() +
- " after sleeping " + failureSleep + "ms. Already tried " + curRetries +
- " time(s).");
- }
-
- /**
- * @throws IOException if the connection is not open.
- */
- private void checkIsOpen() throws IOException {
- if (shouldCloseConnection.get()) {
- throw new ConnectionClosingException(getName() + " is closing");
- }
- }
-
- /* wait till someone signals us to start reading RPC response or
- * it is idle too long, it is marked as to be closed,
- * or the client is marked as not running.
- *
- * @return true if it is time to read a response; false otherwise.
- */
- protected synchronized boolean waitForWork() throws InterruptedException {
- // beware of the concurrent access to the calls list: we can add calls, but as well
- // remove them.
- long waitUntil = EnvironmentEdgeManager.currentTime() + minIdleTimeBeforeClose;
-
- while (true) {
- if (shouldCloseConnection.get()) {
- return false;
- }
-
- if (!running.get()) {
- markClosed(new IOException("stopped with " + calls.size() + " pending request(s)"));
- return false;
- }
-
- if (!calls.isEmpty()) {
- // shouldCloseConnection can be set to true by a parallel thread here. The caller
- // will need to check anyway.
- return true;
- }
-
- if (EnvironmentEdgeManager.currentTime() >= waitUntil) {
- // Connection is idle.
- // We expect the number of calls to be zero here, but actually someone can
- // adds a call at the any moment, as there is no synchronization between this task
- // and adding new calls. It's not a big issue, but it will get an exception.
- markClosed(new IOException(
- "idle connection closed with " + calls.size() + " pending request(s)"));
- return false;
- }
-
- wait(Math.min(minIdleTimeBeforeClose, 1000));
- }
- }
-
- public InetSocketAddress getRemoteAddress() {
- return remoteId.getAddress();
- }
-
- @Override
- public void run() {
- if (LOG.isTraceEnabled()) {
- LOG.trace(getName() + ": starting, connections " + connections.size());
- }
-
- try {
- while (waitForWork()) { // Wait here for work - read or close connection
- readResponse();
- }
- } catch (InterruptedException t) {
- if (LOG.isTraceEnabled()) {
- LOG.trace(getName() + ": interrupted while waiting for call responses");
- }
- markClosed(ExceptionUtil.asInterrupt(t));
- } catch (Throwable t) {
- if (LOG.isDebugEnabled()) {
- LOG.debug(getName() + ": unexpected throwable while waiting for call responses", t);
- }
- markClosed(new IOException("Unexpected throwable while waiting call responses", t));
- }
-
- close();
-
- if (LOG.isTraceEnabled()) {
- LOG.trace(getName() + ": stopped, connections " + connections.size());
- }
- }
-
- private synchronized void disposeSasl() {
- if (saslRpcClient != null) {
- try {
- saslRpcClient.dispose();
- saslRpcClient = null;
- } catch (IOException ioe) {
- LOG.error("Error disposing of SASL client", ioe);
- }
- }
- }
-
- private synchronized boolean shouldAuthenticateOverKrb() throws IOException {
- UserGroupInformation loginUser = UserGroupInformation.getLoginUser();
- UserGroupInformation currentUser =
- UserGroupInformation.getCurrentUser();
- UserGroupInformation realUser = currentUser.getRealUser();
- return authMethod == AuthMethod.KERBEROS &&
- loginUser != null &&
- //Make sure user logged in using Kerberos either keytab or TGT
- loginUser.hasKerberosCredentials() &&
- // relogin only in case it is the login user (e.g. JT)
- // or superuser (like oozie).
- (loginUser.equals(currentUser) || loginUser.equals(realUser));
- }
-
- private synchronized boolean setupSaslConnection(final InputStream in2,
- final OutputStream out2) throws IOException {
- saslRpcClient = new HBaseSaslRpcClient(authMethod, token, serverPrincipal, fallbackAllowed,
- conf.get("hbase.rpc.protection",
- QualityOfProtection.AUTHENTICATION.name().toLowerCase(Locale.ROOT)));
- return saslRpcClient.saslConnect(in2, out2);
- }
-
- /**
- * If multiple clients with the same principal try to connect
- * to the same server at the same time, the server assumes a
- * replay attack is in progress. This is a feature of kerberos.
- * In order to work around this, what is done is that the client
- * backs off randomly and tries to initiate the connection
- * again.
- * The other problem is to do with ticket expiry. To handle that,
- * a relogin is attempted.
- * <p>
- * The retry logic is governed by the {@link #shouldAuthenticateOverKrb}
- * method. In case when the user doesn't have valid credentials, we don't
- * need to retry (from cache or ticket). In such cases, it is prudent to
- * throw a runtime exception when we receive a SaslException from the
- * underlying authentication implementation, so there is no retry from
- * other high level (for eg, HCM or HBaseAdmin).
- * </p>
- */
- private synchronized void handleSaslConnectionFailure(
- final int currRetries,
- final int maxRetries, final Exception ex, final Random rand,
- final UserGroupInformation user)
- throws IOException, InterruptedException{
- user.doAs(new PrivilegedExceptionAction<Object>() {
- @Override
- public Object run() throws IOException, InterruptedException {
- closeConnection();
- if (shouldAuthenticateOverKrb()) {
- if (currRetries < maxRetries) {
- if (LOG.isDebugEnabled()) {
- LOG.debug("Exception encountered while connecting to " +
- "the server : " + ex);
- }
- //try re-login
- if (UserGroupInformation.isLoginKeytabBased()) {
- UserGroupInformation.getLoginUser().reloginFromKeytab();
- } else {
- UserGroupInformation.getLoginUser().reloginFromTicketCache();
- }
- disposeSasl();
- //have granularity of milliseconds
- //we are sleeping with the Connection lock held but since this
- //connection instance is being used for connecting to the server
- //in question, it is okay
- Thread.sleep((rand.nextInt(reloginMaxBackoff) + 1));
- return null;
- } else {
- String msg = "Couldn't setup connection for " +
- UserGroupInformation.getLoginUser().getUserName() +
- " to " + serverPrincipal;
- LOG.warn(msg, ex);
- throw (IOException) new IOException(msg).initCause(ex);
- }
- } else {
- LOG.warn("Exception encountered while connecting to " +
- "the server : " + ex);
- }
- if (ex instanceof RemoteException) {
- throw (RemoteException)ex;
- }
- if (ex instanceof SaslException) {
- String msg = "SASL authentication failed." +
- " The most likely cause is missing or invalid credentials." +
- " Consider 'kinit'.";
- LOG.fatal(msg, ex);
- throw new RuntimeException(msg, ex);
- }
- throw new IOException(ex);
- }
- });
- }
-
- protected synchronized void setupIOstreams() throws IOException {
- if (socket != null) {
- // The connection is already available. Perfect.
- return;
- }
-
- if (shouldCloseConnection.get()){
- throw new ConnectionClosingException("This connection is closing");
- }
-
- if (failedServers.isFailedServer(remoteId.getAddress())) {
- if (LOG.isDebugEnabled()) {
- LOG.debug("Not trying to connect to " + server +
- " this server is in the failed servers list");
- }
- IOException e = new FailedServerException(
- "This server is in the failed servers list: " + server);
- markClosed(e);
- close();
- throw e;
- }
-
- try {
- if (LOG.isDebugEnabled()) {
- LOG.debug("Connecting to " + server);
- }
- short numRetries = 0;
- final short MAX_RETRIES = 5;
- Random rand = null;
- while (true) {
- setupConnection();
- InputStream inStream = NetUtils.getInputStream(socket);
- // This creates a socket with a write timeout. This timeout cannot be changed.
- OutputStream outStream = NetUtils.getOutputStream(socket, writeTO);
- // Write out the preamble -- MAGIC, version, and auth to use.
- writeConnectionHeaderPreamble(outStream);
- if (useSasl) {
- final InputStream in2 = inStream;
- final OutputStream out2 = outStream;
- UserGroupInformation ticket = remoteId.getTicket().getUGI();
- if (authMethod == AuthMethod.KERBEROS) {
- if (ticket != null && ticket.getRealUser() != null) {
- ticket = ticket.getRealUser();
- }
- }
- boolean continueSasl;
- if (ticket == null) throw new FatalConnectionException("ticket/user is null");
- try {
- continueSasl = ticket.doAs(new PrivilegedExceptionAction<Boolean>() {
- @Override
- public Boolean run() throws IOException {
- return setupSaslConnection(in2, out2);
- }
- });
- } catch (Exception ex) {
- ExceptionUtil.rethrowIfInterrupt(ex);
- if (rand == null) {
- rand = new Random();
- }
- handleSaslConnectionFailure(numRetries++, MAX_RETRIES, ex, rand, ticket);
- continue;
- }
- if (continueSasl) {
- // Sasl connect is successful. Let's set up Sasl i/o streams.
- inStream = saslRpcClient.getInputStream(inStream);
- outStream = saslRpcClient.getOutputStream(outStream);
- } else {
- // fall back to simple auth because server told us so.
- authMethod = AuthMethod.SIMPLE;
- useSasl = false;
- }
- }
- this.in = new DataInputStream(new BufferedInputStream(inStream));
- synchronized (this.outLock) {
- this.out = new DataOutputStream(new BufferedOutputStream(outStream));
- }
- // Now write out the connection header
- writeConnectionHeader();
-
- // start the receiver thread after the socket connection has been set up
- start();
- return;
- }
- } catch (Throwable t) {
- IOException e = ExceptionUtil.asInterrupt(t);
- if (e == null) {
- failedServers.addToFailedServers(remoteId.address);
- if (t instanceof LinkageError) {
- // probably the hbase hadoop version does not match the running hadoop version
- e = new DoNotRetryIOException(t);
- } else if (t instanceof IOException) {
- e = (IOException) t;
- } else {
- e = new IOException("Could not set up IO Streams to " + server, t);
- }
- }
- markClosed(e);
- close();
- throw e;
- }
- }
-
- /**
- * Write the RPC header: <MAGIC WORD -- 'HBas'> <ONEBYTE_VERSION> <ONEBYTE_AUTH_TYPE>
- */
- private void writeConnectionHeaderPreamble(OutputStream outStream) throws IOException {
- // Assemble the preamble up in a buffer first and then send it. Writing individual elements,
- // they are getting sent across piecemeal according to wireshark and then server is messing
- // up the reading on occasion (the passed in stream is not buffered yet).
-
- // Preamble is six bytes -- 'HBas' + VERSION + AUTH_CODE
- int rpcHeaderLen = HConstants.RPC_HEADER.length;
- byte [] preamble = new byte [rpcHeaderLen + 2];
- System.arraycopy(HConstants.RPC_HEADER, 0, preamble, 0, rpcHeaderLen);
- preamble[rpcHeaderLen] = HConstants.RPC_CURRENT_VERSION;
- synchronized (this) {
- preamble[rpcHeaderLen + 1] = authMethod.code;
- }
- outStream.write(preamble);
- outStream.flush();
- }
-
- /**
- * Write the connection header.
- */
- private synchronized void writeConnectionHeader() throws IOException {
- synchronized (this.outLock) {
- this.out.writeInt(this.header.getSerializedSize());
- this.header.writeTo(this.out);
- this.out.flush();
- }
- }
-
- /** Close the connection. */
- protected synchronized void close() {
- if (!shouldCloseConnection.get()) {
- LOG.error(getName() + ": the connection is not in the closed state");
- return;
- }
-
- // release the resources
- // first thing to do;take the connection out of the connection list
- synchronized (connections) {
- connections.removeValue(remoteId, this);
- }
-
- // close the streams and therefore the socket
- synchronized(this.outLock) {
- if (this.out != null) {
- IOUtils.closeStream(out);
- this.out = null;
- }
- }
- IOUtils.closeStream(in);
- this.in = null;
- if (this.socket != null) {
- try {
- this.socket.close();
- this.socket = null;
- } catch (IOException e) {
- LOG.error("Error while closing socket", e);
- }
- }
-
- disposeSasl();
-
- // log the info
- if (LOG.isTraceEnabled()) {
- LOG.trace(getName() + ": closing ipc connection to " + server);
- }
-
- cleanupCalls(true);
-
- if (LOG.isTraceEnabled()) {
- LOG.trace(getName() + ": ipc connection to " + server + " closed");
- }
- }
-
- protected void tracedWriteRequest(Call call, int priority, Span span) throws IOException {
- TraceScope ts = Trace.startSpan("RpcClientImpl.tracedWriteRequest", span);
- try {
- writeRequest(call, priority, span);
- } finally {
- ts.close();
- }
- }
-
- /**
- * Initiates a call by sending the parameter to the remote server.
- * Note: this is not called from the Connection thread, but by other
- * threads.
- * @see #readResponse()
- */
- @edu.umd.cs.findbugs.annotations.SuppressWarnings(value="IS2_INCONSISTENT_SYNC",
- justification="Findbugs is misinterpreting locking missing fact that this.outLock is held")
- private void writeRequest(Call call, final int priority, Span span) throws IOException {
- RequestHeader.Builder builder = RequestHeader.newBuilder();
- builder.setCallId(call.id);
- if (span != null) {
- builder.setTraceInfo(
- RPCTInfo.newBuilder().setParentId(span.getSpanId()).setTraceId(span.getTraceId()));
- }
- builder.setMethodName(call.md.getName());
- builder.setRequestParam(call.param != null);
- ByteBuffer cellBlock = ipcUtil.buildCellBlock(this.codec, this.compressor, call.cells);
- if (cellBlock != null) {
- CellBlockMeta.Builder cellBlockBuilder = CellBlockMeta.newBuilder();
- cellBlockBuilder.setLength(cellBlock.limit());
- builder.setCellBlockMeta(cellBlockBuilder.build());
- }
- // Only pass priority if there is one set.
- if (priority != PayloadCarryingRpcController.PRIORITY_UNSET) {
- builder.setPriority(priority);
- }
- builder.setTimeout(call.timeout);
- RequestHeader header = builder.build();
-
- setupIOstreams();
-
- // Now we're going to write the call. We take the lock, then check that the connection
- // is still valid, and, if so we do the write to the socket. If the write fails, we don't
- // know where we stand, we have to close the connection.
- checkIsOpen();
- IOException writeException = null;
- synchronized (this.outLock) {
- if (Thread.interrupted()) throw new InterruptedIOException();
-
- calls.put(call.id, call); // We put first as we don't want the connection to become idle.
- checkIsOpen(); // Now we're checking that it didn't became idle in between.
-
- try {
- 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?
- shouldCloseConnection.set(true);
- writeException = e;
- interrupt();
- }
- }
-
- // call close outside of the synchronized (outLock) to prevent deadlock - HBASE-14474
- if (writeException != null) {
- markClosed(writeException);
- close();
- }
-
- // We added a call, and may be started the connection close. In both cases, we
- // need to notify the reader.
- doNotify();
-
- // Now that we notified, we can rethrow the exception if any. Otherwise we're good.
- if (writeException != null) throw writeException;
- }
-
- @edu.umd.cs.findbugs.annotations.SuppressWarnings(value="NN_NAKED_NOTIFY",
- justification="Presume notifyAll is because we are closing/shutting down")
- private synchronized void doNotify() {
- // Make a separate method so can do synchronize and add findbugs annotation; only one
- // annotation at at time in source 1.7.
- notifyAll(); // Findbugs: NN_NAKED_NOTIFY
- }
-
- /* Receive a response.
- * Because only one receiver, so no synchronization on in.
- */
- protected void readResponse() {
- if (shouldCloseConnection.get()) return;
- Call call = null;
- boolean expectedCall = false;
- try {
- // See HBaseServer.Call.setResponse for where we write out the response.
- // Total size of the response. Unused. But have to read it in anyways.
- int totalSize = in.readInt();
-
- // Read the header
- ResponseHeader responseHeader = ResponseHeader.parseDelimitedFrom(in);
- int id = responseHeader.getCallId();
- call = calls.remove(id); // call.done have to be set before leaving this method
- expectedCall = (call != null && !call.done);
- if (!expectedCall) {
- // So we got a response for which we have no corresponding 'call' here on the client-side.
- // We probably timed out waiting, cleaned up all references, and now the server decides
- // to return a response. There is nothing we can do w/ the response at this stage. Clean
- // out the wire of the response so its out of the way and we can get other responses on
- // this connection.
- 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);
- }
- } else {
- Message value = null;
- if (call.responseDefaultType != null) {
- Builder builder = call.responseDefaultType.newBuilderForType();
- ProtobufUtil.mergeDelimitedFrom(builder, in);
- value = builder.build();
- }
- CellScanner cellBlockScanner = null;
- if (responseHeader.hasCellBlockMeta()) {
- int size = responseHeader.getCellBlockMeta().getLength();
- byte [] cellBlock = new byte[size];
- IOUtils.readFully(this.in, cellBlock, 0, cellBlock.length);
- 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);
- if (e instanceof SocketTimeoutException) {
- // Clean up open calls but don't treat this as a fatal condition,
- // since we expect certain responses to not make it by the specified
- // {@link ConnectionId#rpcTimeout}.
- if (LOG.isTraceEnabled()) LOG.trace("ignored", e);
- } else {
- // Treat this as a fatal condition and close this connection
- markClosed(e);
- }
- } finally {
- cleanupCalls(false);
- }
- }
-
- /**
- * @return True if the exception is a fatal connection exception.
- */
- private boolean isFatalConnectionException(final ExceptionResponse e) {
- return e.getExceptionClassName().
- equals(FatalConnectionException.class.getName());
- }
-
- /**
- * @param e exception to be wrapped
- * @return RemoteException made from passed <code>e</code>
- */
- private RemoteException createRemoteException(final ExceptionResponse e) {
- String innerExceptionClassName = e.getExceptionClassName();
- boolean doNotRetry = e.getDoNotRetry();
- return e.hasHostname()?
- // If a hostname then add it to the RemoteWithExtrasException
- new RemoteWithExtrasException(innerExceptionClassName,
- e.getStackTrace(), e.getHostname(), e.getPort(), doNotRetry):
- new RemoteWithExtrasException(innerExceptionClassName,
- e.getStackTrace(), doNotRetry);
- }
-
- protected synchronized boolean markClosed(IOException e) {
- if (e == null) throw new NullPointerException();
-
- boolean ret = shouldCloseConnection.compareAndSet(false, true);
- if (ret) {
- if (LOG.isTraceEnabled()) {
- LOG.trace(getName() + ": marking at should close, reason: " + e.getMessage());
- }
- if (callSender != null) {
- callSender.close();
- }
- notifyAll();
- }
- return ret;
- }
-
-
- /**
- * Cleanup the calls older than a given timeout, in milli seconds.
- * @param allCalls true for all calls, false for only the calls in timeout
- */
- protected synchronized void cleanupCalls(boolean allCalls) {
- Iterator<Entry<Integer, Call>> itor = calls.entrySet().iterator();
- while (itor.hasNext()) {
- Call c = itor.next().getValue();
- if (c.done) {
- // To catch the calls without timeout that were cancelled.
- itor.remove();
- } else if (allCalls) {
- long waitTime = EnvironmentEdgeManager.currentTime() - c.getStartTime();
- IOException ie = new ConnectionClosingException("Connection to " + getRemoteAddress()
- + " is closing. Call id=" + c.id + ", waitTime=" + waitTime);
- c.setException(ie);
- itor.remove();
- } else if (c.checkAndSetTimeout()) {
- itor.remove();
- } else {
- // We expect the call to be ordered by timeout. It may not be the case, but stopping
- // at the first valid call allows to be sure that we still have something to do without
- // spending too much time by reading the full list.
- break;
- }
- }
- }
- }
-
- /**
- * 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, null);
- }
-
- /**
- * 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
- * @param localAddr client socket bind address
- * @param metrics the connection metrics
- */
- RpcClientImpl(Configuration conf, String clusterId, SocketFactory factory,
- SocketAddress localAddr, MetricsConnection metrics) {
- super(conf, clusterId, localAddr, metrics);
-
- this.socketFactory = factory;
- this.connections = new PoolMap<ConnectionId, Connection>(getPoolType(conf), getPoolSize(conf));
- this.failedServers = new FailedServers(conf);
- }
-
- /**
- * 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);
- }
-
- /**
- * 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,
- MetricsConnection metrics) {
- this(conf, clusterId, NetUtils.getDefaultSocketFactory(conf), localAddr, metrics);
- }
-
- /** Stop all threads related to this client. No further calls may be made
- * using this client. */
- @Override
- public void close() {
- if (LOG.isDebugEnabled()) LOG.debug("Stopping rpc client");
- if (!running.compareAndSet(true, false)) return;
-
- Set<Connection> connsToClose = null;
- // wake up all connections
- synchronized (connections) {
- for (Connection conn : connections.values()) {
- conn.interrupt();
- if (conn.callSender != null) {
- conn.callSender.interrupt();
- }
-
- // In case the CallSender did not setupIOStreams() yet, the Connection may not be started
- // at all (if CallSender has a cancelled Call it can happen). See HBASE-13851
- if (!conn.isAlive()) {
- if (connsToClose == null) {
- connsToClose = new HashSet<Connection>();
- }
- connsToClose.add(conn);
- }
- }
- }
- if (connsToClose != null) {
- for (Connection conn : connsToClose) {
- conn.markClosed(new InterruptedIOException("RpcClient is closing"));
- conn.close();
- }
- }
- // wait until all connections are closed
- while (!connections.isEmpty()) {
- try {
- Thread.sleep(10);
- } catch (InterruptedException e) {
- LOG.info("Interrupted while stopping the client. We still have " + connections.size() +
- " connections.");
- Thread.currentThread().interrupt();
- return;
- }
- }
- }
-
- /** Make a call, passing <code>param</code>, to the IPC server running at
- * <code>address</code> which is servicing the <code>protocol</code> protocol,
- * with the <code>ticket</code> credentials, returning the value.
- * Throws exceptions if there are network problems or if the remote code
- * threw an exception.
- * @param ticket Be careful which ticket you pass. A new user will mean a new Connection.
- * {@link UserProvider#getCurrent()} makes a new instance of User each time so will be a
- * new Connection each time.
- * @return A pair with the Message response and the Cell data (if any).
- * @throws InterruptedException
- * @throws IOException
- */
- @Override
- protected Pair<Message, CellScanner> call(PayloadCarryingRpcController pcrc, MethodDescriptor md,
- Message param, Message returnType, User ticket, InetSocketAddress addr,
- MetricsConnection.CallStats callStats)
- throws IOException, InterruptedException {
- if (pcrc == null) {
- pcrc = new PayloadCarryingRpcController();
- }
- CellScanner cells = pcrc.cellScanner();
-
- final Call call = new Call(this.callIdCnt.getAndIncrement(), md, param, cells, returnType,
- pcrc.getCallTimeout(), MetricsConnection.newCallStats());
-
- final Connection connection = getConnection(ticket, call, addr);
-
- final CallFuture cts;
- if (connection.callSender != null) {
- cts = connection.callSender.sendCall(call, pcrc.getPriority(), Trace.currentSpan());
- pcrc.notifyOnCancel(new RpcCallback<Object>() {
- @Override
- public void run(Object parameter) {
- connection.callSender.remove(cts);
- }
- });
- if (pcrc.isCanceled()) {
- // To finish if the call was cancelled before we set the notification (race condition)
- call.callComplete();
- return new Pair<Message, CellScanner>(call.response, call.cells);
- }
- } else {
- cts = null;
- connection.tracedWriteRequest(call, pcrc.getPriority(), Trace.currentSpan());
- }
-
- while (!call.done) {
- if (call.checkAndSetTimeout()) {
- if (cts != null) connection.callSender.remove(cts);
- break;
- }
- if (connection.shouldCloseConnection.get()) {
- throw new ConnectionClosingException("Call id=" + call.id +
- " on server " + addr + " aborted: connection is closing");
- }
- try {
- synchronized (call) {
- if (call.done) break;
- call.wait(Math.min(call.remainingTime(), 1000) + 1);
- }
- } catch (InterruptedException e) {
- call.setException(new InterruptedIOException());
- if (cts != null) connection.callSender.remove(cts);
- throw e;
- }
- }
-
- if (call.error != null) {
- if (call.error instanceof RemoteException) {
- call.error.fillInStackTrace();
- throw call.error;
- }
- // local exception
- throw wrapException(addr, call.error);
- }
-
- return new Pair<Message, CellScanner>(call.response, call.cells);
- }
-
-
- /**
- * Interrupt the connections to the given ip:port server. This should be called if the server
- * is known as actually dead. This will not prevent current operation to be retried, and,
- * depending on their own behavior, they may retry on the same server. This can be a feature,
- * for example at startup. In any case, they're likely to get connection refused (if the
- * process died) or no route to host: i.e. their next retries should be faster and with a
- * safe exception.
- */
- @Override
- public void cancelConnections(ServerName sn) {
- synchronized (connections) {
- for (Connection connection : connections.values()) {
- if (connection.isAlive() &&
- connection.getRemoteAddress().getPort() == sn.getPort() &&
- connection.getRemoteAddress().getHostName().equals(sn.getHostname())) {
- LOG.info("The server on " + sn.toString() +
- " is dead - stopping the connection " + connection.remoteId);
- connection.interrupt(); // We're interrupting a Reader. It means we want it to finish.
- // This will close the connection as well.
- }
- }
- }
- }
-
- /**
- * Get a connection from the pool, or create a new one and add it to the
- * pool. Connections to a given host/port are reused.
- */
- protected Connection getConnection(User ticket, Call call, InetSocketAddress addr)
- throws IOException {
- if (!running.get()) throw new StoppedRpcClientException();
- Connection connection;
- ConnectionId remoteId =
- new ConnectionId(ticket, call.md.getService().getName(), addr);
- synchronized (connections) {
- connection = connections.get(remoteId);
- if (connection == null) {
- connection = createConnection(remoteId, this.codec, this.compressor);
- connections.put(remoteId, connection);
- }
- }
-
- return connection;
- }
-}
http://git-wip-us.apache.org/repos/asf/hbase/blob/094e9a31/hbase-client/src/main/java/org/apache/hadoop/hbase/ipc/RpcConnection.java
----------------------------------------------------------------------
diff --git a/hbase-client/src/main/java/org/apache/hadoop/hbase/ipc/RpcConnection.java b/hbase-client/src/main/java/org/apache/hadoop/hbase/ipc/RpcConnection.java
new file mode 100644
index 0000000..5e9e97e
--- /dev/null
+++ b/hbase-client/src/main/java/org/apache/hadoop/hbase/ipc/RpcConnection.java
@@ -0,0 +1,260 @@
+/**
+ * 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.ipc;
+
+import io.netty.util.HashedWheelTimer;
+import io.netty.util.Timeout;
+import io.netty.util.TimerTask;
+
+import java.io.IOException;
+import java.net.UnknownHostException;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hbase.HConstants;
+import org.apache.hadoop.hbase.classification.InterfaceAudience;
+import org.apache.hadoop.hbase.codec.Codec;
+import org.apache.hadoop.hbase.protobuf.ProtobufUtil;
+import org.apache.hadoop.hbase.protobuf.generated.AuthenticationProtos;
+import org.apache.hadoop.hbase.protobuf.generated.RPCProtos.ConnectionHeader;
+import org.apache.hadoop.hbase.protobuf.generated.RPCProtos.UserInformation;
+import org.apache.hadoop.hbase.security.AuthMethod;
+import org.apache.hadoop.hbase.security.SecurityInfo;
+import org.apache.hadoop.hbase.util.EnvironmentEdgeManager;
+import org.apache.hadoop.io.Text;
+import org.apache.hadoop.io.compress.CompressionCodec;
+import org.apache.hadoop.security.SecurityUtil;
+import org.apache.hadoop.security.UserGroupInformation;
+import org.apache.hadoop.security.token.Token;
+import org.apache.hadoop.security.token.TokenIdentifier;
+import org.apache.hadoop.security.token.TokenSelector;
+
+/**
+ * Base class for ipc connection.
+ */
+@InterfaceAudience.Private
+abstract class RpcConnection {
+
+ private static final Log LOG = LogFactory.getLog(RpcConnection.class);
+
+ protected final ConnectionId remoteId;
+
+ protected final AuthMethod authMethod;
+
+ protected final boolean useSasl;
+
+ protected final Token<? extends TokenIdentifier> token;
+
+ protected final String serverPrincipal; // server's krb5 principal name
+
+ protected final int reloginMaxBackoff; // max pause before relogin on sasl failure
+
+ protected final Codec codec;
+
+ protected final CompressionCodec compressor;
+
+ protected final HashedWheelTimer timeoutTimer;
+
+ // the last time we were picked up from connection pool.
+ protected long lastTouched;
+
+ protected RpcConnection(Configuration conf, HashedWheelTimer timeoutTimer, ConnectionId remoteId,
+ String clusterId, boolean isSecurityEnabled, Codec codec, CompressionCodec compressor)
+ throws IOException {
+ if (remoteId.getAddress().isUnresolved()) {
+ throw new UnknownHostException("unknown host: " + remoteId.getAddress().getHostName());
+ }
+ this.timeoutTimer = timeoutTimer;
+ this.codec = codec;
+ this.compressor = compressor;
+
+ UserGroupInformation ticket = remoteId.getTicket().getUGI();
+ SecurityInfo securityInfo = SecurityInfo.getInfo(remoteId.getServiceName());
+ this.useSasl = isSecurityEnabled;
+ Token<? extends TokenIdentifier> token = null;
+ String serverPrincipal = null;
+ if (useSasl && securityInfo != null) {
+ AuthenticationProtos.TokenIdentifier.Kind tokenKind = securityInfo.getTokenKind();
+ if (tokenKind != null) {
+ TokenSelector<? extends TokenIdentifier> tokenSelector = AbstractRpcClient.TOKEN_HANDLERS
+ .get(tokenKind);
+ if (tokenSelector != null) {
+ token = tokenSelector.selectToken(new Text(clusterId), ticket.getTokens());
+ } else if (LOG.isDebugEnabled()) {
+ LOG.debug("No token selector found for type " + tokenKind);
+ }
+ }
+ String serverKey = securityInfo.getServerPrincipal();
+ if (serverKey == null) {
+ throw new IOException("Can't obtain server Kerberos config key from SecurityInfo");
+ }
+ serverPrincipal = SecurityUtil.getServerPrincipal(conf.get(serverKey),
+ remoteId.address.getAddress().getCanonicalHostName().toLowerCase());
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("RPC Server Kerberos principal name for service=" + remoteId.getServiceName()
+ + " is " + serverPrincipal);
+ }
+ }
+ this.token = token;
+ this.serverPrincipal = serverPrincipal;
+ if (!useSasl) {
+ authMethod = AuthMethod.SIMPLE;
+ } else if (token != null) {
+ authMethod = AuthMethod.DIGEST;
+ } else {
+ authMethod = AuthMethod.KERBEROS;
+ }
+
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Use " + authMethod + " authentication for service " + remoteId.serviceName
+ + ", sasl=" + useSasl);
+ }
+ reloginMaxBackoff = conf.getInt("hbase.security.relogin.maxbackoff", 5000);
+ this.remoteId = remoteId;
+ }
+
+ private UserInformation getUserInfo(UserGroupInformation ugi) {
+ if (ugi == null || authMethod == AuthMethod.DIGEST) {
+ // Don't send user for token auth
+ return null;
+ }
+ UserInformation.Builder userInfoPB = UserInformation.newBuilder();
+ if (authMethod == AuthMethod.KERBEROS) {
+ // Send effective user for Kerberos auth
+ userInfoPB.setEffectiveUser(ugi.getUserName());
+ } else if (authMethod == AuthMethod.SIMPLE) {
+ // Send both effective user and real user for simple auth
+ userInfoPB.setEffectiveUser(ugi.getUserName());
+ if (ugi.getRealUser() != null) {
+ userInfoPB.setRealUser(ugi.getRealUser().getUserName());
+ }
+ }
+ return userInfoPB.build();
+ }
+
+ protected UserGroupInformation getUGI() {
+ UserGroupInformation ticket = remoteId.getTicket().getUGI();
+ if (authMethod == AuthMethod.KERBEROS) {
+ if (ticket != null && ticket.getRealUser() != null) {
+ ticket = ticket.getRealUser();
+ }
+ }
+ return ticket;
+ }
+
+ protected boolean shouldAuthenticateOverKrb() throws IOException {
+ UserGroupInformation loginUser = UserGroupInformation.getLoginUser();
+ UserGroupInformation currentUser = UserGroupInformation.getCurrentUser();
+ UserGroupInformation realUser = currentUser.getRealUser();
+ return authMethod == AuthMethod.KERBEROS && loginUser != null &&
+ // Make sure user logged in using Kerberos either keytab or TGT
+ loginUser.hasKerberosCredentials() &&
+ // relogin only in case it is the login user (e.g. JT)
+ // or superuser (like oozie).
+ (loginUser.equals(currentUser) || loginUser.equals(realUser));
+ }
+
+ protected void relogin() throws IOException {
+ if (UserGroupInformation.isLoginKeytabBased()) {
+ UserGroupInformation.getLoginUser().reloginFromKeytab();
+ } else {
+ UserGroupInformation.getLoginUser().reloginFromTicketCache();
+ }
+ }
+
+ protected void scheduleTimeoutTask(final Call call) {
+ if (call.timeout > 0) {
+ call.timeoutTask = timeoutTimer.newTimeout(new TimerTask() {
+
+ @Override
+ public void run(Timeout timeout) throws Exception {
+ call.setTimeout(new CallTimeoutException("Call id=" + call.id + ", waitTime="
+ + (EnvironmentEdgeManager.currentTime() - call.getStartTime()) + ", rpcTimetout="
+ + call.timeout));
+ callTimeout(call);
+ }
+ }, call.timeout, TimeUnit.MILLISECONDS);
+ }
+ }
+
+ protected byte[] getConnectionHeaderPreamble() {
+ // Assemble the preamble up in a buffer first and then send it. Writing individual elements,
+ // they are getting sent across piecemeal according to wireshark and then server is messing
+ // up the reading on occasion (the passed in stream is not buffered yet).
+
+ // Preamble is six bytes -- 'HBas' + VERSION + AUTH_CODE
+ int rpcHeaderLen = HConstants.RPC_HEADER.length;
+ byte[] preamble = new byte[rpcHeaderLen + 2];
+ System.arraycopy(HConstants.RPC_HEADER, 0, preamble, 0, rpcHeaderLen);
+ preamble[rpcHeaderLen] = HConstants.RPC_CURRENT_VERSION;
+ synchronized (this) {
+ preamble[rpcHeaderLen + 1] = authMethod.code;
+ }
+ return preamble;
+ }
+
+ protected ConnectionHeader getConnectionHeader() {
+ ConnectionHeader.Builder builder = ConnectionHeader.newBuilder();
+ builder.setServiceName(remoteId.getServiceName());
+ UserInformation userInfoPB;
+ if ((userInfoPB = getUserInfo(remoteId.ticket.getUGI())) != null) {
+ builder.setUserInfo(userInfoPB);
+ }
+ if (this.codec != null) {
+ builder.setCellBlockCodecClass(this.codec.getClass().getCanonicalName());
+ }
+ if (this.compressor != null) {
+ builder.setCellBlockCompressorClass(this.compressor.getClass().getCanonicalName());
+ }
+ builder.setVersionInfo(ProtobufUtil.getVersionInfo());
+ return builder.build();
+ }
+
+ protected abstract void callTimeout(Call call);
+
+ public ConnectionId remoteId() {
+ return remoteId;
+ }
+
+ public long getLastTouched() {
+ return lastTouched;
+ }
+
+ public void setLastTouched(long lastTouched) {
+ this.lastTouched = lastTouched;
+ }
+
+ /**
+ * Tell the idle connection sweeper whether we could be swept.
+ */
+ public abstract boolean isActive();
+
+ /**
+ * Just close connection. Do not need to remove from connection pool.
+ */
+ public abstract void shutdown();
+
+ public abstract void sendRequest(Call call, HBaseRpcController hrc) throws IOException;
+
+ /**
+ * Does the clean up work after the connection is removed from the connection pool
+ */
+ public abstract void cleanupConnection();
+}
http://git-wip-us.apache.org/repos/asf/hbase/blob/094e9a31/hbase-client/src/main/java/org/apache/hadoop/hbase/ipc/RpcControllerFactory.java
----------------------------------------------------------------------
diff --git a/hbase-client/src/main/java/org/apache/hadoop/hbase/ipc/RpcControllerFactory.java b/hbase-client/src/main/java/org/apache/hadoop/hbase/ipc/RpcControllerFactory.java
index faeca8d..ca4073d 100644
--- a/hbase-client/src/main/java/org/apache/hadoop/hbase/ipc/RpcControllerFactory.java
+++ b/hbase-client/src/main/java/org/apache/hadoop/hbase/ipc/RpcControllerFactory.java
@@ -28,7 +28,7 @@ import org.apache.hadoop.hbase.classification.InterfaceAudience;
import org.apache.hadoop.hbase.util.ReflectionUtils;
/**
- * Factory to create a {@link PayloadCarryingRpcController}
+ * Factory to create a {@link HBaseRpcController}.
*/
@InterfaceAudience.Private
public class RpcControllerFactory {
@@ -46,16 +46,16 @@ public class RpcControllerFactory {
this.conf = conf;
}
- public PayloadCarryingRpcController newController() {
- return new PayloadCarryingRpcController();
+ public HBaseRpcController newController() {
+ return new HBaseRpcControllerImpl();
}
- public PayloadCarryingRpcController newController(final CellScanner cellScanner) {
- return new PayloadCarryingRpcController(cellScanner);
+ public HBaseRpcController newController(final CellScanner cellScanner) {
+ return new HBaseRpcControllerImpl(cellScanner);
}
- public PayloadCarryingRpcController newController(final List<CellScannable> cellIterables) {
- return new PayloadCarryingRpcController(cellIterables);
+ public HBaseRpcController newController(final List<CellScannable> cellIterables) {
+ return new HBaseRpcControllerImpl(cellIterables);
}
http://git-wip-us.apache.org/repos/asf/hbase/blob/094e9a31/hbase-client/src/main/java/org/apache/hadoop/hbase/ipc/TimeLimitedRpcController.java
----------------------------------------------------------------------
diff --git a/hbase-client/src/main/java/org/apache/hadoop/hbase/ipc/TimeLimitedRpcController.java b/hbase-client/src/main/java/org/apache/hadoop/hbase/ipc/TimeLimitedRpcController.java
deleted file mode 100644
index de502cb..0000000
--- a/hbase-client/src/main/java/org/apache/hadoop/hbase/ipc/TimeLimitedRpcController.java
+++ /dev/null
@@ -1,142 +0,0 @@
-/**
- * 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.ipc;
-
-import java.io.IOException;
-import java.util.concurrent.atomic.AtomicReference;
-
-import org.apache.hadoop.hbase.classification.InterfaceAudience;
-
-import com.google.protobuf.RpcCallback;
-import com.google.protobuf.RpcController;
-
-@InterfaceAudience.Private
-public class TimeLimitedRpcController implements RpcController {
-
- /**
- * The time, in ms before the call should expire.
- */
- protected volatile Integer callTimeout;
- protected volatile boolean cancelled = false;
- protected final AtomicReference<RpcCallback<Object>> cancellationCb =
- new AtomicReference<RpcCallback<Object>>(null);
-
- protected final AtomicReference<RpcCallback<IOException>> failureCb =
- new AtomicReference<RpcCallback<IOException>>(null);
-
- private IOException exception;
-
- public int getCallTimeout() {
- if (callTimeout != null) {
- return callTimeout;
- } else {
- return 0;
- }
- }
-
- public void setCallTimeout(int callTimeout) {
- this.callTimeout = callTimeout;
- }
-
- public boolean hasCallTimeout(){
- return callTimeout != null;
- }
-
- @Override
- public String errorText() {
- if (exception != null) {
- return exception.getMessage();
- } else {
- return null;
- }
- }
-
- /**
- * For use in async rpc clients
- * @return true if failed
- */
- @Override
- public boolean failed() {
- return this.exception != null;
- }
-
- @Override
- public boolean isCanceled() {
- return cancelled;
- }
-
- @Override
- public void notifyOnCancel(RpcCallback<Object> cancellationCb) {
- this.cancellationCb.set(cancellationCb);
- if (this.cancelled) {
- cancellationCb.run(null);
- }
- }
-
- /**
- * Notify a callback on error.
- * For use in async rpc clients
- *
- * @param failureCb the callback to call on error
- */
- public void notifyOnFail(RpcCallback<IOException> failureCb) {
- this.failureCb.set(failureCb);
- if (this.exception != null) {
- failureCb.run(this.exception);
- }
- }
-
- @Override
- public void reset() {
- exception = null;
- cancelled = false;
- failureCb.set(null);
- cancellationCb.set(null);
- callTimeout = null;
- }
-
- @Override
- public void setFailed(String reason) {
- this.exception = new IOException(reason);
- if (this.failureCb.get() != null) {
- this.failureCb.get().run(this.exception);
- }
- }
-
- /**
- * Set failed with an exception to pass on.
- * For use in async rpc clients
- *
- * @param e exception to set with
- */
- public void setFailed(IOException e) {
- this.exception = e;
- if (this.failureCb.get() != null) {
- this.failureCb.get().run(this.exception);
- }
- }
-
- @Override
- public void startCancel() {
- cancelled = true;
- if (cancellationCb.get() != null) {
- cancellationCb.get().run(null);
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/hbase/blob/094e9a31/hbase-client/src/main/java/org/apache/hadoop/hbase/security/AbstractHBaseSaslRpcClient.java
----------------------------------------------------------------------
diff --git a/hbase-client/src/main/java/org/apache/hadoop/hbase/security/AbstractHBaseSaslRpcClient.java b/hbase-client/src/main/java/org/apache/hadoop/hbase/security/AbstractHBaseSaslRpcClient.java
new file mode 100644
index 0000000..cc0114b
--- /dev/null
+++ b/hbase-client/src/main/java/org/apache/hadoop/hbase/security/AbstractHBaseSaslRpcClient.java
@@ -0,0 +1,197 @@
+/**
+ * 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.security;
+
+import com.google.common.annotations.VisibleForTesting;
+
+import java.io.IOException;
+import java.util.Map;
+
+import javax.security.auth.callback.Callback;
+import javax.security.auth.callback.CallbackHandler;
+import javax.security.auth.callback.NameCallback;
+import javax.security.auth.callback.PasswordCallback;
+import javax.security.auth.callback.UnsupportedCallbackException;
+import javax.security.sasl.RealmCallback;
+import javax.security.sasl.RealmChoiceCallback;
+import javax.security.sasl.Sasl;
+import javax.security.sasl.SaslClient;
+import javax.security.sasl.SaslException;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.hbase.classification.InterfaceAudience;
+import org.apache.hadoop.security.token.Token;
+import org.apache.hadoop.security.token.TokenIdentifier;
+
+/**
+ * A utility class that encapsulates SASL logic for RPC client. Copied from
+ * <code>org.apache.hadoop.security</code>
+ */
+@InterfaceAudience.Private
+public abstract class AbstractHBaseSaslRpcClient {
+
+ private static final Log LOG = LogFactory.getLog(AbstractHBaseSaslRpcClient.class);
+
+ private static final byte[] EMPTY_TOKEN = new byte[0];
+
+ protected final SaslClient saslClient;
+
+ protected final boolean fallbackAllowed;
+
+ protected final Map<String, String> saslProps;
+
+ /**
+ * Create a HBaseSaslRpcClient for an authentication method
+ * @param method the requested authentication method
+ * @param token token to use if needed by the authentication method
+ * @param serverPrincipal the server principal that we are trying to set the connection up to
+ * @param fallbackAllowed does the client allow fallback to simple authentication
+ * @throws IOException
+ */
+ protected AbstractHBaseSaslRpcClient(AuthMethod method, Token<? extends TokenIdentifier> token,
+ String serverPrincipal, boolean fallbackAllowed) throws IOException {
+ this(method, token, serverPrincipal, fallbackAllowed, "authentication");
+ }
+
+ /**
+ * Create a HBaseSaslRpcClient for an authentication method
+ * @param method the requested authentication method
+ * @param token token to use if needed by the authentication method
+ * @param serverPrincipal the server principal that we are trying to set the connection up to
+ * @param fallbackAllowed does the client allow fallback to simple authentication
+ * @param rpcProtection the protection level ("authentication", "integrity" or "privacy")
+ * @throws IOException
+ */
+ protected AbstractHBaseSaslRpcClient(AuthMethod method, Token<? extends TokenIdentifier> token,
+ String serverPrincipal, boolean fallbackAllowed, String rpcProtection) throws IOException {
+ this.fallbackAllowed = fallbackAllowed;
+ saslProps = SaslUtil.initSaslProperties(rpcProtection);
+ switch (method) {
+ case DIGEST:
+ if (LOG.isDebugEnabled()) LOG.debug("Creating SASL " + AuthMethod.DIGEST.getMechanismName()
+ + " client to authenticate to service at " + token.getService());
+ saslClient = createDigestSaslClient(new String[] { AuthMethod.DIGEST.getMechanismName() },
+ SaslUtil.SASL_DEFAULT_REALM, new SaslClientCallbackHandler(token));
+ break;
+ case KERBEROS:
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Creating SASL " + AuthMethod.KERBEROS.getMechanismName()
+ + " client. Server's Kerberos principal name is " + serverPrincipal);
+ }
+ if (serverPrincipal == null || serverPrincipal.length() == 0) {
+ throw new IOException("Failed to specify server's Kerberos principal name");
+ }
+ String[] names = SaslUtil.splitKerberosName(serverPrincipal);
+ if (names.length != 3) {
+ throw new IOException(
+ "Kerberos principal does not have the expected format: " + serverPrincipal);
+ }
+ saslClient = createKerberosSaslClient(
+ new String[] { AuthMethod.KERBEROS.getMechanismName() }, names[0], names[1]);
+ break;
+ default:
+ throw new IOException("Unknown authentication method " + method);
+ }
+ if (saslClient == null) {
+ throw new IOException("Unable to find SASL client implementation");
+ }
+ }
+
+ protected SaslClient createDigestSaslClient(String[] mechanismNames, String saslDefaultRealm,
+ CallbackHandler saslClientCallbackHandler) throws IOException {
+ return Sasl.createSaslClient(mechanismNames, null, null, saslDefaultRealm, saslProps,
+ saslClientCallbackHandler);
+ }
+
+ protected SaslClient createKerberosSaslClient(String[] mechanismNames, String userFirstPart,
+ String userSecondPart) throws IOException {
+ return Sasl.createSaslClient(mechanismNames, null, userFirstPart, userSecondPart, saslProps,
+ null);
+ }
+
+ public byte[] getInitialResponse() throws SaslException {
+ if (saslClient.hasInitialResponse()) {
+ return saslClient.evaluateChallenge(EMPTY_TOKEN);
+ } else {
+ return EMPTY_TOKEN;
+ }
+ }
+
+ public boolean isComplete() {
+ return saslClient.isComplete();
+ }
+
+ public byte[] evaluateChallenge(byte[] challenge) throws SaslException {
+ return saslClient.evaluateChallenge(challenge);
+ }
+
+ /** Release resources used by wrapped saslClient */
+ public void dispose() {
+ SaslUtil.safeDispose(saslClient);
+ }
+
+ @VisibleForTesting
+ static class SaslClientCallbackHandler implements CallbackHandler {
+ private final String userName;
+ private final char[] userPassword;
+
+ public SaslClientCallbackHandler(Token<? extends TokenIdentifier> token) {
+ this.userName = SaslUtil.encodeIdentifier(token.getIdentifier());
+ this.userPassword = SaslUtil.encodePassword(token.getPassword());
+ }
+
+ @Override
+ public void handle(Callback[] callbacks) throws UnsupportedCallbackException {
+ NameCallback nc = null;
+ PasswordCallback pc = null;
+ RealmCallback rc = null;
+ for (Callback callback : callbacks) {
+ if (callback instanceof RealmChoiceCallback) {
+ continue;
+ } else if (callback instanceof NameCallback) {
+ nc = (NameCallback) callback;
+ } else if (callback instanceof PasswordCallback) {
+ pc = (PasswordCallback) callback;
+ } else if (callback instanceof RealmCallback) {
+ rc = (RealmCallback) callback;
+ } else {
+ throw new UnsupportedCallbackException(callback, "Unrecognized SASL client callback");
+ }
+ }
+ if (nc != null) {
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("SASL client callback: setting username: " + userName);
+ }
+ nc.setName(userName);
+ }
+ if (pc != null) {
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("SASL client callback: setting userPassword");
+ }
+ pc.setPassword(userPassword);
+ }
+ if (rc != null) {
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("SASL client callback: setting realm: " + rc.getDefaultText());
+ }
+ rc.setText(rc.getDefaultText());
+ }
+ }
+ }
+}