@Test
    public void testClientConnectionClose() throws Exception {
        String node = "0";
        SecurityProtocol securityProtocol = SecurityProtocol.SASL_SSL;
        TestJaasConfig jaasConfig = configureMechanisms("PLAIN", Collections.singletonList("PLAIN"));
        jaasConfig.setClientOptions("PLAIN", TestJaasConfig.USERNAME, "invalidpassword");

        server = createEchoServer(securityProtocol);
        createClientConnection(securityProtocol, node);

        Map<?, ?> delayedClosingChannels = NetworkTestUtils.delayedClosingChannels(server.selector());

        // Wait until server has established connection with client and has processed the auth failure
        TestUtils.waitForCondition(() -> {
            poll(selector);
            return !server.selector().channels().isEmpty();
        }, "Timeout waiting for connection");
        TestUtils.waitForCondition(() -> {
            poll(selector);
            return failedAuthenticationDelayMs == 0 || !delayedClosingChannels.isEmpty();
        }, "Timeout waiting for auth failure");

        selector.close();
        selector = null;

        // Now that client connection is closed, wait until server notices the disconnection and removes it from the
        // list of connected channels and from delayed response for auth failure
        TestUtils.waitForCondition(() -> failedAuthenticationDelayMs == 0 || delayedClosingChannels.isEmpty(),
                "Timeout waiting for delayed response remove");
        TestUtils.waitForCondition(() -> server.selector().channels().isEmpty(),
                "Timeout waiting for connection close");

        // Try forcing completion of delayed channel close
        TestUtils.waitForCondition(() -> time.milliseconds() > startTimeMs + failedAuthenticationDelayMs + 1,
                "Timeout when waiting for auth failure response timeout to elapse");
        NetworkTestUtils.completeDelayedChannelClose(server.selector(), time.nanoseconds());
    }
Tests client connection close before response for authentication failure is sent.
clients/src/test/java/org/apache/kafka/common/security/authenticator/SaslAuthenticatorFailureDelayTest.java