private void verifyNoUnnecessaryPollWithBytesBuffered(Consumer<SelectionKey> disableRead)
            throws Exception {
        this.selector.close();

        String node1 = "1";
        String node2 = "2";
        final AtomicInteger node1Polls = new AtomicInteger();

        this.channelBuilder = new TestSslChannelBuilder(ConnectionMode.CLIENT);
        this.channelBuilder.configure(sslClientConfigs);
        this.selector = new Selector(5000, metrics, time, "MetricGroup", channelBuilder, new LogContext()) {
            @Override
            void pollSelectionKeys(Set<SelectionKey> selectionKeys, boolean isImmediatelyConnected, long currentTimeNanos) {
                for (SelectionKey key : selectionKeys) {
                    KafkaChannel channel = (KafkaChannel) key.attachment();
                    if (channel != null && channel.id().equals(node1))
                        node1Polls.incrementAndGet();
                }
                super.pollSelectionKeys(selectionKeys, isImmediatelyConnected, currentTimeNanos);
            }
        };

        // Get node1 into bytes buffered state and then disable read on the socket.
        // Truncate the read buffers to ensure that there is buffered data, but not enough to make progress.
        int largeRequestSize = 100 * 1024;
        connect(node1, new InetSocketAddress("localhost", server.port));
        selector.send(createSend(node1,  TestUtils.randomString(largeRequestSize)));
        waitForBytesBuffered(selector, node1);
        TestSslChannelBuilder.TestSslTransportLayer.transportLayers.get(node1).truncateReadBuffer();
        disableRead.accept(selector.channel(node1).selectionKey());

        // Clear poll count and count the polls from now on
        node1Polls.set(0);

        // Process sends and receives on node2. Test verifies that we don't process node1
        // unnecessarily on each of these polls.
        connect(node2, new InetSocketAddress("localhost", server.port));
        int received = 0;
        String request = TestUtils.randomString(10);
        selector.send(createSend(node2, request));
        while (received < 100) {
            received += selector.completedReceives().size();
            if (!selector.completedSends().isEmpty()) {
                selector.send(createSend(node2, request));
            }
            selector.poll(5);
        }

        // Verify that pollSelectionKeys was invoked once to process buffered data
        // but not again since there isn't sufficient data to process.
        assertEquals(1, node1Polls.get());
        selector.close(node1);
        selector.close(node2);
        verifySelectorEmpty();
    }
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