@Test
    public void testChannelCloseWhileProcessingReceives() throws Exception {
        int numChannels = 4;
        Map<String, KafkaChannel> channels = TestUtils.fieldValue(selector, Selector.class, "channels");
        Set<SelectionKey> selectionKeys = new HashSet<>();
        for (int i = 0; i < numChannels; i++) {
            String id = String.valueOf(i);
            KafkaChannel channel = mock(KafkaChannel.class);
            channels.put(id, channel);
            when(channel.id()).thenReturn(id);
            when(channel.state()).thenReturn(ChannelState.READY);
            when(channel.isConnected()).thenReturn(true);
            when(channel.ready()).thenReturn(true);
            when(channel.read()).thenReturn(1L);

            SelectionKey selectionKey = mock(SelectionKey.class);
            when(channel.selectionKey()).thenReturn(selectionKey);
            when(selectionKey.isValid()).thenReturn(true);
            when(selectionKey.isReadable()).thenReturn(true);
            when(selectionKey.readyOps()).thenReturn(SelectionKey.OP_READ);
            when(selectionKey.attachment())
                    .thenReturn(channel)
                    .thenReturn(null);
            selectionKeys.add(selectionKey);

            NetworkReceive receive = mock(NetworkReceive.class);
            when(receive.source()).thenReturn(id);
            when(receive.size()).thenReturn(10);
            when(receive.bytesRead()).thenReturn(1);
            when(receive.payload()).thenReturn(ByteBuffer.allocate(10));
            when(channel.maybeCompleteReceive()).thenReturn(receive);
        }

        selector.pollSelectionKeys(selectionKeys, false, System.nanoTime());
        assertEquals(numChannels, selector.completedReceives().size());
        Set<KafkaChannel> closed = new HashSet<>();
        Set<KafkaChannel> notClosed = new HashSet<>();
        for (NetworkReceive receive : selector.completedReceives()) {
            KafkaChannel channel = selector.channel(receive.source());
            assertNotNull(channel);
            if (closed.size() < 2) {
                selector.close(channel.id());
                closed.add(channel);
            } else
                notClosed.add(channel);
        }
        assertEquals(notClosed, new HashSet<>(selector.channels()));
        closed.forEach(channel -> assertNull(selector.channel(channel.id())));

        selector.poll(0);
        assertEquals(0, selector.completedReceives().size());
    }
Ensure that no errors are thrown if channels are closed while processing multiple completed receives
clients/src/test/java/org/apache/kafka/common/network/SelectorTest.java