@Test
    public void testNormalOperation() throws Exception {
        int conns = 5;
        int reqs = 500;

        // create connections
        InetSocketAddress addr = new InetSocketAddress("localhost", server.port);
        for (int i = 0; i < conns; i++)
            connect(Integer.toString(i), addr);
        // send echo requests and receive responses
        Map<String, Integer> requests = new HashMap<>();
        Map<String, Integer> responses = new HashMap<>();
        int responseCount = 0;
        for (int i = 0; i < conns; i++) {
            String node = Integer.toString(i);
            selector.send(createSend(node, node + "-0"));
        }

        // loop until we complete all requests
        while (responseCount < conns * reqs) {
            // do the i/o
            selector.poll(0L);

            assertEquals(0, selector.disconnected().size(), "No disconnects should have occurred.");

            // handle any responses we may have gotten
            for (NetworkReceive receive : selector.completedReceives()) {
                String[] pieces = asString(receive).split("-");
                assertEquals(2, pieces.length, "Should be in the form 'conn-counter'");
                assertEquals(receive.source(), pieces[0], "Check the source");
                assertEquals(0, receive.payload().position(), "Check that the receive has kindly been rewound");
                if (responses.containsKey(receive.source())) {
                    assertEquals((int) responses.get(receive.source()), Integer.parseInt(pieces[1]), "Check the request counter");
                    responses.put(receive.source(), responses.get(receive.source()) + 1);
                } else {
                    assertEquals(0, Integer.parseInt(pieces[1]), "Check the request counter");
                    responses.put(receive.source(), 1);
                }
                responseCount++;
            }

            // prepare new sends for the next round
            for (NetworkSend send : selector.completedSends()) {
                String dest = send.destinationId();
                if (requests.containsKey(dest))
                    requests.put(dest, requests.get(dest) + 1);
                else
                    requests.put(dest, 1);
                if (requests.get(dest) < reqs)
                    selector.send(createSend(dest, dest + "-" + requests.get(dest)));
            }
        }
        if (channelBuilder instanceof PlaintextChannelBuilder) {
            assertEquals(0, cipherMetrics(metrics).size());
        } else {
            waitForCondition(() -> cipherMetrics(metrics).size() == 1,
                "Waiting for cipher metrics to be created.");
            assertEquals(5, cipherMetrics(metrics).get(0).metricValue());
        }
    }
Send multiple requests to several connections in parallel. Validate that responses are received in the order that
requests were sent.
clients/src/test/java/org/apache/kafka/common/network/SelectorTest.java