@Test
    public void testNodeLatencyStats() throws Exception {
        try (Metrics m = new Metrics()) {
            // Create a new record accumulator with non-0 partitionAvailabilityTimeoutMs
            // otherwise it wouldn't update the stats.
            RecordAccumulator.PartitionerConfig config = new RecordAccumulator.PartitionerConfig(false, 42);
            long totalSize = 1024 * 1024;
            accumulator = new RecordAccumulator(logContext, batchSize, Compression.NONE, 0, 0L, 0L,
                DELIVERY_TIMEOUT_MS, config, m, "producer-metrics", time, apiVersions, null,
                new BufferPool(totalSize, batchSize, m, time, "producer-internal-metrics"));

            SenderMetricsRegistry senderMetrics = new SenderMetricsRegistry(m);
            Sender sender = new Sender(logContext, client, metadata, this.accumulator, false, MAX_REQUEST_SIZE, ACKS_ALL, 1,
                senderMetrics, time, REQUEST_TIMEOUT, 1000L, null, new ApiVersions());

            // Produce and send batch.
            long time1 = time.milliseconds();
            appendToAccumulator(tp0, 0L, "key", "value");
            sender.runOnce();
            assertEquals(1, client.inFlightRequestCount(), "We should have a single produce request in flight.");

            // We were able to send the batch out, so both the ready and drain values should be the same.
            RecordAccumulator.NodeLatencyStats stats = accumulator.getNodeLatencyStats(0);
            assertEquals(time1, stats.drainTimeMs);
            assertEquals(time1, stats.readyTimeMs);

            // Make the node 1 not ready.
            client.throttle(metadata.fetch().nodeById(0), 100);

            // Time passes, but we don't have anything to send.
            time.sleep(10);
            sender.runOnce();
            assertEquals(1, client.inFlightRequestCount(), "We should have a single produce request in flight.");

            // Stats shouldn't change as we didn't have anything ready.
            assertEquals(time1, stats.drainTimeMs);
            assertEquals(time1, stats.readyTimeMs);

            // Produce a new batch, but we won't be able to send it because node is not ready.
            long time2 = time.milliseconds();
            appendToAccumulator(tp0, 0L, "key", "value");
            sender.runOnce();
            assertEquals(1, client.inFlightRequestCount(), "We should have a single produce request in flight.");

            // The ready time should move forward, but drain time shouldn't change.
            assertEquals(time1, stats.drainTimeMs);
            assertEquals(time2, stats.readyTimeMs);

            // Time passes, we keep trying to send, but the node is not ready.
            time.sleep(10);
            time2 = time.milliseconds();
            sender.runOnce();
            assertEquals(1, client.inFlightRequestCount(), "We should have a single produce request in flight.");

            // The ready time should move forward, but drain time shouldn't change.
            assertEquals(time1, stats.drainTimeMs);
            assertEquals(time2, stats.readyTimeMs);

            // Finally, time passes beyond the throttle and the node is ready.
            time.sleep(100);
            time2 = time.milliseconds();
            sender.runOnce();
            assertEquals(2, client.inFlightRequestCount(), "We should have 2 produce requests in flight.");

            // Both times should move forward
            assertEquals(time2, stats.drainTimeMs);
            assertEquals(time2, stats.readyTimeMs);
        }
    }
Tests that topics are added to the metadata list when messages are available to send
and expired if not used during a metadata refresh interval.
clients/src/test/java/org/apache/kafka/clients/producer/internals/SenderTest.java