@Test
    public void testPartialSendAndReceiveReflectedInMetrics() throws Exception {
        // We use a large payload to attempt to trigger the partial send and receive logic.
        int payloadSize = 20 * BUFFER_SIZE;
        String payload = TestUtils.randomString(payloadSize);
        String nodeId = "0";
        blockingConnect(nodeId);
        ByteBufferSend send = ByteBufferSend.sizePrefixed(ByteBuffer.wrap(payload.getBytes()));
        NetworkSend networkSend = new NetworkSend(nodeId, send);

        selector.send(networkSend);
        KafkaChannel channel = selector.channel(nodeId);

        KafkaMetric outgoingByteTotal = findUntaggedMetricByName("outgoing-byte-total");
        KafkaMetric incomingByteTotal = findUntaggedMetricByName("incoming-byte-total");

        waitForCondition(() -> {
            long bytesSent = send.size() - send.remaining();
            assertEquals(bytesSent, ((Double) outgoingByteTotal.metricValue()).longValue());

            NetworkReceive currentReceive = channel.currentReceive();
            if (currentReceive != null) {
                assertEquals(currentReceive.bytesRead(), ((Double) incomingByteTotal.metricValue()).intValue());
            }

            selector.poll(50);
            return !selector.completedReceives().isEmpty();
        }, "Failed to receive expected response");

        KafkaMetric requestTotal = findUntaggedMetricByName("request-total");
        assertEquals(1, ((Double) requestTotal.metricValue()).intValue());

        KafkaMetric responseTotal = findUntaggedMetricByName("response-total");
        assertEquals(1, ((Double) responseTotal.metricValue()).intValue());
    }
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