@Override
    @Test
    public void testMuteOnOOM() throws Exception {
        //clean up default selector, replace it with one that uses a finite mem pool
        selector.close();
        MemoryPool pool = new SimpleMemoryPool(900, 900, false, null);
        //the initial channel builder is for clients, we need a server one
        String tlsProtocol = "TLSv1.2";
        File trustStoreFile = TestUtils.tempFile("truststore", ".jks");
        Map<String, Object> sslServerConfigs = new TestSslUtils.SslConfigsBuilder(ConnectionMode.SERVER)
                .tlsProtocol(tlsProtocol)
                .createNewTrustStore(trustStoreFile)
                .build();
        channelBuilder = new SslChannelBuilder(ConnectionMode.SERVER, null, false, new LogContext());
        channelBuilder.configure(sslServerConfigs);
        selector = new Selector(NetworkReceive.UNLIMITED, 5000, metrics, time, "MetricGroup",
                new HashMap<>(), true, false, channelBuilder, pool, new LogContext());

        try (ServerSocketChannel ss = ServerSocketChannel.open()) {
            ss.bind(new InetSocketAddress(0));

            InetSocketAddress serverAddress = (InetSocketAddress) ss.getLocalAddress();

            SslSender sender1 = createSender(tlsProtocol, serverAddress, randomPayload(900));
            SslSender sender2 = createSender(tlsProtocol, serverAddress, randomPayload(900));
            sender1.start();
            sender2.start();

            SocketChannel channelX = ss.accept(); //not defined if its 1 or 2
            channelX.configureBlocking(false);
            SocketChannel channelY = ss.accept();
            channelY.configureBlocking(false);
            selector.register("clientX", channelX);
            selector.register("clientY", channelY);

            boolean handshaked = false;
            NetworkReceive firstReceive = null;
            long deadline = System.currentTimeMillis() + 5000;
            //keep calling poll until:
            //1. both senders have completed the handshakes (so server selector has tried reading both payloads)
            //2. a single payload is actually read out completely (the other is too big to fit)
            while (System.currentTimeMillis() < deadline) {
                selector.poll(10);

                Collection<NetworkReceive> completed = selector.completedReceives();
                if (firstReceive == null) {
                    if (!completed.isEmpty()) {
                        assertEquals(1, completed.size(), "expecting a single request");
                        firstReceive = completed.iterator().next();
                        assertTrue(selector.isMadeReadProgressLastPoll());
                        assertEquals(0, pool.availableMemory());
                    }
                } else {
                    assertTrue(completed.isEmpty(), "only expecting single request");
                }

                handshaked = sender1.waitForHandshake(1) && sender2.waitForHandshake(1);

                if (handshaked && firstReceive != null && selector.isOutOfMemory())
                    break;
            }
            assertTrue(handshaked, "could not initiate connections within timeout");

            selector.poll(10);
            assertTrue(selector.completedReceives().isEmpty());
            assertEquals(0, pool.availableMemory());
            assertNotNull(firstReceive, "First receive not complete");
            assertTrue(selector.isOutOfMemory(), "Selector not out of memory");

            firstReceive.close();
            assertEquals(900, pool.availableMemory()); //memory has been released back to pool

            Collection<NetworkReceive> completed = Collections.emptyList();
            deadline = System.currentTimeMillis() + 5000;
            while (System.currentTimeMillis() < deadline && completed.isEmpty()) {
                selector.poll(1000);
                completed = selector.completedReceives();
            }
            assertEquals(1, completed.size(), "could not read remaining request within timeout");
            assertEquals(0, pool.availableMemory());
            assertFalse(selector.isOutOfMemory());
        }
    }
A set of tests for the selector. These use a test harness that runs a simple socket server that echos back responses.
clients/src/test/java/org/apache/kafka/common/network/SslSelectorTest.java