@ Override public void doBootstrap ( TransportClient client , Channel channel ) { SparkSaslClient saslClient = new SparkSaslClient ( appId , secretKeyHolder , conf . saslEncryption ( ) ) ; try { byte [ ] payload = saslClient . firstToken ( ) ; while ( ! saslClient . isComplete ( ) ) { SaslMessage msg = new SaslMessage ( appId , payload ) ; ByteBuf buf = Unpooled . buffer ( msg . encodedLength ( ) + ( int ) msg . body ( ) . size ( ) ) ; msg . encode ( buf ) ; buf . writeBytes ( msg . body ( ) . nioByteBuffer ( ) ) ; ByteBuffer response = client . sendRpcSync ( buf . nioBuffer ( ) , conf . authRTTimeoutMs ( ) ) ; payload = saslClient . response ( JavaUtils . bufferToArray ( response ) ) ; } client . setClientId ( appId ) ; if ( conf . saslEncryption ( ) ) { if ( ! SparkSaslServer . QOP_AUTH_CONF . equals ( saslClient . getNegotiatedProperty ( Sasl . QOP ) ) ) { throw new RuntimeException ( new SaslException ( "Encryption requests by negotiated non-encrypted connection." ) ) ; } SaslEncryption . addToChannel ( channel , saslClient , conf . maxSaslEncryptedBlockSize ( ) ) ; saslClient = null ; logger . debug ( "Channel {} configured for encryption." , client ) ; } } catch ( IOException ioe ) { throw new RuntimeException ( ioe ) ; } finally { if ( saslClient != null ) { try { // Once authentication is complete, the server will trust all remaining communication. saslClient . dispose ( ) ; } catch ( RuntimeException e ) { logger . error ( "Error while disposing SASL client" , e ) ; } } } }
Performs SASL authentication by sending a token and then proceeding with the SASL challenge - response tokens until we either successfully authenticate or throw an exception due to mismatch .