@Test
    public void testSaslAuthenticationMaxReceiveSize() throws Exception {
        SecurityProtocol securityProtocol = SecurityProtocol.SASL_PLAINTEXT;
        configureMechanisms("PLAIN", Collections.singletonList("PLAIN"));

        // test auth with 1KB receive size
        saslServerConfigs.put(BrokerSecurityConfigs.SASL_SERVER_MAX_RECEIVE_SIZE_CONFIG, "1024");
        server = createEchoServer(securityProtocol);

        // test valid sasl authentication
        String node1 = "valid";
        checkAuthenticationAndReauthentication(securityProtocol, node1);

        // test with handshake request with large mechanism string
        byte[] bytes = new byte[1024];
        new Random().nextBytes(bytes);
        String mechanism = new String(bytes, StandardCharsets.UTF_8);
        String node2 = "invalid1";
        createClientConnection(SecurityProtocol.PLAINTEXT, node2);
        SaslHandshakeRequest handshakeRequest = buildSaslHandshakeRequest(mechanism, ApiKeys.SASL_HANDSHAKE.latestVersion());
        RequestHeader header = new RequestHeader(ApiKeys.SASL_HANDSHAKE, handshakeRequest.version(), "someclient", nextCorrelationId++);
        NetworkSend send = new NetworkSend(node2, handshakeRequest.toSend(header));
        selector.send(send);
        //we will get exception in server and connection gets closed.
        NetworkTestUtils.waitForChannelClose(selector, node2, ChannelState.READY.state());
        selector.close();

        String node3 = "invalid2";
        createClientConnection(SecurityProtocol.PLAINTEXT, node3);
        sendHandshakeRequestReceiveResponse(node3, ApiKeys.SASL_HANDSHAKE.latestVersion());

        // test with sasl authenticate request with large auth_byes string
        String authString = "\u0000" + TestJaasConfig.USERNAME + "\u0000" +  new String(bytes, StandardCharsets.UTF_8);
        ByteBuffer authBuf = ByteBuffer.wrap(Utils.utf8(authString));
        SaslAuthenticateRequestData data = new SaslAuthenticateRequestData().setAuthBytes(authBuf.array());
        SaslAuthenticateRequest request = new SaslAuthenticateRequest.Builder(data).build();
        header = new RequestHeader(ApiKeys.SASL_AUTHENTICATE, request.version(), "someclient", nextCorrelationId++);
        send = new NetworkSend(node3, request.toSend(header));
        selector.send(send);
        NetworkTestUtils.waitForChannelClose(selector, node3, ChannelState.READY.state());
        server.verifyAuthenticationMetrics(1, 2);
    }
Test SASL/PLAIN with sasl.authentication.max.receive.size config
clients/src/test/java/org/apache/kafka/common/security/authenticator/SaslAuthenticatorTest.java