Package org.apache.hadoop.hbase.ipc

Class BlockingRpcConnection

java.lang.Object

org.apache.hadoop.hbase.ipc.RpcConnection

org.apache.hadoop.hbase.ipc.BlockingRpcConnection

All Implemented Interfaces:

Runnable

@Private
class BlockingRpcConnection
extends RpcConnection
implements Runnable

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.

Nested Class Summary

Nested Classes

Modifier and Type	Class	Description
private class	BlockingRpcConnection.CallSender	If the client wants to interrupt its calls easily (i.e.

Field Summary

Fields

Modifier and Type	Field	Description
private final ConcurrentMap<Integer,Call>	calls
private final BlockingRpcConnection.CallSender	callSender
private boolean	closed
private byte[]	connectionHeaderPreamble
private byte[]	connectionHeaderWithLength
private DataInputStream	in
private static final org.slf4j.Logger	LOG
private DataOutputStream	out
private final BlockingRpcClient	rpcClient
private HBaseSaslRpcClient	saslRpcClient
protected Socket	socket
private Thread	thread
private final String	threadName
private boolean	waitingConnectionHeaderResponse

Fields inherited from class org.apache.hadoop.hbase.ipc.RpcConnection

cellBlockBuilder, codec, compressor, conf, CRYPTO_AES_ENABLED_DEFAULT, CRYPTO_AES_ENABLED_KEY, lastTouched, metrics, provider, reloginMaxBackoff, remoteId, securityInfo, timeoutTimer, token, useSasl

Constructor Summary

Constructors

Constructor	Description
BlockingRpcConnection(BlockingRpcClient rpcClient, ConnectionId remoteId)

Method Summary

Modifier and Type	Method	Description
protected void	callTimeout(Call call)
private String	chooseServerPrincipal(InputStream inStream, OutputStream outStream)
void	cleanupConnection()	Does the clean up work after the connection is removed from the connection pool
private void	closeConn(IOException e)
private void	closeSocket()
private void	createStreams(InputStream inStream, OutputStream outStream)
private void	disposeSasl()
private void	getConnectionRegistry(InputStream inStream, OutputStream outStream, Call connectionRegistryCall)
private void	handleConnectionFailure(int curRetries, int maxRetries, IOException ioe)	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.
private void	handleSaslConnectionFailure(int currRetries, int maxRetries, Exception ex, org.apache.hadoop.security.UserGroupInformation user, String serverPrincipal)	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.
boolean	isActive()	Tell the idle connection sweeper whether we could be swept.
private void	negotiateCryptoAes(org.apache.hadoop.hbase.shaded.protobuf.generated.RPCProtos.CryptoCipherMeta cryptoCipherMeta)
private void	processResponseForConnectionHeader()
private void	readResponse()
void	run()
void	sendRequest(Call call, HBaseRpcController pcrc)
protected void	setupConnection()
private void	setupIOstreams(Call connectionRegistryCall)
private boolean	setupSaslConnection(InputStream in2, OutputStream out2, String serverPrincipal)
void	shutdown()	Just close connection.
private boolean	waitForWork()
private void	writeConnectionHeader()	Write the connection header.
private void	writeConnectionHeaderPreamble(OutputStream out)	Write the RPC header: <MAGIC WORD -- 'HBas'> <ONEBYTE_VERSION> <ONEBYTE_AUTH_TYPE>
private void	writeRequest(Call call)	Initiates a call by sending the parameter to the remote server.

Methods inherited from class org.apache.hadoop.hbase.ipc.RpcConnection

chooseServerPrincipal, createSecurityPreambleCall, getConnectionHeader, getConnectionHeaderPreamble, getLastTouched, getRemoteInetAddress, getServerPrincipals, isKerberosAuth, randomSelect, readResponse, remoteId, saslNegotiationDone, scheduleTimeoutTask, setLastTouched

Methods inherited from class java.lang.Object

clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

Field Details

LOG

private static final org.slf4j.Logger LOG

rpcClient

private final BlockingRpcClient rpcClient

threadName

private final String threadName

thread

private Thread thread

socket

protected Socket socket

in

private DataInputStream in

out

private DataOutputStream out

saslRpcClient

private HBaseSaslRpcClient saslRpcClient

calls

private final ConcurrentMap<Integer,Call> calls

callSender

private final BlockingRpcConnection.CallSender callSender

closed

private boolean closed

connectionHeaderPreamble

private byte[] connectionHeaderPreamble

connectionHeaderWithLength

private byte[] connectionHeaderWithLength

waitingConnectionHeaderResponse

private boolean waitingConnectionHeaderResponse

Constructor Details

BlockingRpcConnection

BlockingRpcConnection(BlockingRpcClient rpcClient,
 ConnectionId remoteId)
               throws IOException

Throws:

IOException

Method Details

setupConnection

protected void setupConnection()
                        throws IOException

Throws:

IOException

handleConnectionFailure

private void handleConnectionFailure(int curRetries,
 int maxRetries,
 IOException ioe)
                              throws IOException

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.

Parameters:

curRetries - current number of retries

maxRetries - max number of retries allowed

ioe - failure reason

Throws:

IOException - if max number of retries is reached

waitForWork

private boolean waitForWork()

run

public void run()

Specified by:

run in interface Runnable

disposeSasl

private void disposeSasl()

setupSaslConnection

private boolean setupSaslConnection(InputStream in2,
 OutputStream out2,
 String serverPrincipal)
                             throws IOException

Throws:

IOException

handleSaslConnectionFailure

private void handleSaslConnectionFailure(int currRetries,
 int maxRetries,
 Exception ex,
 org.apache.hadoop.security.UserGroupInformation user,
 String serverPrincipal)
                                  throws IOException,
InterruptedException

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.

The retry logic is governed by the SaslClientAuthenticationProvider.canRetry() method. Some providers have the ability to obtain new credentials and then re-attempt to authenticate with HBase services. Other providers will continue to fail if they failed the first time -- for those, we want to fail-fast.

Throws:

IOException

InterruptedException

getConnectionRegistry

private void getConnectionRegistry(InputStream inStream,
 OutputStream outStream,
 Call connectionRegistryCall)
                            throws IOException

Throws:

IOException

createStreams

private void createStreams(InputStream inStream,
 OutputStream outStream)

chooseServerPrincipal

private String chooseServerPrincipal(InputStream inStream,
 OutputStream outStream)
                              throws IOException

Throws:

IOException

setupIOstreams

private void setupIOstreams(Call connectionRegistryCall)
                     throws IOException

Throws:

IOException

writeConnectionHeaderPreamble

private void writeConnectionHeaderPreamble(OutputStream out)
                                    throws IOException

Write the RPC header: <MAGIC WORD -- 'HBas'> <ONEBYTE_VERSION> <ONEBYTE_AUTH_TYPE>

Throws:

IOException

writeConnectionHeader

private void writeConnectionHeader()
                            throws IOException

Write the connection header.

Throws:

IOException

processResponseForConnectionHeader

private void processResponseForConnectionHeader()
                                         throws IOException

Throws:

IOException

negotiateCryptoAes

private void negotiateCryptoAes(org.apache.hadoop.hbase.shaded.protobuf.generated.RPCProtos.CryptoCipherMeta cryptoCipherMeta)
                         throws IOException

Throws:

IOException

writeRequest

private void writeRequest(Call call)
                   throws IOException

Initiates a call by sending the parameter to the remote server. Note: this is not called from the Connection thread, but by other threads.

Throws:

IOException

See Also:

• readResponse()

readResponse

private void readResponse()

callTimeout

protected void callTimeout(Call call)

Specified by:

callTimeout in class RpcConnection

closeSocket

private void closeSocket()

closeConn

private void closeConn(IOException e)

shutdown

public void shutdown()

Description copied from class: RpcConnection

Just close connection. Do not need to remove from connection pool.

Specified by:

shutdown in class RpcConnection

cleanupConnection

public void cleanupConnection()

Description copied from class: RpcConnection

Does the clean up work after the connection is removed from the connection pool

Specified by:

cleanupConnection in class RpcConnection

sendRequest

public void sendRequest(Call call,
 HBaseRpcController pcrc)
                 throws IOException

Specified by:

sendRequest in class RpcConnection

Throws:

IOException

isActive

public boolean isActive()

Description copied from class: RpcConnection

Tell the idle connection sweeper whether we could be swept.

Specified by:

isActive in class RpcConnection