@Test
    public void testCloseAllChannels() throws Exception {
        AtomicInteger closedChannelsCount = new AtomicInteger(0);
        ChannelBuilder channelBuilder = new PlaintextChannelBuilder(null) {
            private int channelIndex = 0;
            @Override
            KafkaChannel buildChannel(String id, TransportLayer transportLayer, Supplier<Authenticator> authenticatorCreator,
                                      int maxReceiveSize, MemoryPool memoryPool, ChannelMetadataRegistry metadataRegistry) {
                return new KafkaChannel(id, transportLayer, authenticatorCreator, maxReceiveSize, memoryPool, metadataRegistry) {
                    private final int index = channelIndex++;
                    @Override
                    public void close() throws IOException {
                        closedChannelsCount.getAndIncrement();
                        super.close();
                        if (index == 0) throw new RuntimeException("you should fail");
                    }
                };
            }
        };
        channelBuilder.configure(clientConfigs());
        Selector selector = new Selector(CONNECTION_MAX_IDLE_MS, new Metrics(), new MockTime(), "MetricGroup", channelBuilder, new LogContext());
        selector.connect("0", new InetSocketAddress("localhost", server.port), BUFFER_SIZE, BUFFER_SIZE);
        selector.connect("1", new InetSocketAddress("localhost", server.port), BUFFER_SIZE, BUFFER_SIZE);
        assertThrows(RuntimeException.class, selector::close);
        assertEquals(2, closedChannelsCount.get());
    }
Validate that we can send and receive a message larger than the receive and send buffer size
clients/src/test/java/org/apache/kafka/common/network/SelectorTest.java