private void testClose(Args args, SecurityProtocol securityProtocol, ChannelBuilder clientChannelBuilder) throws Exception {
        String node = "0";
        server = createEchoServer(args, securityProtocol);
        clientChannelBuilder.configure(args.sslClientConfigs);
        this.selector = new Selector(5000, new Metrics(), TIME, "MetricGroup", clientChannelBuilder, new LogContext());
        InetSocketAddress addr = new InetSocketAddress("localhost", server.port());
        selector.connect(node, addr, BUFFER_SIZE, BUFFER_SIZE);

        NetworkTestUtils.waitForChannelReady(selector, node);
        // `waitForChannelReady` waits for client-side channel to be ready. This is sufficient for other tests
        // operating on the client-side channel. But here, we are muting the server-side channel below, so we
        // need to wait for the server-side channel to be ready as well.
        TestUtils.waitForCondition(() -> server.selector().channels().stream().allMatch(KafkaChannel::ready),
                "Channel not ready");

        final ByteArrayOutputStream bytesOut = new ByteArrayOutputStream();
        server.outputChannel(Channels.newChannel(bytesOut));
        server.selector().muteAll();
        byte[] message = TestUtils.randomString(100).getBytes();
        int count = 20;
        final int totalSendSize = count * (message.length + 4);
        for (int i = 0; i < count; i++) {
            selector.send(new NetworkSend(node, ByteBufferSend.sizePrefixed(ByteBuffer.wrap(message))));
            do {
                selector.poll(0L);
            } while (selector.completedSends().isEmpty());
        }
        server.selector().unmuteAll();
        selector.close(node);
        TestUtils.waitForCondition(() ->
            bytesOut.toByteArray().length == totalSendSize, 5000, "All requests sent were not processed");
    }
Tests that handshake failures are propagated only after writes complete, even when
there are delays in writes to ensure that clients see an authentication exception
rather than a connection failure.
clients/src/test/java/org/apache/kafka/common/network/SslTransportLayerTest.java