@Override
        public KafkaPrincipal principal() {
            InetAddress clientAddress = transportLayer.socketChannel().socket().getInetAddress();
            // listenerName should only be null in Client mode where principal() should not be called
            if (listenerName == null)
                throw new IllegalStateException("Unexpected call to principal() when listenerName is null");
            return principalBuilder.build(new PlaintextAuthenticationContext(clientAddress, listenerName.value()));
        }
Constructs a plaintext channel builder. ListenerName is non-null whenever
it's instantiated in the broker and null otherwise.
clients/src/main/java/org/apache/kafka/common/network/PlaintextChannelBuilder.java