@Override
    public KafkaChannel buildChannel(String id, SelectionKey key, int maxReceiveSize,
                                     MemoryPool memoryPool, ChannelMetadataRegistry metadataRegistry) throws KafkaException {
        PlaintextTransportLayer transportLayer = null;
        try {
            transportLayer = buildTransportLayer(key);
            final PlaintextTransportLayer finalTransportLayer = transportLayer;
            Supplier<Authenticator> authenticatorCreator = () -> new PlaintextAuthenticator(configs, finalTransportLayer, listenerName);
            return buildChannel(id, transportLayer, authenticatorCreator, maxReceiveSize,
                    memoryPool != null ? memoryPool : MemoryPool.NONE, metadataRegistry);
        } catch (Exception e) {
            // Ideally these resources are closed by the KafkaChannel but this builder should close the resources instead
            // if an error occurs due to which KafkaChannel is not created.
            Utils.closeQuietly(transportLayer, "transport layer for channel Id: " + id);
            throw new KafkaException(e);
        }
    }
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