@ParameterizedTest
    @ArgumentsSource(SslTransportLayerArgumentsProvider.class)
    public void testNetworkThreadTimeRecorded(Args args) throws Exception {
        LogContext logContext = new LogContext();
        ChannelBuilder channelBuilder = new SslChannelBuilder(ConnectionMode.CLIENT, null, false, logContext);
        channelBuilder.configure(args.sslClientConfigs);
        try (Selector selector = new Selector(NetworkReceive.UNLIMITED, Selector.NO_IDLE_TIMEOUT_MS, new Metrics(), Time.SYSTEM,
                "MetricGroup", new HashMap<>(), false, true, channelBuilder, MemoryPool.NONE, logContext)) {

            String node = "0";
            server = createEchoServer(args, SecurityProtocol.SSL);
            InetSocketAddress addr = new InetSocketAddress("localhost", server.port());
            selector.connect(node, addr, BUFFER_SIZE, BUFFER_SIZE);

            String message = TestUtils.randomString(1024 * 1024);
            NetworkTestUtils.waitForChannelReady(selector, node);
            final KafkaChannel channel = selector.channel(node);
            assertTrue(channel.getAndResetNetworkThreadTimeNanos() > 0, "SSL handshake time not recorded");
            assertEquals(0, channel.getAndResetNetworkThreadTimeNanos(), "Time not reset");

            selector.mute(node);
            selector.send(new NetworkSend(node, ByteBufferSend.sizePrefixed(ByteBuffer.wrap(message.getBytes()))));
            while (selector.completedSends().isEmpty()) {
                selector.poll(100L);
            }
            long sendTimeNanos = channel.getAndResetNetworkThreadTimeNanos();
            assertTrue(sendTimeNanos > 0, "Send time not recorded: " + sendTimeNanos);
            assertEquals(0, channel.getAndResetNetworkThreadTimeNanos(), "Time not reset");
            assertFalse(channel.hasBytesBuffered(), "Unexpected bytes buffered");
            assertEquals(0, selector.completedReceives().size());

            selector.unmute(node);
            // Wait for echo server to send the message back
            TestUtils.waitForCondition(() -> {
                try {
                    selector.poll(100L);
                } catch (IOException e) {
                    return false;
                }
                return !selector.completedReceives().isEmpty();
            }, "Timed out waiting for a message to receive from echo server");

            long receiveTimeNanos = channel.getAndResetNetworkThreadTimeNanos();
            assertTrue(receiveTimeNanos > 0, "Receive time not recorded: " + receiveTimeNanos);
        }
    }
Tests that time spent on the network thread is accumulated on each channel
clients/src/test/java/org/apache/kafka/common/network/SslTransportLayerTest.java