@Test
    public void registerFailure() throws Exception {
        final String channelId = "1";

        final ChannelBuilder channelBuilder = mock(ChannelBuilder.class);

        when(channelBuilder.buildChannel(eq(channelId), any(SelectionKey.class), anyInt(), any(MemoryPool.class),
                any(ChannelMetadataRegistry.class))).thenThrow(new RuntimeException("Test exception"));

        try (MockedConstruction<Selector.SelectorChannelMetadataRegistry> mockedMetadataRegistry =
                     mockConstruction(Selector.SelectorChannelMetadataRegistry.class)) {
            Selector selector = new Selector(CONNECTION_MAX_IDLE_MS, new Metrics(), new MockTime(), "MetricGroup", channelBuilder, new LogContext());
            final SocketChannel socketChannel = SocketChannel.open();
            socketChannel.configureBlocking(false);
            IOException e = assertThrows(IOException.class, () -> selector.register(channelId, socketChannel));
            assertTrue(e.getCause().getMessage().contains("Test exception"), "Unexpected exception: " + e);
            assertFalse(socketChannel.isOpen(), "Socket not closed");
            // Ideally, metadataRegistry is closed by the KafkaChannel but if the KafkaChannel is not created due to
            // an error such as in a case like this, the Selector should be closing the metadataRegistry instead.
            verify(mockedMetadataRegistry.constructed().get(0)).close();
            selector.close();
        }
    }
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