@Override
    public void handshake() throws IOException {
        if (state == State.NOT_INITIALIZED) {
            try {
                startHandshake();
            } catch (SSLException e) {
                maybeProcessHandshakeFailure(e, false, null);
            }
        }
        if (ready())
            throw renegotiationException();
        if (state == State.CLOSING)
            throw closingException();

        int read = 0;
        boolean readable = key.isReadable();
        try {
            // Read any available bytes before attempting any writes to ensure that handshake failures
            // reported by the peer are processed even if writes fail (since peer closes connection
            // if handshake fails)
            if (readable)
                read = readFromSocketChannel();

            doHandshake();
            if (ready())
                updateBytesBuffered(true);
        } catch (SSLException e) {
            maybeProcessHandshakeFailure(e, true, null);
        } catch (IOException e) {
            maybeThrowSslAuthenticationException();

            // This exception could be due to a write. If there is data available to unwrap in the buffer, or data available
            // in the socket channel to read and unwrap, process the data so that any SSL handshake exceptions are reported.
            try {
                do {
                    log.trace("Process any available bytes from peer, netReadBuffer {} netWriterBuffer {} handshakeStatus {} readable? {}",
                        netReadBuffer, netWriteBuffer, handshakeStatus, readable);
                    handshakeWrapAfterFailure(false);
                    handshakeUnwrap(false, true);
                } while (readable && readFromSocketChannel() > 0);
            } catch (SSLException e1) {
                maybeProcessHandshakeFailure(e1, false, e);
            }

            // If we get here, this is not a handshake failure, throw the original IOException
            throw e;
        }

        // Read from socket failed, so throw any pending handshake exception or EOF exception.
        if (read == -1) {
            maybeThrowSslAuthenticationException();
            throw new EOFException("EOF during handshake, handshake status is " + handshakeStatus);
        }
    }
Performs SSL handshake, non blocking.
Before application data (kafka protocols) can be sent client & kafka broker must
perform ssl handshake.
During the handshake SSLEngine generates encrypted data that will be transported over socketChannel.
Each SSLEngine operation generates SSLEngineResult , of which SSLEngineResult.handshakeStatus field is used to
determine what operation needs to occur to move handshake along.
A typical handshake might look like this.
+-------------+----------------------------------+-------------+
|  client     |  SSL/TLS message                 | HSStatus    |
+-------------+----------------------------------+-------------+
| wrap()      | ClientHello                      | NEED_UNWRAP |
| unwrap()    | ServerHello/Cert/ServerHelloDone | NEED_WRAP   |
| wrap()      | ClientKeyExchange                | NEED_WRAP   |
| wrap()      | ChangeCipherSpec                 | NEED_WRAP   |
| wrap()      | Finished                         | NEED_UNWRAP |
| unwrap()    | ChangeCipherSpec                 | NEED_UNWRAP |
| unwrap()    | Finished                         | FINISHED    |
+-------------+----------------------------------+-------------+

@throws IOException if read/write fails
@throws SslAuthenticationException if handshake fails with an {@link SSLException}
clients/src/main/java/org/apache/kafka/common/network/SslTransportLayer.java