code: @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);
          }
        }
      }
    }
docs: 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.